From e483ba205fc8cf455d1745fbf5e33b379a6c06e0 Mon Sep 17 00:00:00 2001 From: "dr.dimitru" Date: Sun, 13 Sep 2026 00:18:01 +0300 Subject: [PATCH 1/3] =?UTF-8?q?=F0=9F=98=8E=20skills:=20version=20gate=20+?= =?UTF-8?q?=20lazy-loaded=20legacy-v5=20reference=20for=20Node=2014/16=20p?= =?UTF-8?q?ins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SKILL.md describes 6.x; a project pinned to josk@5.0.0 (the last Node 14/16 release) was being handed PostgresAdapter, pause()/resume(), concurrency, execute and lockLeaseTime, and the v6 'default' prefix, none of which apply there. A short version gate at the top of SKILL.md routes such projects to references/legacy-v5.md, read only behind that gate: the 3.x–5.x ladder, 5.0.0 exports/options/defaults, the ready() handler contract (sync zero-arg handlers are never rescheduled), the '' vs 'default' prefix change, and the Redis 4→5 key layout. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_017C1PrDpdMoybvFaAWvzktR --- AGENTS.md | 2 +- CLAUDE.md | 2 +- skills/josk/SKILL.md | 7 ++- skills/josk/references/legacy-v5.md | 81 +++++++++++++++++++++++++++++ 4 files changed, 89 insertions(+), 3 deletions(-) create mode 100644 skills/josk/references/legacy-v5.md diff --git a/AGENTS.md b/AGENTS.md index 87fe6f6..7844e95 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -13,7 +13,7 @@ Single cluster-wide claim per scheduled tick in horizontally scaled Node.js/Bun. - `*.d.ts`: Generated from JSDoc in `index.js` + adapters via `tsc --emitDeclarationOnly` on `prepublishOnly`. Do not edit manually. - `docs/adapter-api.md`: full adapter contract. - `docs/README.md`: docs index (additional guides + migration list). `docs/migration-v{4-v5,v5-v6,v6-v6.1,v6.1-v6.2}.md`: version upgrade guides. -- `skills/josk/`: Agent Skill source (open cross-tool standard) — `SKILL.md` + `references/{api,adapters,patterns,meteor,troubleshooting}.md`. Installed cross-tool via `npx skills add veliovgroup/josk` (Claude Code, Codex, Cursor, Copilot, Windsurf, Cline, Continue, Goose, Aider, +50 more). Excluded from npm tarball via `.npmignore`. Keep in sync with public API: when adding/changing options, methods, adapter constructors, execution semantics, or migration notes, update the matching reference file. `description` frontmatter in `SKILL.md` must stay ≤ 1024 chars. +- `skills/josk/`: Agent Skill source (open cross-tool standard) — `SKILL.md` + `references/{api,adapters,patterns,meteor,troubleshooting,legacy-v5}.md` (`legacy-v5.md` = frozen 5.0.0 surface for Node 14–16; never update it for new features). Installed cross-tool via `npx skills add veliovgroup/josk` (Claude Code, Codex, Cursor, Copilot, Windsurf, Cline, Continue, Goose, Aider, +50 more). Excluded from npm tarball via `.npmignore`. Keep in sync with public API: when adding/changing options, methods, adapter constructors, execution semantics, or migration notes, update the matching reference file. `description` frontmatter in `SKILL.md` must stay ≤ 1024 chars. - README.md, CHANGELOG.md, package.json (exports map, types, prepublishOnly now includes tsc). ## Code Style diff --git a/CLAUDE.md b/CLAUDE.md index a1eed42..cdb12a8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,7 @@ Primary agent guidance for this repo lives in [AGENTS.md](AGENTS.md) — read it The repo ships an [Agent Skill](https://inference.sh/blog/skills/agent-skills-overview) for **users of `josk`** (not for working on this repo itself). It uses the open, cross-tool `SKILL.md` standard and installs into 50+ AI coding agents via `npx skills add veliovgroup/josk`, including Claude Code, Codex, Cursor, Copilot, Windsurf, Cline, Continue, Roo Code, OpenCode, Goose, Aider, Gemini CLI, Kimi CLI, Tabnine, and more. When working on `josk` source, treat the skill as a downstream consumer of the public API. -- Source: [`skills/josk/`](skills/josk/) — `SKILL.md` plus `references/{api,adapters,patterns,meteor,troubleshooting}.md`. +- Source: [`skills/josk/`](skills/josk/) — `SKILL.md` plus `references/{api,adapters,patterns,meteor,troubleshooting,legacy-v5}.md` (`legacy-v5.md` is the frozen 5.0.0 surface for Node 14–16 users, reached only through the version gate at the top of `SKILL.md` — never update it for new features). - Cross-tool installer: [`npx skills`](https://github.com/vercel-labs/skills). Standard repo layout (`skills//SKILL.md`); no manifest required. - Excluded from the npm tarball via `.npmignore`. diff --git a/skills/josk/SKILL.md b/skills/josk/SKILL.md index b8ae85e..fc76a37 100644 --- a/skills/josk/SKILL.md +++ b/skills/josk/SKILL.md @@ -1,6 +1,6 @@ --- name: josk -description: "Guides JoSk integration for horizontally scaled Node.js and Bun apps — cluster-wide scheduling via Redis, MongoDB, or PostgreSQL so each tick runs on one instance. Use when writing or reviewing recurring jobs, cron-style tasks, `setInterval`/`setTimeout`/`setImmediate` work, periodic background jobs (queues, sync, polling, cleanup), multi-instance / Kubernetes / PM2 / Meteor deployments, the `josk` npm package, or `ostrio:cron-jobs`. Also when the user names JoSk, `RedisAdapter`, `MongoAdapter`, `PostgresAdapter`, Redis Cluster / KeyDB Cluster / Valkey `useHashTags`, KeyDB active-replication, MailTime RedisQueue pairing, at-least-once / at-most-once semantics, zombie recovery, leases, `zombieTime`, `lockLeaseTime`, `execute`, `concurrency`, `pause()`/`resume()` instance backpressure, or comparisons to Agenda, Bree, node-cron, Bull, or BullMQ." +description: "Guides JoSk integration for horizontally scaled Node.js and Bun apps — cluster-wide scheduling via Redis, MongoDB, or PostgreSQL so each tick runs on one instance. Use when writing or reviewing recurring jobs, cron-style tasks, `setInterval`/`setTimeout`/`setImmediate` work, periodic background jobs (queues, sync, polling, cleanup), multi-instance / Kubernetes / PM2 / Meteor deployments, the `josk` npm package, or `ostrio:cron-jobs`. Also when the user names JoSk, `RedisAdapter`, `MongoAdapter`, `PostgresAdapter`, Redis Cluster / KeyDB Cluster / Valkey `useHashTags`, KeyDB active-replication, MailTime RedisQueue pairing, at-least-once / at-most-once semantics, zombie recovery, leases, `zombieTime`, `lockLeaseTime`, `execute`, `concurrency`, `pause()`/`resume()` instance backpressure, or comparisons to Agenda, Bree, node-cron, Bull, or BullMQ. Also covers legacy josk 5.x pinned on Node 14/16." --- # JoSk @@ -8,6 +8,10 @@ description: "Guides JoSk integration for horizontally scaled Node.js and Bun ap Distributed `setInterval` / `setTimeout` / `setImmediate` for Node ≥20.9 and Bun ≥1.1. Server-only. Schedule in Redis, MongoDB, or PostgreSQL; lease + atomic claim limit duplicate ticks. +## Version gate — check first + +If `package.json` pins `josk` below `6.0.0`, or the runtime is Node < 20.9 (Node 14/16 hosts), stop here and read [references/legacy-v5.md](references/legacy-v5.md). Everything else in this skill describes 6.x and names exports, methods, and options that do not exist in 5.x (`PostgresAdapter`, `pause()`/`resume()`, `concurrency`, `execute`, `lockLeaseTime`, `useHashTags`, auto-`ready()` for sync handlers). + ## Quick start JoSk does not open connections — pass a connected client. Always wire `onError` and `destroy()` on shutdown. Read [references/](references/) lazily; do not guess v4/v5/v6 semantics from memory. @@ -37,6 +41,7 @@ await jobs.setInterval(async () => { /* idempotent work */ }, 60_000, 'poll-1m') | Handlers, CRON, concurrency, shutdown | [references/patterns.md](references/patterns.md) | | Meteor / `ostrio:cron-jobs` | [references/meteor.md](references/meteor.md) | | Zombies, jitter, migrations, KeyDB / Valkey | [references/troubleshooting.md](references/troubleshooting.md) | +| `josk@5` and older on Node 14/16 | [references/legacy-v5.md](references/legacy-v5.md) | | Email queue on JoSk (`mail-time`) | **REQUIRED** `mail-time` skill (`npx skills add veliovgroup/mail-time`) | ## Mental model diff --git a/skills/josk/references/legacy-v5.md b/skills/josk/references/legacy-v5.md new file mode 100644 index 0000000..6d66e85 --- /dev/null +++ b/skills/josk/references/legacy-v5.md @@ -0,0 +1,81 @@ +# JoSk 3.x–5.x (legacy, Node 14.20+ / 16) + +Read this file instead of `api.md` / `patterns.md` when `package.json` pins `josk` below `6.0.0` or the runtime is Node < 20.9. `5.0.0` is the last release that runs on Node 14/16 (`engines.node >=14.20.0`); `6.0.0` and later require Node ≥ 20.9 / Bun ≥ 1.1. `mail-time@3.x` depends on `josk@^5` — its scheduler side is described here. + +## Version ladder + +| josk | Node | Construction | Notes | +|---|---|---|---| +| 3.x / 4.x | ≥ 14.20 | `new JoSk({ db, prefix })` | Mongo built in, no adapter classes | +| **5.0.0** | ≥ 14.20 | `new JoSk({ adapter: new MongoAdapter({ db }) })` | `MongoAdapter`, `RedisAdapter` — **last Node 16 release** | +| 6.0.0+ | ≥ 20.9 | same | everything the rest of this skill describes | + +## What 5.x does NOT have + +Do not write any of these against 5.x — they are `undefined` and throw at runtime: + +- Exports: `PostgresAdapter`. Only `JoSk`, `MongoAdapter`, `RedisAdapter` exist. +- Methods: `pause()`, `resume()`. Only `setInterval`, `setTimeout`, `setImmediate`, `clearInterval`, `clearTimeout`, `destroy`, `ping`. +- Options: `concurrency`, `execute`, `lockLeaseTime`, `useHashTags`, `resetOnInit` on `JoSk` itself (it is an **adapter** option). +- Semantics: no lease ownership check on release (any instance can release the prefix lock), no auto-`ready()` for sync zero-arg handlers, no per-task concurrency cap — every due task runs in parallel on the lease winner. + +## Exports and constructor + +```js +const { JoSk, MongoAdapter, RedisAdapter } = require('josk'); // CJS +// import { JoSk, MongoAdapter, RedisAdapter } from 'josk'; // ESM +``` + +| `JoSk` option | Default | Notes | +|---|---|---| +| `adapter` | required | instance with `acquireLock, releaseLock, remove, add, update, iterate, ping` | +| `zombieTime` | `900000` (15 min) | a claimed task older than this is re-claimed; keep ≥ slowest handler + margin | +| `minRevolvingDelay` / `maxRevolvingDelay` | `128` / `768` ms | random tick jitter between storage scans | +| `onError(title, { description, error, uid })` | none | wire it — without it errors go to `debug` output only | +| `onExecuted(uid, { uid, date, delay, timestamp })` | none | | +| `autoClear` | `false` | remove tasks found in storage but not registered in this process | +| `debug` | `false` | | + +| Adapter option | `MongoAdapter` | `RedisAdapter` | +|---|---|---| +| store | `db` (connected `Db`, official driver) | `client` (connected `redis@4` client) | +| `prefix` | `''` → collection `__JobTasks__` (v6 changed the default to `'default'` → `__JobTasks__default`) | `''` | +| `lockCollectionName` | `'__JobTasks__.lock'` | same, as a key namespace | +| `resetOnInit` | `false` | `false` — when `true`, clears every stored task of this prefix at boot; never in a rolling deploy | + +Methods: `setInterval(fn, delayMs, uid)`, `setTimeout(fn, delayMs, uid)`, `setImmediate(fn, uid)` → `Promise`; `clearInterval(id)` / `clearTimeout(id)` accept the string or the pending Promise; `destroy()` is synchronous and stops the tick; `ping()` → `Promise<{ status }>`. + +## Handler contract in 5.x + +```js +jobs.setInterval(async () => { await work(); }, 60_000, 'poll-1m'); // Promise → ready() called for you +jobs.setInterval((ready) => { work(() => ready()); }, 60_000, 'poll-cb'); // callback style → you call ready() +jobs.setInterval(() => { work(); }, 60_000, 'poll-sync'); // WRONG in 5.x: never rescheduled +``` + +The handler always receives `ready`. If it returns a Promise, JoSk awaits it and calls `ready()` itself. If it returns anything else, JoSk returns without calling `ready()` — a sync handler that neither returns a Promise nor calls `ready` leaves the interval task claimed until `zombieTime` (15 min by default) rescues it. Calling `ready()` twice, or returning a Promise and also calling it, throws `Resolution method is overspecified`. `ready(date | timestamp)` reschedules an interval task to that moment instead of `now + delay`. + +## Rules specific to 5.x + +- **Throttle inside the handler.** There is no `concurrency`; if a task must not overlap with itself, keep the work idempotent and short, or guard with your own flag. +- **Stopping for a deploy** is `destroy()` on the instance and a normal process exit; other instances keep the prefix running. There is no `pause()`. +- **Do not default the prefix by omission across a v5 → v6 upgrade.** v5 `''` and v6 `'default'` are different collections; pass `prefix: ''` explicitly if the data must survive. +- **Redis 4→5 key layout.** 5.0.0 stores tasks in a sorted set `josk::schedule` plus hash `josk::tasks`; 4.x `josk::task:*` keys are only cleaned by a one-time boot with `resetOnInit: true` (stop every instance first). +- `onError` receives a **title string first**, then the details object — the opposite of most Node error callbacks. + +## Example — CJS Node 16, legacy empty prefix + +```js +const { JoSk, MongoAdapter } = require('josk'); + +const jobs = new JoSk({ + adapter: new MongoAdapter({ db, prefix: '' }), // keeps the __JobTasks__ collection + zombieTime: 120_000, + onError(title, { error, uid }) { log('[josk]', title, uid, error); }, +}); + +jobs.setInterval(async () => { await pollQueue(); }, 60_000, 'poll-1m'); +jobs.setTimeout(async () => { await sendFollowUp(); }, 5 * 60_000, 'follow-up-once'); + +process.on('SIGTERM', () => { jobs.destroy(); }); +``` From b38316b69a40cb017bee7028036464cb5639238b Mon Sep 17 00:00:00 2001 From: "dr.dimitru" Date: Thu, 24 Sep 2026 23:53:06 +0300 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=94=A7=20fix:=20keep=20an=20interval'?= =?UTF-8?q?s=20stored=20schedule=20when=20setInterval=20re-registers=20it?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit JoSk#setInterval() calls adapter.add() on every call, and every adapter's add() reset the task's next run to now + delay. Each process re-registers its intervals on boot, so every boot of any instance pushed every interval back by a full delay for the whole cluster. A crash-looping instance, frequent rolling deploys, or dev hot reloads starved intervals: a 10-minute Mongo-backed interval went 70 minutes without running after five reloads. add() now keeps the stored next run when the task already exists, is not deleted, is an interval, has the same delay, and the stored time is earlier than now + delay. A past-due task stays past-due and runs on the first revolution after boot; the atomic claim still lets only one instance run it. New tasks, deleted tasks, a changed delay, a changed kind, and every setTimeout/setImmediate registration still schedule at now + delay. A later stored time (e.g. a zombie park) is replaced by now + delay, so no run is ever scheduled later than before. Each adapter decides in one atomic statement: - MongoAdapter: upsert through an update pipeline with $cond (uid and delay wrapped in $literal). setTimeout keeps the plain $set upsert. Pipeline updates need mongod 4.2+; README already requires 4.4+. - RedisAdapter: ADD_TASK_SCRIPT decides in Lua and writes the same value to the task hash and the schedule ZSET. It still refuses to revive a tombstoned (isDeleted) entry, as before. - PostgresAdapter: INSERT ... ON CONFLICT DO UPDATE with a CASE on execute_at. The fix is on by default: the old reset is a bug for any multi-instance or often-restarted deployment. To restart the countdown on purpose, change delay or clear the task before setInterval(). Tests: Jest unit checks of the add() statement per adapter; live contract tests for all three adapters (keep earlier, past-due stays due and gets claimed, later time replaced, changed delay, deleted task, setTimeout and kind change, $-prefixed uid); mocha restart tests per adapter (first instance dies, second instance re-registers). Docs: README setInterval section, adapter contract, blank-example, skill references. index.cjs and .d.ts regenerated via prepublishOnly. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_014t7uCinRDEGFWJcVwtk3Uh --- README.md | 4 +- adapters/blank-example.d.ts | 4 +- adapters/blank-example.js | 4 +- adapters/mongo.js | 35 ++- adapters/postgres.js | 13 +- adapters/redis.js | 24 +- docs/adapter-api.md | 2 + index.cjs | 76 +++++- index.d.ts | 4 +- index.js | 4 +- skills/josk/references/adapters.md | 1 + skills/josk/references/api.md | 2 + skills/josk/references/troubleshooting.md | 4 + test/interval-restart-tests.js | 82 ++++++ test/jest/adapters.test.js | 305 ++++++++++++++++++++++ test/npm-mongo.js | 12 + test/npm-postgres.js | 12 + test/npm-redis.js | 9 + 18 files changed, 571 insertions(+), 26 deletions(-) create mode 100644 test/interval-restart-tests.js diff --git a/README.md b/README.md index 79caa78..6c49cfc 100644 --- a/README.md +++ b/README.md @@ -351,12 +351,14 @@ jobs.setInterval(task, 2 * 60000, 'task-2m'); // every two minutes ### `setInterval(func, delay, uid)` - `func` {*Function*} - Function to call on schedule -- `delay` {*Number*} - Delay for the first run and interval between further executions in milliseconds +- `delay` {*Number*} - Delay for the first run of a new task and interval between further executions in milliseconds - `uid` {*String*} - Unique app-wide task id - Returns: {*`Promise`*} *Set task into interval execution loop.* `ready()` *callback is passed as the first argument into a task function.* +__Re-registration keeps the schedule.__ Each process calls `setInterval()` again when it boots. If the task already exists in storage with the same `delay`, JoSk keeps its stored next run when that is earlier than `now + delay`. Restarts, crash loops, and rolling deploys of any instance no longer push the interval back by a full `delay`. A task that fell due while instances were down runs on the first scheduler revolution after boot, on one instance. A new task, or one registered with a different `delay`, is scheduled at `now + delay`. To restart the countdown on purpose, change `delay` or call `clearInterval(uid + 'setInterval')` before `setInterval()`. In `josk@6.3.0` and earlier, every `setInterval()` call reset the next run to `now + delay`. The [CRON helper](#cron) passes a new `delay` on each boot, so CRON tasks still move to the next CRON time. + In the example below, the next task __will not be scheduled__ until the current is ready: ```js diff --git a/adapters/blank-example.d.ts b/adapters/blank-example.d.ts index 9d04653..7dd97ec 100644 --- a/adapters/blank-example.d.ts +++ b/adapters/blank-example.d.ts @@ -66,7 +66,9 @@ export class BlankAdapter { /** * @async * @memberOf BlankAdapter - * Upsert task in storage + * Upsert task in storage. An existing, non-deleted interval re-registered with the + * same delay must keep its stored executeAt when that is earlier than now + delay, + * in one atomic operation (see docs/adapter-api.md); this example always resets * @name add * @param {string} uid - Unique ID of task * @param {boolean} isInterval - true/false defining loop or one-time task diff --git a/adapters/blank-example.js b/adapters/blank-example.js index 621fa4d..088ebe2 100644 --- a/adapters/blank-example.js +++ b/adapters/blank-example.js @@ -134,7 +134,9 @@ class BlankAdapter { /** * @async * @memberOf BlankAdapter - * Upsert task in storage + * Upsert task in storage. An existing, non-deleted interval re-registered with the + * same delay must keep its stored executeAt when that is earlier than now + delay, + * in one atomic operation (see docs/adapter-api.md); this example always resets * @name add * @param {string} uid - Unique ID of task * @param {boolean} isInterval - true/false defining loop or one-time task diff --git a/adapters/mongo.js b/adapters/mongo.js index a2ca55d..d3dd762 100644 --- a/adapters/mongo.js +++ b/adapters/mongo.js @@ -85,6 +85,36 @@ const ensureIndex = async (collection, keys, opts) => { } }; +/** + * Update pipeline for re-registering an interval. Keeps the stored `executeAt` + * when the task is an existing, non-deleted interval with the same `delay` and + * the stored time is earlier than `executeAt`; otherwise schedules `executeAt`. + * Field references inside one `$set` stage read the pre-update document. + * @param {string} uid + * @param {number} delay + * @param {Date} executeAt + * @returns {object[]} + */ +const intervalUpsertPipeline = (uid, delay, executeAt) => [{ + $set: { + executeAt: { + $cond: [{ + $and: [ + { $eq: ['$isInterval', true] }, + { $eq: ['$isDeleted', false] }, + { $eq: ['$delay', { $literal: delay }] }, + { $eq: [{ $type: '$executeAt' }, 'date'] }, + { $lt: ['$executeAt', { $literal: executeAt }] } + ] + }, '$executeAt', { $literal: executeAt }] + }, + uid: { $literal: uid }, + delay: { $literal: delay }, + isInterval: true, + isDeleted: false + } +}]; + /** Class representing MongoDB adapter for JoSk */ class MongoAdapter { /** @@ -286,13 +316,14 @@ class MongoAdapter { await this.ready(); try { + const executeAt = new Date(Date.now() + delay); await this.collection.updateOne({ uid - }, { + }, isInterval ? intervalUpsertPipeline(uid, delay, executeAt) : { $set: { uid, delay, - executeAt: new Date(Date.now() + delay), + executeAt, isInterval, isDeleted: false } diff --git a/adapters/postgres.js b/adapters/postgres.js index 228d74e..bed97ec 100644 --- a/adapters/postgres.js +++ b/adapters/postgres.js @@ -375,12 +375,23 @@ class PostgresAdapter { await this.ready(); try { + // Re-registering an existing interval with the same delay keeps its + // stored execute_at when that is earlier, so process restarts do not + // push the next run back by a full delay. SET expressions read the old row. const res = await this.client.query( `INSERT INTO josk_tasks (prefix, uid, delay, execute_at, is_interval, is_deleted, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, false, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (prefix, uid) DO UPDATE SET delay = EXCLUDED.delay, - execute_at = EXCLUDED.execute_at, + execute_at = CASE + WHEN EXCLUDED.is_interval = true + AND josk_tasks.is_interval = true + AND josk_tasks.is_deleted = false + AND josk_tasks.delay = EXCLUDED.delay + AND josk_tasks.execute_at < EXCLUDED.execute_at + THEN josk_tasks.execute_at + ELSE EXCLUDED.execute_at + END, is_interval = EXCLUDED.is_interval, is_deleted = false, updated_at = CURRENT_TIMESTAMP diff --git a/adapters/redis.js b/adapters/redis.js index 664d0a4..722de4f 100644 --- a/adapters/redis.js +++ b/adapters/redis.js @@ -45,13 +45,16 @@ const RELEASE_LOCK_SCRIPT = ` return 0 `; +// Re-registering an existing interval with the same delay keeps its stored +// executeAt when that is earlier than the new one, so process restarts do not +// push the next run back by a full delay. Hash and schedule ZSET get the same value. const ADD_TASK_SCRIPT = ` local payload = redis.call('HGET', KEYS[2], ARGV[1]) + local delay = tonumber(ARGV[2]) + local executeAt = tonumber(ARGV[3]) + local isInterval = ARGV[4] == '1' local task = payload and cjson.decode(payload) or { uid = ARGV[1], - delay = tonumber(ARGV[2]), - executeAt = tonumber(ARGV[3]), - isInterval = ARGV[4] == '1', isDeleted = false } @@ -59,13 +62,20 @@ const ADD_TASK_SCRIPT = ` return 0 end - task.delay = tonumber(ARGV[2]) - task.executeAt = tonumber(ARGV[3]) - task.isInterval = ARGV[4] == '1' + if payload and isInterval and task.isInterval == true and tonumber(task.delay) == delay then + local storedExecuteAt = tonumber(task.executeAt) + if storedExecuteAt and storedExecuteAt < executeAt then + executeAt = storedExecuteAt + end + end + + task.delay = delay + task.executeAt = executeAt + task.isInterval = isInterval task.isDeleted = false redis.call('HSET', KEYS[2], ARGV[1], cjson.encode(task)) - redis.call('ZADD', KEYS[1], tonumber(ARGV[3]), ARGV[1]) + redis.call('ZADD', KEYS[1], executeAt, ARGV[1]) return 1 `; diff --git a/docs/adapter-api.md b/docs/adapter-api.md index 85ee9d9..7dd627b 100644 --- a/docs/adapter-api.md +++ b/docs/adapter-api.md @@ -15,6 +15,7 @@ Start from [`blank-example.js`](https://github.com/veliovgroup/josk/blob/master/ - `ready()` optional but recommended. Use it to finish schema/index/init work before first storage op. - Prefer storage-server time over client time when comparing lease expirations. Mixed client clocks across a cluster will cause incorrect lock ownership otherwise. See `adapters/postgres.js` (`CURRENT_TIMESTAMP` in `acquireLock`) for a reference pattern. - Call `joskInstance.__execute(task)` fire-and-forget (do not `await`). JoSk handles internal concurrency and error wrapping. +- `add()` for an interval (`isInterval === true`) keeps the stored `executeAt` when the stored task exists, is not deleted, is an interval, has the same `delay`, and its `executeAt` is earlier than `now + delay`. In every other case (new task, deleted task, changed `delay`, one-shot task) store `now + delay`. Do it in one atomic storage operation and write the same value to every place the adapter stores the schedule (e.g. the Redis task hash and schedule ZSET). Every process re-registers its intervals on boot; resetting unconditionally lets frequent restarts of any instance postpone intervals cluster-wide. See the `add()` implementations in `adapters/mongo.js` (update pipeline), `adapters/redis.js` (Lua), and `adapters/postgres.js` (`ON CONFLICT ... CASE`). ## Adapter Class API @@ -40,6 +41,7 @@ Start from [`blank-example.js`](https://github.com/veliovgroup/josk/blob/master/ - `{string} uid` - `{boolean} isInterval` - `{number} delay` + - upsert; an unchanged interval keeps an earlier stored `executeAt` (see Design Rules) - async `Adapter#update(task, nextExecuteAt) - {Promise}` - `{object} task` - `{Date} nextExecuteAt` diff --git a/index.cjs b/index.cjs index d63526a..3ec7357 100644 --- a/index.cjs +++ b/index.cjs @@ -89,6 +89,36 @@ const ensureIndex = async (collection, keys, opts) => { } }; +/** + * Update pipeline for re-registering an interval. Keeps the stored `executeAt` + * when the task is an existing, non-deleted interval with the same `delay` and + * the stored time is earlier than `executeAt`; otherwise schedules `executeAt`. + * Field references inside one `$set` stage read the pre-update document. + * @param {string} uid + * @param {number} delay + * @param {Date} executeAt + * @returns {object[]} + */ +const intervalUpsertPipeline = (uid, delay, executeAt) => [{ + $set: { + executeAt: { + $cond: [{ + $and: [ + { $eq: ['$isInterval', true] }, + { $eq: ['$isDeleted', false] }, + { $eq: ['$delay', { $literal: delay }] }, + { $eq: [{ $type: '$executeAt' }, 'date'] }, + { $lt: ['$executeAt', { $literal: executeAt }] } + ] + }, '$executeAt', { $literal: executeAt }] + }, + uid: { $literal: uid }, + delay: { $literal: delay }, + isInterval: true, + isDeleted: false + } +}]; + /** Class representing MongoDB adapter for JoSk */ class MongoAdapter { /** @@ -290,13 +320,14 @@ class MongoAdapter { await this.ready(); try { + const executeAt = new Date(Date.now() + delay); await this.collection.updateOne({ uid - }, { + }, isInterval ? intervalUpsertPipeline(uid, delay, executeAt) : { $set: { uid, delay, - executeAt: new Date(Date.now() + delay), + executeAt, isInterval, isDeleted: false } @@ -554,13 +585,16 @@ const RELEASE_LOCK_SCRIPT = ` return 0 `; +// Re-registering an existing interval with the same delay keeps its stored +// executeAt when that is earlier than the new one, so process restarts do not +// push the next run back by a full delay. Hash and schedule ZSET get the same value. const ADD_TASK_SCRIPT = ` local payload = redis.call('HGET', KEYS[2], ARGV[1]) + local delay = tonumber(ARGV[2]) + local executeAt = tonumber(ARGV[3]) + local isInterval = ARGV[4] == '1' local task = payload and cjson.decode(payload) or { uid = ARGV[1], - delay = tonumber(ARGV[2]), - executeAt = tonumber(ARGV[3]), - isInterval = ARGV[4] == '1', isDeleted = false } @@ -568,13 +602,20 @@ const ADD_TASK_SCRIPT = ` return 0 end - task.delay = tonumber(ARGV[2]) - task.executeAt = tonumber(ARGV[3]) - task.isInterval = ARGV[4] == '1' + if payload and isInterval and task.isInterval == true and tonumber(task.delay) == delay then + local storedExecuteAt = tonumber(task.executeAt) + if storedExecuteAt and storedExecuteAt < executeAt then + executeAt = storedExecuteAt + end + end + + task.delay = delay + task.executeAt = executeAt + task.isInterval = isInterval task.isDeleted = false redis.call('HSET', KEYS[2], ARGV[1], cjson.encode(task)) - redis.call('ZADD', KEYS[1], tonumber(ARGV[3]), ARGV[1]) + redis.call('ZADD', KEYS[1], executeAt, ARGV[1]) return 1 `; @@ -1490,12 +1531,23 @@ class PostgresAdapter { await this.ready(); try { + // Re-registering an existing interval with the same delay keeps its + // stored execute_at when that is earlier, so process restarts do not + // push the next run back by a full delay. SET expressions read the old row. const res = await this.client.query( `INSERT INTO josk_tasks (prefix, uid, delay, execute_at, is_interval, is_deleted, created_at, updated_at) VALUES ($1, $2, $3, $4, $5, false, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (prefix, uid) DO UPDATE SET delay = EXCLUDED.delay, - execute_at = EXCLUDED.execute_at, + execute_at = CASE + WHEN EXCLUDED.is_interval = true + AND josk_tasks.is_interval = true + AND josk_tasks.is_deleted = false + AND josk_tasks.delay = EXCLUDED.delay + AND josk_tasks.execute_at < EXCLUDED.execute_at + THEN josk_tasks.execute_at + ELSE EXCLUDED.execute_at + END, is_interval = EXCLUDED.is_interval, is_deleted = false, updated_at = CURRENT_TIMESTAMP @@ -1921,7 +1973,9 @@ class JoSk { /** * @async * @memberOf JoSk - * Create recurring task (loop) + * Create recurring task (loop). Re-registering a stored task with the same + * `delay` (e.g. on process boot) keeps its next run when that is earlier than + * `now + delay`; otherwise the next run is `now + delay`. * @name setInterval * @param {JoSkTaskHandler} func - Function (task) to execute * @param {number} delay - Delay between task execution in milliseconds diff --git a/index.d.ts b/index.d.ts index a9d9370..4bafde5 100644 --- a/index.d.ts +++ b/index.d.ts @@ -99,7 +99,9 @@ export class JoSk { /** * @async * @memberOf JoSk - * Create recurring task (loop) + * Create recurring task (loop). Re-registering a stored task with the same + * `delay` (e.g. on process boot) keeps its next run when that is earlier than + * `now + delay`; otherwise the next run is `now + delay`. * @name setInterval * @param {JoSkTaskHandler} func - Function (task) to execute * @param {number} delay - Delay between task execution in milliseconds diff --git a/index.js b/index.js index 32d76bc..038f800 100644 --- a/index.js +++ b/index.js @@ -242,7 +242,9 @@ class JoSk { /** * @async * @memberOf JoSk - * Create recurring task (loop) + * Create recurring task (loop). Re-registering a stored task with the same + * `delay` (e.g. on process boot) keeps its next run when that is earlier than + * `now + delay`; otherwise the next run is `now + delay`. * @name setInterval * @param {JoSkTaskHandler} func - Function (task) to execute * @param {number} delay - Delay between task execution in milliseconds diff --git a/skills/josk/references/adapters.md b/skills/josk/references/adapters.md index 94bfd83..88eb68b 100644 --- a/skills/josk/references/adapters.md +++ b/skills/josk/references/adapters.md @@ -253,6 +253,7 @@ interface JoSkAdapter { - **Atomic due-task claim.** Do not `find all due → update later`. Use a single atomic operation (Lua, `FOR UPDATE SKIP LOCKED`, atomic `findOneAndUpdate`) to claim and return the task in one round-trip. - **`iterate(nextExecuteAt, lock, executeMode)`** is the entry point JoSk calls each tick. Claim one task (for `executeMode === 'one'`) or as many as the lease lets you (`executeMode === 'batch'`) and call `this.joskInstance.__execute(task)` **fire-and-forget** for each — JoSk handles internal concurrency and error wrapping. - **Storage-server time** for lease comparisons. Mixed client clocks across a cluster cause incorrect lock ownership. See `adapters/postgres.js` for the `CURRENT_TIMESTAMP` pattern. +- **`add()` keeps an unchanged interval's schedule.** When an existing, non-deleted interval re-registers with the same `delay`, keep its stored `executeAt` if it is earlier than `now + delay`; otherwise store `now + delay`. Do it in one atomic operation and update every copy of the schedule (the Redis adapter writes the task hash and the schedule ZSET). One-shot tasks always store `now + delay`. Unconditional resets let restarts of any instance postpone intervals cluster-wide. - **`ready()`** is optional but recommended for adapters that need to create schemas, indexes, or run migrations before the first storage op. ### Task object shape (what to pass to `__execute`) diff --git a/skills/josk/references/api.md b/skills/josk/references/api.md index 1a20d34..fc8c48f 100644 --- a/skills/josk/references/api.md +++ b/skills/josk/references/api.md @@ -81,6 +81,8 @@ Schedules a recurring task. The returned string is the internal timer id (the `u **Execution guarantee:** at-least-once per scheduled tick. The storage row for the task stays during execution; if the handler does not signal completion within `zombieTime`, the task is re-claimed and may run again. Make recurring handlers idempotent. +**Re-registration:** every boot calls `setInterval` again. An existing interval with the same `delay` keeps its stored next run when that is earlier than `now + delay`, so restarts and rolling deploys do not postpone it. A past-due task runs on the first revolution after boot, and one instance claims it. New tasks and a changed `delay` schedule at `now + delay`. `josk@6.3.0` and earlier reset to `now + delay` on every call. CRON helpers pass a fresh `delay` each boot, so they move to the next CRON time as before. + ### `setTimeout(handler, delay, uid)` → `Promise` Schedules a one-shot task after `delay` ms. Returns the internal timer id (`uid` + `setTimeout`). diff --git a/skills/josk/references/troubleshooting.md b/skills/josk/references/troubleshooting.md index c2e7242..cb238e6 100644 --- a/skills/josk/references/troubleshooting.md +++ b/skills/josk/references/troubleshooting.md @@ -90,6 +90,10 @@ Redis uses relative `PEXPIRE` TTL. Postgres computes lease expiry from `CURRENT_ If app-node clocks may diverge, prefer `PostgresAdapter`: `acquireLock` computes and compares lease expiry from server time. +## "Why does my interval run late or never after restarts?" + +`josk@6.3.0` and earlier reset an interval's next run to `now + delay` on every `setInterval()` call. Every boot of every instance re-registers its intervals, so a crash-looping instance, frequent rolling deploys, or Meteor dev hot reloads push each interval back by a full `delay` for the whole cluster. A 10-minute interval can go an hour without running. Later releases keep the stored next run when an existing interval re-registers with the same `delay` and the stored time is earlier. Upgrade. + ## "Why does my interval drift by ~1 second?" The effective interval is `delay + uniform(min, max) + round-trip`, not `delay` exactly. Defaults put `min=128`, `max=768`, so the upper bound is `delay + 768ms + storage latency`. To tighten: lower `maxRevolvingDelay`. To get exact wall-clock cadence (e.g. fire at the top of every minute), use the CRON pattern in `patterns.md` and call `ready(nextDate)`. diff --git a/test/interval-restart-tests.js b/test/interval-restart-tests.js new file mode 100644 index 0000000..63be47b --- /dev/null +++ b/test/interval-restart-tests.js @@ -0,0 +1,82 @@ +import { assert } from 'chai'; +import { destroyJobs, uniqueId, wait, waitUntil } from './helpers.js'; + +/** JoSk integration tests use ≥2048 ms task delay (storage + revolving jitter). */ +const TASK_DELAY = 2048; + +/** + * Simulates process restarts: instance A registers an interval and dies, + * instance B boots on the same prefix and re-registers the same uid + delay. + * @param {string} label + * @param {{ + * createJob: (prefix: string, resetOnInit: boolean) => import('../index.js').JoSk, + * cleanup: (prefix: string) => Promise + * }} hooks + */ +export const registerIntervalRestartTests = (label, hooks) => { + const { createJob, cleanup } = hooks; + + describe(`${label} - setInterval re-registration after restart`, function () { + this.slow(8000); + this.timeout(15000); + + const restart = async (prefix, uid, gap, runs) => { + const first = createJob(prefix, true); + const registeredAt = Date.now(); + await first.setInterval(() => {}, TASK_DELAY, uid); + first.destroy(); + + await wait(gap); + + const second = createJob(prefix, false); + const reRegisteredAt = Date.now(); + const timerId = await second.setInterval(() => { + runs.push(Date.now()); + }, TASK_DELAY, uid); + + return { second, timerId, registeredAt, reRegisteredAt }; + }; + + it('keeps the original schedule when an instance restarts before the task is due', async function () { + const prefix = uniqueId('interval-restart'); + const runs = []; + let ctx; + + try { + ctx = await restart(prefix, uniqueId('restart-not-due'), 1500, runs); + await waitUntil(() => runs.length >= 1, { + timeout: TASK_DELAY * 3, + message: 'interval did not run after restart' + }); + + const sinceFirstRegistration = runs[0] - ctx.registeredAt; + assert.isAtLeast(sinceFirstRegistration, TASK_DELAY, 'interval does not run before its stored time'); + assert.isBelow(sinceFirstRegistration, TASK_DELAY + 1000, 'restart does not push the interval back by another delay'); + await ctx.second.clearInterval(ctx.timerId); + } finally { + destroyJobs(ctx?.second); + await cleanup(prefix); + } + }); + + it('runs a past-due interval right after the restart', async function () { + const prefix = uniqueId('interval-restart-due'); + const runs = []; + let ctx; + + try { + ctx = await restart(prefix, uniqueId('restart-past-due'), TASK_DELAY + 1000, runs); + await waitUntil(() => runs.length >= 1, { + timeout: TASK_DELAY * 3, + message: 'past-due interval did not run after restart' + }); + + assert.isBelow(runs[0] - ctx.reRegisteredAt, 1000, 'past-due interval runs on the first revolutions after boot'); + await ctx.second.clearInterval(ctx.timerId); + } finally { + destroyJobs(ctx?.second); + await cleanup(prefix); + } + }); + }); +}; diff --git a/test/jest/adapters.test.js b/test/jest/adapters.test.js index 4d06414..534327e 100644 --- a/test/jest/adapters.test.js +++ b/test/jest/adapters.test.js @@ -58,6 +58,13 @@ const expectDueTaskClaiming = async (adapter, harness) => { expect(harness.__execute).not.toHaveBeenCalled(); }; +const INTERVAL_DELAY = 60000; + +const expectResetSchedule = (executeAt, before, delay = INTERVAL_DELAY) => { + expect(executeAt).toBeGreaterThanOrEqual(before + delay); + expect(executeAt).toBeLessThanOrEqual(Date.now() + delay); +}; + const createRedisClient = (opts = {}) => { const evalResults = [...(opts.evalResults || [])]; return { @@ -280,6 +287,27 @@ describe('RedisAdapter unit coverage', () => { expect(harness.__errorHandler).toHaveBeenCalledTimes(2); }); + it('add() passes uid, delay, now + delay, and the interval flag to the add script', async () => { + const { adapter, client } = await setupRedisAdapter({ + evalResults: [1, 1] + }); + const before = Date.now(); + + await expect(adapter.add('redis-add-interval', true, INTERVAL_DELAY)).resolves.toBe(true); + await expect(adapter.add('redis-add-timeout', false, INTERVAL_DELAY)).resolves.toBe(true); + + const [intervalSource, intervalCall] = client.eval.mock.calls[0]; + const [, timeoutCall] = client.eval.mock.calls[1]; + expect(intervalSource).toContain('storedExecuteAt < executeAt'); + expect(intervalSource).toContain("redis.call('ZADD', KEYS[1], executeAt, ARGV[1])"); + expect(intervalCall.keys).toEqual([adapter.scheduleKey, adapter.tasksKey]); + expect(intervalCall.arguments[0]).toBe('redis-add-interval'); + expect(intervalCall.arguments[1]).toBe(`${INTERVAL_DELAY}`); + expectResetSchedule(Number(intervalCall.arguments[2]), before); + expect(intervalCall.arguments[3]).toBe('1'); + expect(timeoutCall.arguments[3]).toBe('0'); + }); + it('rejects prefixes containing characters outside the allowed set', () => { const dummyClient = createRedisClient(); const make = (prefix) => () => new RedisAdapter({ client: dummyClient, prefix }); @@ -834,6 +862,56 @@ describe('MongoAdapter unit coverage', () => { expect(harness.__errorHandler.mock.calls.length).toBeGreaterThanOrEqual(5); }); + it('add(interval) upserts via a pipeline that keeps an earlier stored executeAt', async () => { + const { adapter, collection } = await setupMongoAdapter(); + const before = Date.now(); + + await expect(adapter.add('$mongo-add-interval', true, INTERVAL_DELAY)).resolves.toBe(true); + + const [filter, update, options] = collection.updateOne.mock.calls[0]; + expect(filter).toEqual({ uid: '$mongo-add-interval' }); + expect(options).toEqual({ upsert: true }); + expect(Array.isArray(update)).toBe(true); + expect(update).toHaveLength(1); + + const stage = update[0].$set; + expect(stage.uid).toEqual({ $literal: '$mongo-add-interval' }); + expect(stage.delay).toEqual({ $literal: INTERVAL_DELAY }); + expect(stage.isInterval).toBe(true); + expect(stage.isDeleted).toBe(false); + + const [condition, keep, fallback] = stage.executeAt.$cond; + const next = fallback.$literal; + expect(keep).toBe('$executeAt'); + expect(next).toBeInstanceOf(Date); + expectResetSchedule(+next, before); + expect(condition.$and).toEqual([ + { $eq: ['$isInterval', true] }, + { $eq: ['$isDeleted', false] }, + { $eq: ['$delay', { $literal: INTERVAL_DELAY }] }, + { $eq: [{ $type: '$executeAt' }, 'date'] }, + { $lt: ['$executeAt', { $literal: next }] } + ]); + }); + + it('add(timeout) keeps the plain $set upsert that always resets executeAt', async () => { + const { adapter, collection } = await setupMongoAdapter(); + const before = Date.now(); + + await expect(adapter.add('mongo-add-timeout', false, INTERVAL_DELAY)).resolves.toBe(true); + + const [filter, update, options] = collection.updateOne.mock.calls[0]; + expect(filter).toEqual({ uid: 'mongo-add-timeout' }); + expect(options).toEqual({ upsert: true }); + expect(update.$set).toMatchObject({ + uid: 'mongo-add-timeout', + delay: INTERVAL_DELAY, + isInterval: false, + isDeleted: false + }); + expectResetSchedule(+update.$set.executeAt, before); + }); + it('handles empty one-mode claims and legacy claim results', async () => { const task = { _id: 'legacy-claim', @@ -1166,6 +1244,31 @@ describe('PostgresAdapter unit coverage', () => { expect(harness.__errorHandler.mock.calls.length).toBeGreaterThanOrEqual(6); }); + it('add() upserts in one statement that keeps an earlier execute_at only for unchanged intervals', async () => { + const addQueries = []; + const { adapter } = await setupPostgresAdapter((sql, values) => { + if (sql.includes('INSERT INTO josk_tasks')) { + addQueries.push({ sql, values }); + return { rowCount: 1, rows: [{ uid: values[1] }] }; + } + return { rows: [], rowCount: 0 }; + }); + const before = Date.now(); + + await expect(adapter.add('postgres-add-interval', true, INTERVAL_DELAY)).resolves.toBe(true); + await expect(adapter.add('postgres-add-timeout', false, INTERVAL_DELAY)).resolves.toBe(true); + + expect(addQueries).toHaveLength(2); + const [intervalQuery, timeoutQuery] = addQueries; + const sql = intervalQuery.sql.replace(/\s+/g, ' '); + expect(sql).toContain('ON CONFLICT (prefix, uid) DO UPDATE SET'); + expect(sql).toContain('execute_at = CASE WHEN EXCLUDED.is_interval = true AND josk_tasks.is_interval = true AND josk_tasks.is_deleted = false AND josk_tasks.delay = EXCLUDED.delay AND josk_tasks.execute_at < EXCLUDED.execute_at THEN josk_tasks.execute_at ELSE EXCLUDED.execute_at END'); + expect(intervalQuery.values.slice(0, 3)).toEqual([adapter.prefix, 'postgres-add-interval', INTERVAL_DELAY]); + expectResetSchedule(intervalQuery.values[3], before); + expect(intervalQuery.values[4]).toBe(true); + expect(timeoutQuery.values[4]).toBe(false); + }); + it('handles empty one-mode iteration and full batch page continuation', async () => { const { adapter, harness } = await setupPostgresAdapter(); const tasks = Array.from({ @@ -1387,6 +1490,28 @@ const adapterSuites = [{ return { adapter, + // ADD_TASK_SCRIPT has always refused to revive a tombstoned (isDeleted) hash entry. + revivesDeletedTasks: false, + async readTask(uid) { + const payload = await client.hGet(adapter.tasksKey, uid); + if (!payload) { + return null; + } + const task = JSON.parse(String(payload)); + const score = await client.zScore(adapter.scheduleKey, uid); + return { + delay: Number(task.delay), + executeAt: Number(task.executeAt), + scheduledAt: score === null ? null : Number(score), + isInterval: task.isInterval === true, + isDeleted: task.isDeleted === true + }; + }, + async markDeleted(uid) { + const task = JSON.parse(String(await client.hGet(adapter.tasksKey, uid))); + task.isDeleted = true; + await client.hSet(adapter.tasksKey, uid, JSON.stringify(task)); + }, async cleanup() { await client.del([adapter.scheduleKey, adapter.tasksKey, adapter.lockKey]); await client.quit(); @@ -1411,6 +1536,23 @@ const adapterSuites = [{ return { adapter, + revivesDeletedTasks: true, + async readTask(uid) { + const task = await adapter.collection.findOne({ uid }); + if (!task) { + return null; + } + return { + delay: Number(task.delay), + executeAt: +task.executeAt, + scheduledAt: +task.executeAt, + isInterval: task.isInterval === true, + isDeleted: task.isDeleted === true + }; + }, + async markDeleted(uid) { + await adapter.collection.updateOne({ uid }, { $set: { isDeleted: true } }); + }, async cleanup() { await adapter.collection.drop().catch(() => {}); await adapter.lockCollection.deleteMany({ @@ -1436,6 +1578,27 @@ const adapterSuites = [{ return { adapter, + revivesDeletedTasks: true, + async readTask(uid) { + const res = await client.query( + 'SELECT delay, execute_at, is_interval, is_deleted FROM josk_tasks WHERE prefix = $1 AND uid = $2', + [adapter.prefix, uid] + ); + const task = res.rows[0]; + if (!task) { + return null; + } + return { + delay: Number(task.delay), + executeAt: Number(task.execute_at), + scheduledAt: Number(task.execute_at), + isInterval: task.is_interval === true, + isDeleted: task.is_deleted === true + }; + }, + async markDeleted(uid) { + await client.query('UPDATE josk_tasks SET is_deleted = true WHERE prefix = $1 AND uid = $2', [adapter.prefix, uid]); + }, async cleanup() { await client.query('DELETE FROM josk_tasks WHERE prefix = $1', [adapter.prefix]).catch(() => {}); await client.query('DELETE FROM josk_locks WHERE lock_key = $1', [adapter.lockKey]).catch(() => {}); @@ -1515,5 +1678,147 @@ for (const suite of adapterSuites) { expect(harness.__errorHandler).toHaveBeenCalledTimes(1); expect(harness.__errorHandler.mock.calls[0][1]).toContain('[update] [nextExecuteAt]'); }); + + const seedTask = async (adapter, uid, isInterval, storedExecuteAt) => { + await expect(adapter.add(uid, isInterval, INTERVAL_DELAY)).resolves.toBe(true); + await expect(adapter.update({ uid }, new Date(storedExecuteAt))).resolves.toBe(true); + await expect(context.readTask(uid)).resolves.toMatchObject({ + executeAt: storedExecuteAt, + scheduledAt: storedExecuteAt + }); + }; + + it('creates a new interval task at now + delay', async () => { + const adapter = await setupAdapter(); + const before = Date.now(); + + await expect(adapter.add('interval-new', true, INTERVAL_DELAY)).resolves.toBe(true); + + const task = await context.readTask('interval-new'); + expectResetSchedule(task.executeAt, before); + expect(task).toMatchObject({ + delay: INTERVAL_DELAY, + scheduledAt: task.executeAt, + isInterval: true, + isDeleted: false + }); + }); + + it('keeps an earlier stored executeAt when an interval re-registers with the same delay', async () => { + const adapter = await setupAdapter(); + const storedExecuteAt = Date.now() + 20000; + await seedTask(adapter, 'interval-keep', true, storedExecuteAt); + + await expect(adapter.add('interval-keep', true, INTERVAL_DELAY)).resolves.toBe(true); + + await expect(context.readTask('interval-keep')).resolves.toMatchObject({ + delay: INTERVAL_DELAY, + executeAt: storedExecuteAt, + scheduledAt: storedExecuteAt, + isInterval: true, + isDeleted: false + }); + }); + + it('keeps a past-due interval past-due on re-registration so the next tick claims it', async () => { + const adapter = await setupAdapter(); + const storedExecuteAt = Date.now() - 5000; + await seedTask(adapter, 'interval-past-due', true, storedExecuteAt); + + await expect(adapter.add('interval-past-due', true, INTERVAL_DELAY)).resolves.toBe(true); + await expect(context.readTask('interval-past-due')).resolves.toMatchObject({ + executeAt: storedExecuteAt, + scheduledAt: storedExecuteAt + }); + + const claimed = await adapter.iterate(new Date(Date.now() + 60000), createLock('past-due-owner'), 'batch'); + expect(claimed).toBe(1); + expect(harness.__execute).toHaveBeenCalledTimes(1); + expect(harness.__execute.mock.calls[0][0].uid).toBe('interval-past-due'); + }); + + it('replaces a later stored executeAt (e.g. a zombie park time) with now + delay', async () => { + const adapter = await setupAdapter(); + await seedTask(adapter, 'interval-later', true, Date.now() + (10 * INTERVAL_DELAY)); + const before = Date.now(); + + await expect(adapter.add('interval-later', true, INTERVAL_DELAY)).resolves.toBe(true); + + const task = await context.readTask('interval-later'); + expectResetSchedule(task.executeAt, before); + expect(task.scheduledAt).toBe(task.executeAt); + }); + + it('resets to now + delay when an interval re-registers with a different delay', async () => { + const adapter = await setupAdapter(); + const newDelay = INTERVAL_DELAY * 2; + await seedTask(adapter, 'interval-new-delay', true, Date.now() + 20000); + const before = Date.now(); + + await expect(adapter.add('interval-new-delay', true, newDelay)).resolves.toBe(true); + + const task = await context.readTask('interval-new-delay'); + expectResetSchedule(task.executeAt, before, newDelay); + expect(task).toMatchObject({ + delay: newDelay, + scheduledAt: task.executeAt, + isInterval: true + }); + }); + + it('never keeps the stored executeAt of a deleted task', async () => { + const adapter = await setupAdapter(); + const storedExecuteAt = Date.now() + 20000; + await seedTask(adapter, 'interval-deleted', true, storedExecuteAt); + await context.markDeleted('interval-deleted'); + const before = Date.now(); + + const added = await adapter.add('interval-deleted', true, INTERVAL_DELAY); + const task = await context.readTask('interval-deleted'); + + if (context.revivesDeletedTasks) { + expect(added).toBe(true); + expectResetSchedule(task.executeAt, before); + expect(task).toMatchObject({ + scheduledAt: task.executeAt, + isDeleted: false + }); + } else { + expect(added).toBe(false); + expect(task.isDeleted).toBe(true); + } + }); + + it('resets setTimeout tasks and interval-kind changes to now + delay', async () => { + const adapter = await setupAdapter(); + await seedTask(adapter, 'timeout-reset', false, Date.now() + 20000); + await seedTask(adapter, 'kind-change', true, Date.now() + 20000); + const before = Date.now(); + + await expect(adapter.add('timeout-reset', false, INTERVAL_DELAY)).resolves.toBe(true); + await expect(adapter.add('kind-change', false, INTERVAL_DELAY)).resolves.toBe(true); + + const timeoutTask = await context.readTask('timeout-reset'); + expectResetSchedule(timeoutTask.executeAt, before); + expect(timeoutTask).toMatchObject({ + scheduledAt: timeoutTask.executeAt, + isInterval: false + }); + + const kindTask = await context.readTask('kind-change'); + expectResetSchedule(kindTask.executeAt, before); + expect(kindTask.isInterval).toBe(false); + }); + + it('stores a $-prefixed interval uid verbatim across re-registration', async () => { + const adapter = await setupAdapter(); + const storedExecuteAt = Date.now() + 20000; + await seedTask(adapter, '$dollar-interval', true, storedExecuteAt); + + await expect(adapter.add('$dollar-interval', true, INTERVAL_DELAY)).resolves.toBe(true); + await expect(context.readTask('$dollar-interval')).resolves.toMatchObject({ + executeAt: storedExecuteAt + }); + }); }); } diff --git a/test/npm-mongo.js b/test/npm-mongo.js index b666a0a..26198f7 100644 --- a/test/npm-mongo.js +++ b/test/npm-mongo.js @@ -5,6 +5,7 @@ import { CronExpressionParser } from 'cron-parser'; import { it, describe, before, after } from 'mocha'; import { assert } from 'chai'; import { closeMongoClient, destroyJobs, uniqueId, wait, waitUntil } from './helpers.js'; +import { registerIntervalRestartTests } from './interval-restart-tests.js'; import { registerPauseResumeTests } from './pause-resume-tests.js'; if (!process.env.MONGO_URL) { @@ -1162,6 +1163,17 @@ describe('Mongo - JoSk', function () { }); }); + registerIntervalRestartTests('Mongo', { + createJob: (prefix, resetOnInit) => new JoSk({ + adapter: new MongoAdapter({ db, prefix, resetOnInit }), + ...racingJoSkOpts + }), + cleanup: async (prefix) => { + await db.collection(`__JobTasks__${prefix}`).drop().catch(() => {}); + await db.collection('__JobTasks__.lock').deleteMany({ uniqueName: `__JobTasks__${prefix}` }); + } + }); + registerPauseResumeTests('Mongo', { createJob: (prefix, resetOnInit) => new JoSk({ adapter: new MongoAdapter({ db, prefix, resetOnInit }), diff --git a/test/npm-postgres.js b/test/npm-postgres.js index 7357a69..aa3ac95 100644 --- a/test/npm-postgres.js +++ b/test/npm-postgres.js @@ -4,6 +4,7 @@ import { CronExpressionParser } from 'cron-parser'; import { it, describe, before, after } from 'mocha'; import { assert } from 'chai'; import { destroyJobs, uniqueId, wait, waitUntil } from './helpers.js'; +import { registerIntervalRestartTests } from './interval-restart-tests.js'; import { registerPauseResumeTests } from './pause-resume-tests.js'; if (!process.env.PG_URL) { @@ -565,6 +566,17 @@ describePostgres('PostgresAdapter + JoSk', function () { execute: 'one' }; + registerIntervalRestartTests('Postgres', { + createJob: (prefix, resetOnInit) => new JoSk({ + adapter: new PostgresAdapter({ client: pool, prefix, resetOnInit }), + ...racingJoSkOpts + }), + cleanup: async (prefix) => { + await pool.query('DELETE FROM josk_tasks WHERE prefix = $1', [prefix]); + await pool.query('DELETE FROM josk_locks WHERE lock_key = $1', [`josk-${prefix}.lock`]); + } + }); + registerPauseResumeTests('Postgres', { createJob: (prefix, resetOnInit) => new JoSk({ adapter: new PostgresAdapter({ client: pool, prefix, resetOnInit }), diff --git a/test/npm-redis.js b/test/npm-redis.js index 0dd4e61..bbe935f 100644 --- a/test/npm-redis.js +++ b/test/npm-redis.js @@ -5,6 +5,7 @@ import { CronExpressionParser } from 'cron-parser'; import { it, describe, before, after } from 'mocha'; import { assert } from 'chai'; import { destroyJobs, quitRedisClient, uniqueId, wait, waitUntil } from './helpers.js'; +import { registerIntervalRestartTests } from './interval-restart-tests.js'; import { registerPauseResumeTests } from './pause-resume-tests.js'; if (!process.env.REDIS_URL) { @@ -1122,6 +1123,14 @@ describe('Redis - JoSk', function () { }); }); + registerIntervalRestartTests('Redis', { + createJob: (prefix, resetOnInit) => new JoSk({ + adapter: new RedisAdapter({ client, prefix, resetOnInit }), + ...racingJoSkOpts + }), + cleanup: (prefix) => client.del([`josk:${prefix}:schedule`, `josk:${prefix}:tasks`, `josk:${prefix}:lock`]) + }); + registerPauseResumeTests('Redis', { createJob: (prefix, resetOnInit) => new JoSk({ adapter: new RedisAdapter({ client, prefix, resetOnInit }), From 28d1c078da2b84b5441cd7175cac764f81e8dfee Mon Sep 17 00:00:00 2001 From: "dr.dimitru" Date: Sat, 26 Sep 2026 13:14:22 +0300 Subject: [PATCH 3/3] release: prepare josk 6.4.0 --- README.md | 2 +- package-lock.json | 4 ++-- package.js | 2 +- package.json | 2 +- skills/josk/SKILL.md | 2 +- 5 files changed, 6 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 6c49cfc..9f3ceca 100644 --- a/README.md +++ b/README.md @@ -357,7 +357,7 @@ jobs.setInterval(task, 2 * 60000, 'task-2m'); // every two minutes *Set task into interval execution loop.* `ready()` *callback is passed as the first argument into a task function.* -__Re-registration keeps the schedule.__ Each process calls `setInterval()` again when it boots. If the task already exists in storage with the same `delay`, JoSk keeps its stored next run when that is earlier than `now + delay`. Restarts, crash loops, and rolling deploys of any instance no longer push the interval back by a full `delay`. A task that fell due while instances were down runs on the first scheduler revolution after boot, on one instance. A new task, or one registered with a different `delay`, is scheduled at `now + delay`. To restart the countdown on purpose, change `delay` or call `clearInterval(uid + 'setInterval')` before `setInterval()`. In `josk@6.3.0` and earlier, every `setInterval()` call reset the next run to `now + delay`. The [CRON helper](#cron) passes a new `delay` on each boot, so CRON tasks still move to the next CRON time. +When an active interval with a valid stored `executeAt` is registered again with the same `delay`, JoSk schedules the earlier of that timestamp and `now + delay`. A due or past-due interval becomes eligible on the first scheduler revolution after boot. New intervals and intervals registered with a different `delay` start at `now + delay`. `setTimeout()` and `setImmediate()` schedule from registration time. To restart an interval's countdown, clear it before registering it again or change its `delay`. The [CRON helper](#cron) computes a new delay on each boot and schedules the next CRON occurrence. In the example below, the next task __will not be scheduled__ until the current is ready: diff --git a/package-lock.json b/package-lock.json index 702e400..3e18eaa 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "josk", - "version": "6.3.0", + "version": "6.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "josk", - "version": "6.3.0", + "version": "6.4.0", "license": "BSD-3-Clause", "devDependencies": { "chai": "^6.2.2", diff --git a/package.js b/package.js index 8385b56..a1f3459 100644 --- a/package.js +++ b/package.js @@ -1,6 +1,6 @@ Package.describe({ name: 'ostrio:cron-jobs', - version: '6.3.0', + version: '6.4.0', summary: 'Tasks/CRON scheduler and manager for horizontally scaled multi-server apps', git: 'https://github.com/veliovgroup/josk', documentation: 'README.md' diff --git a/package.json b/package.json index accc0e6..616dcc8 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "josk", - "version": "6.3.0", + "version": "6.4.0", "description": "Tasks/CRON scheduler and manager for horizontally scaled multi-server applications. Supports Redis, MongoDB, PostgreSQL adapters.", "main": "index.js", "type": "module", diff --git a/skills/josk/SKILL.md b/skills/josk/SKILL.md index fc76a37..83f7023 100644 --- a/skills/josk/SKILL.md +++ b/skills/josk/SKILL.md @@ -1,6 +1,6 @@ --- name: josk -description: "Guides JoSk integration for horizontally scaled Node.js and Bun apps — cluster-wide scheduling via Redis, MongoDB, or PostgreSQL so each tick runs on one instance. Use when writing or reviewing recurring jobs, cron-style tasks, `setInterval`/`setTimeout`/`setImmediate` work, periodic background jobs (queues, sync, polling, cleanup), multi-instance / Kubernetes / PM2 / Meteor deployments, the `josk` npm package, or `ostrio:cron-jobs`. Also when the user names JoSk, `RedisAdapter`, `MongoAdapter`, `PostgresAdapter`, Redis Cluster / KeyDB Cluster / Valkey `useHashTags`, KeyDB active-replication, MailTime RedisQueue pairing, at-least-once / at-most-once semantics, zombie recovery, leases, `zombieTime`, `lockLeaseTime`, `execute`, `concurrency`, `pause()`/`resume()` instance backpressure, or comparisons to Agenda, Bree, node-cron, Bull, or BullMQ. Also covers legacy josk 5.x pinned on Node 14/16." +description: "Use when integrating, comparing, or debugging JoSk distributed scheduling in Node.js/Bun, including recurring or multi-instance jobs, Redis/KeyDB/MongoDB/PostgreSQL adapters, or legacy JoSk 5.x on Node 14/16." --- # JoSk