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
84 changes: 84 additions & 0 deletions .github/workflows/test-mongo-compatibility.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
name: Optional Mongo-compatible service tests

on:
workflow_dispatch:

jobs:
configured:
name: Check optional endpoint secrets
runs-on: ubuntu-latest
outputs:
cosmos: ${{ steps.check.outputs.cosmos }}
docdb: ${{ steps.check.outputs.docdb }}
steps:
- id: check
env:
COSMOS_MONGO_URL: ${{ secrets.COSMOS_MONGO_URL }}
DOCDB_URL: ${{ secrets.DOCDB_URL }}
run: |
if [ -n "$COSMOS_MONGO_URL" ]; then
echo 'cosmos=true' >> "$GITHUB_OUTPUT"
else
echo 'cosmos=false' >> "$GITHUB_OUTPUT"
echo '::notice::COSMOS_MONGO_URL is not configured; Cosmos test skipped.'
fi
if [ -n "$DOCDB_URL" ]; then
echo 'docdb=true' >> "$GITHUB_OUTPUT"
else
echo 'docdb=false' >> "$GITHUB_OUTPUT"
echo '::notice::DOCDB_URL is not configured; DocumentDB test skipped.'
fi

test-cosmos:
name: Azure Cosmos DB Mongo API / mongodb@7
if: needs.configured.outputs.cosmos == 'true'
needs: configured
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4

- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '22'
cache: npm

- name: Install dependencies
run: npm ci || npm install

- name: Pin MongoDB driver
run: npm install --no-save mongodb@7

- name: Mocha live tests (Cosmos DB Mongo API)
env:
MONGO_URL: ${{ secrets.COSMOS_MONGO_URL }}
run: npm run test:mongo

test-documentdb:
name: AWS DocumentDB / mongodb@7
if: needs.configured.outputs.docdb == 'true'
needs: configured
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4

- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '22'
cache: npm

- name: Install dependencies
run: npm ci || npm install

- name: Pin MongoDB driver
run: npm install --no-save mongodb@7

- name: Download AWS RDS CA bundle
run: curl -fsSLo "$RUNNER_TEMP/global-bundle.pem" https://truststore.pki.rds.amazonaws.com/global/global-bundle.pem

- name: Mocha live tests (DocumentDB)
env:
MONGO_URL: ${{ secrets.DOCDB_URL }}
NODE_EXTRA_CA_CERTS: ${{ runner.temp }}/global-bundle.pem
run: npm run test:mongo
62 changes: 57 additions & 5 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ env:
# pg@7 breaks on Node 14+ (readyState === 'open' before connect); engines require Node >= 20.9
LATEST_PG_DRIVER: '8'
# Latest server images — literals in services.*.image and job name (env not allowed there)
# redis:8-alpine mongo:8 postgres:18-alpine
# redis:8-alpine mongo:8 postgres:18-alpine valkey/valkey:8.1.9-alpine eqalpha/keydb:latest

jobs:
# Types, adapter guards, Jest core + coverage (once, all services)
Expand Down Expand Up @@ -324,6 +324,52 @@ jobs:
REDIS_URL: redis://127.0.0.1:6379
run: npm run test:redis

# Redis-compatible engines × redis@5 (Node 22), standalone only
test-servers-redis-compatible:
name: ${{ matrix.engine }} / Node 22 / redis@5
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
include:
- engine: Valkey 8.1.9
image: valkey/valkey:8.1.9-alpine
- engine: KeyDB latest
image: eqalpha/keydb:latest

services:
redis:
image: ${{ matrix.image }}
ports:
- 6379:6379
options: >-
--health-cmd "redis-cli ping || valkey-cli ping || keydb-cli ping"
--health-interval 5s
--health-timeout 3s
--health-retries 10

steps:
- uses: actions/checkout@v4

- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: '22'
cache: npm

- name: Install dependencies
run: npm ci || npm install

- name: Pin redis@${{ env.LATEST_REDIS_DRIVER }}
run: |
npm install --no-save redis@${{ env.LATEST_REDIS_DRIVER }}
node -e "console.log('redis driver:', require('redis/package.json').version)"

- name: Mocha live tests (Redis protocol)
env:
REDIS_URL: redis://127.0.0.1:6379
run: npm run test:redis

# MongoDB server LTS+ lines (latest driver, Node 22)
test-servers-mongo:
name: Servers mongo:${{ matrix.mongo-server }}
Expand Down Expand Up @@ -567,7 +613,7 @@ jobs:
PG_URL: postgres://postgres:postgres@127.0.0.1:5432/josk_test
run: bun test ./test/jest/adapters.test.js

# Meteor package tests — adapter suites only (METEOR_TEST_SUITE in package.js)
# Meteor package tests — adapter suites plus TypeScript runtime tests in one Redis cell
# Mongo adapter: no mongo service — Meteor test-packages starts its bundled mongod when MONGO_URL is unset
test-meteor-mongo:
name: Meteor ${{ matrix.meteor }} / mongo adapter
Expand Down Expand Up @@ -647,12 +693,18 @@ jobs:
curl -fsSL "https://install.meteor.com?release=${{ matrix.meteor }}" | sh
echo "$HOME/.meteor" >> $GITHUB_PATH

- name: Meteor package tests (Redis)
- name: Meteor default package suite (3.4 / Redis 8)
if: ${{ matrix.meteor == '3.4' && matrix.redis-server == '8' }}
env:
REDIS_URL: redis://127.0.0.1:6379
run: meteor test-packages ./ --driver-package=meteortesting:mocha --once

- name: Meteor package tests (Redis adapter)
if: ${{ matrix.meteor != '3.4' || matrix.redis-server != '8' }}
env:
METEOR_TEST_SUITE: redis
REDIS_URL: redis://127.0.0.1:6379
run: |
meteor test-packages ./ --driver-package=meteortesting:mocha --once
run: meteor test-packages ./ --driver-package=meteortesting:mocha --once

test-meteor-postgres:
name: Meteor ${{ matrix.meteor }} / pg:${{ matrix.postgres-server }}
Expand Down
9 changes: 8 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1 +1,8 @@
See full change-log at [releases on GitHub](https://github.com/veliovgroup/josk/releases)
## 6.4.0

- Suppress stale same-process interval updates, retain superseded handlers during shutdown, and clarify custom adapter fencing requirements.
- Report unfinished shutdown handlers, hand interval claims back, and abandon one-shots to preserve at-most-once execution.
- Add standalone KeyDB and Valkey CI smoke tests; clarify Mongo-compatible service coverage.
- Exercise TypeScript runtime tests through `meteor test-packages`; type-check all adapter contracts and `shutdown()` through the Meteor import with `tsc`.

See full change-log at [releases on GitHub](https://github.com/veliovgroup/josk/releases)
14 changes: 7 additions & 7 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ __Note: JoSk is the server-only package.__
## Prerequisites

- `node@>=20.9.0`
- `redis-server@>=5.0.0`, KeyDB, or Valkey for `RedisAdapter`, with the `redis@^4` or `redis@^5` NPM package
- `redis-server@>=5.0.0` or a single-writer KeyDB/Valkey server for `RedisAdapter`, with `redis@^4` or `redis@^5`. CI targets selected standalone images; see [Redis Adapter](#redis-adapter) for exact coverage.
- `mongod@>=4.4` for `MongoAdapter`, with the official `mongodb` NPM package (only the official driver is tested)
- `postgres@>=12` for `PostgresAdapter`, with `pg@>=8.0.3` (`pg@7` does not connect on Node 14+)
- `bun@>=1.1.0` (optional), see [Bun runtime](#bun-runtime)
Expand Down Expand Up @@ -175,10 +175,10 @@ JoSk has no dependencies. Install the driver for your adapter: `redis`, `mongodb

#### Redis Adapter

`RedisAdapter` keeps due timestamps in a sorted set and task payloads in a hash, and claims due tasks with Lua scripts. It works with Redis, [KeyDB](https://docs.keydb.dev/), and Valkey.
`RedisAdapter` keeps due timestamps in a sorted set and task payloads in a hash, and claims due tasks with Lua scripts. It uses the Redis-compatible commands shared by Redis, [KeyDB](https://docs.keydb.dev/), and Valkey; CI targets Redis 6/7/8, `eqalpha/keydb:latest`, and `valkey/valkey:8.1.9-alpine` in standalone mode using `redis@5`.

- Use one writable primary. Do not route JoSk reads or writes to replicas; claims must be visible to all instances at once.
- For Redis Cluster or KeyDB Cluster, pass `useHashTags: true`.
- For Redis Cluster, pass `useHashTags: true`; CI tests a 3-master Redis Cluster. KeyDB/Valkey Cluster modes are not separately tested.
- Avoid KeyDB active-replication (multi-master). Its conflict resolution can let two writers claim the same task.
- For strict single-claim scheduling across data centers, use a strongly consistent store, or PostgreSQL with one write authority.

Expand All @@ -194,7 +194,7 @@ const jobs = new JoSk({
adapter: new RedisAdapter({
client: redisClient,
prefix: 'app-scheduler',
// useHashTags: true, // Enable for Redis Cluster / KeyDB Cluster
// useHashTags: true, // Redis Cluster; KeyDB/Valkey Cluster not CI-tested
}),
onError(reason, details) {
// Catches exceptions thrown inside scheduled tasks
Expand Down Expand Up @@ -418,7 +418,7 @@ Stops this instance's scheduler. After `destroy()`, only `clearTimeout()` and `c
- `opts.timeout` {*Number*} - [Optional] Milliseconds to wait for running handlers to call `ready()`. Default: `10000`
- Returns: {*`Promise<boolean>`*} `true` if every running handler finished within `timeout`

Calls `destroy()`, waits for running handlers, then hands unfinished interval claims back to storage so another instance runs them on its next poll instead of after `zombieTime`. A handler that calls `ready()` after its claim was handed back does not change the schedule. Keep `timeout` below your platform's termination grace period.
Calls `destroy()`, waits for running handlers, then hands unfinished interval claims back to storage so another instance runs them on its next poll instead of after `zombieTime`. A handler that calls `ready()` after its claim was handed back does not change the schedule. Repeated calls share the first shutdown attempt and timeout; when it expires, JoSk reports every unfinished handler to `onError` (or `console.error`), hands current interval claims back, and abandons one-shot tasks to preserve at-most-once behavior. Superseded handlers also count as unfinished, but their obsolete claims are left unchanged. Keep at-most-once handlers idempotent, or set `timeout` longer than the longest one-shot handler; keep it below your platform's termination grace period.

```js
const shutdown = async () => {
Expand Down Expand Up @@ -481,7 +481,7 @@ Queue-polling pattern: after winning a tick, claim rows from your own queue, cal
| `setTimeout`, `setImmediate` | **At-most-once** across the cluster | Task is removed from storage *before* the handler runs. If the process dies mid-run, the run is lost. |
| `setInterval` | **At-least-once** per tick (until cleared) | Task stays in storage while running. If `ready()` is not called within `zombieTime`, another instance may claim and run it again. Make handlers idempotent. |

`zombieTime` (default 15 minutes) is the safety net for stuck handlers. Set it above your slowest legitimate handler plus storage latency. Restarting an app does not shorten a running interval's hold. So after an unclean kill, recovery waits the full `zombieTime`, not `delay`, but a rolling deploy cannot start a second copy of a handler that is still running. Call [`shutdown()`](#shutdownopts) on `SIGTERM` to hand running claims back instead of leaving them to `zombieTime`. A handler that finishes after its claim was recovered cannot overwrite the newer run's schedule. See [monitoring and recovery](docs/monitoring.md).
`zombieTime` (default 15 minutes) is the safety net for stuck handlers. Set it above your slowest legitimate handler plus storage latency. Restarting an app does not shorten a running interval's hold. So after an unclean kill, recovery waits the full `zombieTime`, not `delay`, but a rolling deploy cannot start a second copy of a handler that is still running. Call [`shutdown()`](#shutdownopts) on `SIGTERM` to hand running claims back instead of leaving them to `zombieTime`. JoSk skips a late `ready()` write when a newer same-uid run starts in the same process. Built-in adapters also fence writes across instances; custom adapters need equivalent fencing for that guarantee. See [adapter-api.md](docs/adapter-api.md) and [monitoring and recovery](docs/monitoring.md).

- `execute: 'batch'` (default) claims all due tasks under one lease, for throughput. `execute: 'one'` claims one task per lease, for smaller bursts and fairer spread across instances.
- `concurrency` caps parallel handlers in this instance. Default `Infinity`, like Node's timers. Set a limit when handlers share DB connections or API rate limits.
Expand Down Expand Up @@ -589,7 +589,7 @@ DELETE FROM josk_locks WHERE lock_key = 'josk-cluster-scheduler.lock';

### MongoDB connection fine tuning

Replica-set tuning, dedicated-DB advice, the index list, and Mongoose / CosmosDB / DocumentDB notes: [`docs/mongodb.md`](docs/mongodb.md).
Replica-set tuning, dedicated-DB advice, the index list, and Mongo-compatible service notes: [`docs/mongodb.md`](docs/mongodb.md). Cosmos DB and DocumentDB aren't part of default CI; optional endpoint tests require configured secrets.

## Prefix mapping

Expand Down
5 changes: 3 additions & 2 deletions adapters/blank-example.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,8 @@ export class BlankAdapter {
/**
* @async
* @memberOf BlankAdapter
* Update next execution timestamp
* Update next execution timestamp. Apply the lease filter, schedule write, and
* claimLeaseId clear atomically; without a lease, preserve unconditional updates.
* @name update
* @param {object} task - Full object of task from storage
* @param {Date} nextExecuteAt - Date defining time of next execution
Expand All @@ -89,7 +90,7 @@ export class BlankAdapter {
/**
* @async
* @memberOf BlankAdapter
* Claim due tasks atomically and execute them
* Claim due tasks atomically, persist lock.leaseId as claimLeaseId, and execute them
* @name iterate
* @param {Date} nextExecuteAt - Date defining time of next execution for zombie recovery
* @param {{ ownerId: string, leaseId: string }} lock
Expand Down
21 changes: 15 additions & 6 deletions adapters/blank-example.js
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,8 @@ class BlankAdapter {
/**
* @async
* @memberOf BlankAdapter
* Update next execution timestamp
* Update next execution timestamp. Apply the lease filter, schedule write, and
* claimLeaseId clear atomically; without a lease, preserve unconditional updates.
* @name update
* @param {object} task - Full object of task from storage
* @param {Date} nextExecuteAt - Date defining time of next execution
Expand All @@ -178,10 +179,18 @@ class BlankAdapter {
await this.ready();

try {
return await this.requiredOption.update({
const filter = {
scope: this.uniqueName,
uid: task.uid,
executeAt: +nextExecuteAt
uid: task.uid
};
if (typeof task.claimLeaseId === 'string' && task.claimLeaseId !== '') {
filter.claimLeaseId = task.claimLeaseId;
}

return await this.requiredOption.update({
filter,
set: { executeAt: +nextExecuteAt },
unset: ['claimLeaseId']
});
} catch (opError) {
this.joskInstance.__errorHandler(opError, '[StorageAdapter] [update] [opError]', 'Exception inside StorageAdapter#update() method', task.uid);
Expand All @@ -192,7 +201,7 @@ class BlankAdapter {
/**
* @async
* @memberOf BlankAdapter
* Claim due tasks atomically and execute them
* Claim due tasks atomically, persist lock.leaseId as claimLeaseId, and execute them
* @name iterate
* @param {Date} nextExecuteAt - Date defining time of next execution for zombie recovery
* @param {{ ownerId: string, leaseId: string }} lock
Expand All @@ -219,7 +228,7 @@ class BlankAdapter {
}

executed++;
this.joskInstance.__execute(task);
this.joskInstance.__execute({ ...task, claimLeaseId: lock.leaseId });
}

return executed;
Expand Down
19 changes: 18 additions & 1 deletion docs/adapter-api.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ Start from [`blank-example.js`](https://github.com/veliovgroup/josk/blob/master/
- Keep second-layer scheduler lock. Use owner-bound lease token. Never release foreign lease.
- Derive lock lifetime from the lock object itself — prefer the relative `lock.leaseMs`; use `lock.expireAt` / `lock.expiresAtMs` only as a fallback for locks minted without it. Never substitute `zombieTime` (a substituted long TTL freezes the whole prefix for up to `zombieTime` when a holder dies uncleanly), and never re-derive a duration as `expiresAtMs - Date.now()` when `leaseMs` is present — that second app-clock read is distorted by any clock step between mint and acquire.
- Claim due tasks atomically in storage. Do not `find all due -> update later`.
- Fence `update()` on the claim: return `claimLeaseId: lock.leaseId` on claimed tasks, and when `update()` receives a task with `claimLeaseId`, write only if the stored lease still matches, then clear it. Otherwise a handler that finishes after zombie recovery overwrites the newer run's schedule. A task without `claimLeaseId` updates unconditionally. JoSk also calls `update(task, now)` to hand a claim back on `destroy()` / `shutdown()`.
- Fence `update()` on the claim: atomically store and return `claimLeaseId: lock.leaseId`; `update()` must match that lease, update the schedule, and clear the lease in one write. Without a token, update unconditionally for compatibility. JoSk also suppresses a superseded run's late `ready()` write when a newer same-uid run starts in this process; cross-instance safety still needs adapter fencing. JoSk calls `update(task, now)` to hand a claim back on `destroy()` / `shutdown()`.
- `iterate()` should claim and execute `one` or `batch` depending on `executeMode`.
- `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.
Expand Down Expand Up @@ -79,4 +79,21 @@ Inside `Adapter#iterate()` call `this.joskInstance.__execute(task)` with:
4. Call `this.joskInstance.__execute(task)`.
5. Release scheduler lease only if owner token still matches.

For example, atomically persist and return the claim token:

```js
return { ...task, claimLeaseId: lock.leaseId };
```

A Mongo-style atomic `update()` filters on the token when present, updates the schedule, and clears it. Without a token, it filters by task identity only:

```js
const filter = { uid: task.uid };
if (task.claimLeaseId) filter.claimLeaseId = task.claimLeaseId;
await collection.updateOne(filter, {
$set: { executeAt: nextExecuteAt },
$unset: { claimLeaseId: '' }
});
```

Global lock alone is not enough for duplicate prevention. Atomic task claim is required.
Loading
Loading