Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/<name>/SKILL.md`); no manifest required.
- Excluded from the npm tarball via `.npmignore`.

Expand Down
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<string>`*}

*Set task into interval execution loop.* `ready()` *callback is passed as the first argument into a task function.*

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:

```js
Expand Down
4 changes: 3 additions & 1 deletion adapters/blank-example.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion adapters/blank-example.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
35 changes: 33 additions & 2 deletions adapters/mongo.js
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
/**
Expand Down Expand Up @@ -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
}
Expand Down
13 changes: 12 additions & 1 deletion adapters/postgres.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
24 changes: 17 additions & 7 deletions adapters/redis.js
Original file line number Diff line number Diff line change
Expand Up @@ -45,27 +45,37 @@ 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
}

if payload and task.isDeleted then
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
`;

Expand Down
2 changes: 2 additions & 0 deletions docs/adapter-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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<boolean>}`
- `{object} task`
- `{Date} nextExecuteAt`
Expand Down
76 changes: 65 additions & 11 deletions index.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
/**
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -554,27 +585,37 @@ 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
}

if payload and task.isDeleted then
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
`;

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion index.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion index.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.js
Original file line number Diff line number Diff line change
@@ -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'
Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
Loading
Loading