diff --git a/agents/bunqueue-assistant.md b/agents/bunqueue-assistant.md index 5b41b2bf..749578f6 100644 --- a/agents/bunqueue-assistant.md +++ b/agents/bunqueue-assistant.md @@ -68,11 +68,11 @@ Validated on the client with the broker's validator; an invalid option throws. | Option | Meaning | Default | |---|---|---| | `priority` | Integer, higher runs first. | `0` | -| `delay` | Ms until ready (max 365 days; negative = now). | `0` | +| `delay` | Run-time offset in ms, bounded to ±4,320,000,000,000,000; negative keeps a past run time (ready immediately; among equal-priority FIFO jobs, earlier run times go first). Priority and LIFO ordering still apply. | `0` | | `attempts` | Max attempts including the first. | `3` | | `backoff` | Ms (exponential with jitter, cap 1 h) or `{type:'fixed'\|'exponential', delay, maxDelay?}`. | `1000`, exponential | | `timeout` | Processing timeout in ms (the processor `signal` aborts). | none | -| `jobId` | Custom id; while that job is live, `add` returns it instead of creating one. Not with `repeat`. | UUIDv7 | +| `jobId` | Custom id; returns the live generation while ownership is retained. Memory/SQLite use a bounded 50,000-ID default cache; PostgreSQL checks the database. See [custom-ID limits](https://bunqueue.dev/guide/queue/deduplication/#custom-id-limits). Not with `repeat`. | UUIDv7 | | `removeOnComplete` / `removeOnFail` | Drop the job on completion / final failure. Only booleans are honoured. | `false` | | `durable` | Write straight to SQLite (no write buffer); in TCP it skips the auto-batcher. | `false` | | `stallTimeout` | Stall interval for this job only. | queue `stallInterval` (30 s) | @@ -82,7 +82,7 @@ Validated on the client with the broker's validator; an invalid option throws. | `stackTraceLimit` | Stack lines kept on failure. | `10` | | `failParentOnFailure` / `removeDependencyOnFailure` / `continueParentOnFailure` / `ignoreDependencyOnFailure` | What a failed child does to its parent. | `false` | | `timestamp` | Overrides creation time. | `Date.now()` | -| `deduplication` `{id, ttl?, extend?, replace?}` | A duplicate add with the same id returns the existing job; `ttl` bounds the key, `extend` restarts it, `replace` swaps a pending owner. | none | +| `deduplication` `{id, ttl?, extend?, replace?}` | While the unique key is owned, a duplicate returns the existing job; `ttl` bounds the key, `extend` restarts it, `replace` swaps a pending owner. Separate from custom-ID ownership. | none | | `group` `{id, maxSize?, priority?}` | Puts the job in a group; `maxSize` caps pending group jobs. | none | | `keepLogs`, `sizeLimit`, `debounce` | Stored and shown on the job; not enforced by `Queue.add` (`debounce` is applied by `Bunqueue`). | none | @@ -179,7 +179,7 @@ DLQ config defaults: `autoRetry` false, `autoRetryInterval` 1 h, `maxAutoRetries ### Queue: schedulers (repeatable jobs) -`RepeatOpts`: `pattern | every`, `limit`, `immediately`, `count`, `prevMillis`, `offset`, `jobId`, `timezone` (embedded default UTC, not sent over TCP), `skipMissedOnRestart` (true), `skipIfNoWorker` (false), `preventOverlap` (true). +`RepeatOpts`: `pattern | every`, `limit`, `immediately`, `count`, `prevMillis`, `offset`, `jobId`, `timezone` (forwarded in both modes; default UTC), `skipMissedOnRestart` (true), `skipIfNoWorker` (false), `preventOverlap` (true). | API | What it does | Embedded | TCP | |---|---|---|---| diff --git a/docs/agents/client-reference.md b/docs/agents/client-reference.md index 44ddd454..ddcd3ab3 100644 --- a/docs/agents/client-reference.md +++ b/docs/agents/client-reference.md @@ -65,6 +65,7 @@ const flow = new Workflow('order') .step('notify-warehouse', async () => ({ notified: true })) .step('notify-email', async () => ({ sent: true })) ) + .step('classify', async () => ({ tier: 'basic' })) .branch((ctx) => (ctx.steps['classify'] as { tier: string }).tier) .path('vip', (w) => w.step('vip-handler', async () => ({ discount: 20 }))) .path('basic', (w) => w.step('basic-handler', async () => ({ discount: 0 }))) @@ -192,10 +193,10 @@ const queue3 = new Queue('jobs', { autoBatch: { enabled: false } }); interface JobOptions { priority?: number; // Higher = sooner delay?: number; // ms before processing - attempts?: number; // Max retries (default: 3) + attempts?: number; // Maximum attempts, including the first (default: 3) backoff?: number; // Retry backoff (default: 1000ms) timeout?: number; // Processing timeout - jobId?: string; // Custom ID (idempotent) + jobId?: string; // Custom ID; retained-ownership dedup (memory/SQLite default cache: 50,000 IDs; PostgreSQL checks DB) removeOnComplete?: boolean; removeOnFail?: boolean; durable?: boolean; // Bypass write buffer diff --git a/docs/agents/project-reference.md b/docs/agents/project-reference.md index 2965a27c..838470ea 100644 --- a/docs/agents/project-reference.md +++ b/docs/agents/project-reference.md @@ -75,8 +75,8 @@ High-performance job queue server for Bun. SQLite persistence, cron jobs, priori 1. **PUSH**: Client → TcpPool → TcpServer → QueueManager → Shard[hash(queue)] → PriorityQueue → WriteBuffer → SQLite 2. **PULL**: Client → TcpServer → QueueManager → Shard → PriorityQueue.pop() → Job (state: active) -3. **ACK**: Client → TcpServer → AckBatcher → Shard.complete() → jobResults (LRU) + completedJobs (Set) -4. **FAIL**: Client → TcpServer → Shard.fail() → retry (backoff) OR → DLQ (max attempts) +3. **ACK**: Worker → optional client-side AckBatcher → TCP handler (`handlers/core.ts`) → QueueManager.ack → `operations/ack/completion.ts` → shard/index/storage, jobResults (LRU) + completedJobs (Set) +4. **FAIL**: Client → TCP handler (`handlers/core.ts`) → QueueManager.fail → `operations/ack/failure.ts` → shard/index/storage → retry (backoff) OR → DLQ (max attempts) ## Directory Structure @@ -148,7 +148,7 @@ Cleanup runs every 10s (it also purges expired `removedResults`). Evicts 10% whe TCP_PORT=6789 HTTP_PORT=6790 HOST=0.0.0.0 BUNQUEUE_DATA_PATH=./data/bunq.db AUTH_TOKENS=token1,token2 CORS_ALLOW_ORIGIN=* -METRICS_AUTH=false # Require auth for /metrics endpoint +METRICS_AUTH=false # Require auth for /prometheus scrape endpoint TCP_SOCKET_PATH= # RESERVED, not wired: TCP always binds HOST:TCP_PORT # (client ConnectionOptions.socketPath is not read either) HTTP_SOCKET_PATH= # Unix socket for HTTP (works) @@ -257,9 +257,14 @@ bun run bench # Native macOS benchmarks only | Mode | Throughput | Data Loss Risk | | ------------------ | --------------- | -------------- | -| Buffered (default) | ~100k jobs/sec | Up to 10ms | -| Durable | ~10k jobs/sec | None | -| Auto-batch (TCP) | ~145k ops/s (concurrent), ~10k ops/s (sequential) | None (same as PUSH/PUSHB) | +| Buffered (default) | ~100k jobs/sec | Pending application buffer; typical flush interval 10 ms, longer during stalls/retries | +| Durable | ~10k jobs/sec | Bypasses the application admission buffer; host/storage power-loss limitations remain | +| Auto-batch (TCP) | ~145k ops/s (concurrent), ~10k ops/s (sequential) | Inherits the selected PUSH/PUSHB admission durability | + +These throughput figures are indicative historical measurements, not results of +this documentation review. SQLite uses WAL with `synchronous=NORMAL`; durable +admission does not promise unconditional power-loss safety. See +[Persistence](../features/persistence.md). ## Background Tasks diff --git a/docs/architecture.md b/docs/architecture.md index bbed6d61..8f406e5f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -462,8 +462,11 @@ remote server via `embeddedQueue.forward({ to, queue })` ([`src/client/forwarder.ts`](../src/client/forwarder.ts)). Remote failures fall back to local retry/DLQ. With persistent local storage, queued work survives uplink outages while its process or volume survives; a deterministic remote jobId -`fwd::` dedupes re-forwards within the server's custom-id -retention window. Trade-off: at-least-once delivery with bounded dedup; ideal for +`fwd::` suppresses re-forwards while remote live ownership is +retained. Memory/SQLite use a bounded cache (50,000 IDs by default); PostgreSQL +checks database identity. Cache eviction can end local-engine suppression before +completion, so retain application idempotency for external effects. Trade-off: +at-least-once delivery with bounded dedup; ideal for IoT/edge that must tolerate intermittent connectivity. See [Store-and-Forward](./features/store-and-forward.md). diff --git a/docs/data-model.md b/docs/data-model.md index 384082fb..ff2193ee 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -70,13 +70,14 @@ The core record. Defined in `src/domain/types/jobs/model.ts:41-90`. export interface Job { readonly id: JobId; // UUIDv7, branded string (types/jobs/model.ts:1) readonly queue: string; + readonly name: string; // processor routing name, separate from user data readonly data: unknown; // user payload (msgpack BLOB on disk) readonly priority: number; // ungrouped: higher first; grouped: 0 first, then ascending readonly createdAt: number; // epoch ms readonly lifo: boolean; // tie-break among equal priority // Scheduling - runAt: number; // createdAt + delay (a negative delay counts as 0); <=now means ready + runAt: number; // createdAt + delay; negative delay keeps a past run time; <=now means ready startedAt: number | null; // set when pulled (active) completedAt: number | null; // set when completed @@ -143,13 +144,19 @@ export interface Job { ``` `customId` is a broker-wide identity key, not a per-queue key. When supplied it -also becomes `jobs.id`, whose SQLite primary key is global. Re-adding a live -custom ID from any queue returns the existing generation; after a terminal +also becomes `jobs.id`, whose SQLite primary key is global. Ordinary +memory/SQLite adds return a live generation only while its ownership is retained +in the bounded custom-ID LRU (50,000 IDs by default). SQLite admission uses +upserts, so the primary key does not preserve the original live generation +after eviction; durable admission does not remove that limit. PostgreSQL checks +locked database identity independently. See +[Deduplication & Unique Jobs](./features/deduplication-and-unique.md). After a terminal generation, reuse retires the completed/DLQ record and its result in the same SQLite transaction that inserts the new generation (also for a non-durable add, which then bypasses the `WriteBuffer`), and only afterwards retires the -in-memory record and result/log ownership. At no point may two rows or two live -jobs share the same custom ID, and a crash never leaves zero generations. +in-memory record and result/log ownership. This transactional terminal-reuse +path is distinct from the unresolved live-ownership eviction limitation; +ordinary adds must not be used as an unconditional uniqueness guarantee. Persisted job IDs must be well-formed Unicode. Standalone custom IDs and every planned atomic-flow ID are rejected if they contain an isolated UTF-16 surrogate, because such values do not round-trip through Bun SQLite TEXT and @@ -172,7 +179,8 @@ Notable supporting types: - `JobId = string & { readonly __brand: 'JobId' }` — UUIDv7, generated by `Bun.randomUUIDv7()` (`src/domain/job/ids.ts:7-8`). - `RepeatConfig` (`src/domain/types/jobs/model.ts:13-25`) — repeatable-job config (`every`, `pattern`, - `limit`, `count`, `tz`, …). In-memory only on a `Job`; not a SQLite column. + `limit`, `count`, `tz`, …). No dedicated legacy SQLite column; persisted in + `jobs.extended_options`. PostgreSQL stores it in the complete job payload. - `BackoffConfig` (`src/domain/types/jobs/model.ts:27-31`) — `{ type: 'fixed' | 'exponential'; delay; maxDelay? }`. It is also the object form of `JobInput.backoff`, `PushCommand.backoff` and `CronJobOptions.backoff`. `DEFAULT_MAX_BACKOFF = 3_600_000` (1h) is the cap when @@ -267,7 +275,8 @@ Allowed transitions (enforced across `pull`/`ack`/`fail` operations and > required and must be a finite number in both modes (TCP/HTTP handlers, > `QueueManager.changeDelay`/`changeWaitingDelay`, and the client job methods before > sending); a negative delay keeps its past run time (`runAt = now + delay`: the job is -> ready at once, ahead of later ready jobs; the PostgreSQL engine applies `now + +> ready at once, ahead of equal-priority FIFO jobs with later run times; priority +> and LIFO ordering still apply. The PostgreSQL engine applies `now + > max(0, delay)`, as on 2.9.10) and one beyond ±`MAX_JOB_DURATION_MS` is clamped, as > 2.9.10 applied them (`delayArgument`); a > non-finite `timestamp` is rejected. In-queue @@ -300,7 +309,8 @@ by `createJob` (`src/domain/job/create.ts`) using `JOB_DEFAULTS` (`src/domain/jo `timestamp`, `delay`, `backoff` (or object `backoff.delay`), `priority` or `maxAttempts` falls back to now, no delay or the default. A negative `delay` is kept as in 2.9.10 (`run_at = created_at + delay`, bounded to ±`MAX_JOB_DURATION_MS`): the job is -never `delayed` and sorts ahead of ready jobs with a later `run_at`. +never `delayed` and, among equal-priority FIFO jobs, sorts ahead of ready jobs +with a later `run_at`. Priority and LIFO ordering still apply. `lifo` is stored as a boolean whatever the input (`1` is `true`), as the heap comparator requires; `ChangePriority` normalizes its `lifo` the same way. @@ -319,7 +329,7 @@ its number; "clamped" values above the maximum are stored as the maximum | Option | Bound | | ------------------------------------------ | ----------------------------------------------------------------------------------------------- | | `priority` | finite (grouped: integer 0..2,097,151) | -| `delay` | finite; negative = a past run time (ready, sorts first); clamped to ±MAX | +| `delay` | finite; negative = a past run time (ready; earlier run times go first among equal-priority FIFO jobs); clamped to ±MAX | | `ttl`, `timeout` | finite, >= 0; clamped | | `stallTimeout` | finite | | `maxAttempts` | any number, ±Infinity allowed; stored 1 or less → 1, fraction rounded up, at most 2,147,483,647 | @@ -333,6 +343,7 @@ its number; "clamped" values above the maximum are stored as the maximum | Option | Type | Default | | --------------------------- | ------------------------------------------------------------------- | ------------ | | `data` | `unknown` (required) | — | +| `name` | `string` (optional) | `'default'` | | `priority` | `number` | `0` | | `delay` | `number` (ms) | `0` | | `maxAttempts` | `number` | `3` | @@ -839,6 +850,7 @@ export interface PushCommand extends BaseCommand { // src/domain/types/commands/core.ts:5-52 readonly cmd: 'PUSH'; readonly queue: string; + readonly name?: string; // processor routing name, default 'default' readonly data: unknown; readonly priority?: number; readonly delay?: number; @@ -860,7 +872,7 @@ export interface PullCommand extends BaseCommand { // src/domain/types/commands/core.ts:65-72 readonly cmd: 'PULL'; readonly queue: string; - readonly timeout?: number; // long-poll ms, finite 0..60000 (QueueManager.pull* clamps) + readonly timeout?: number; // wire long-poll ms, finite 0..60000; direct QueueManager waits differ readonly owner?: string; // lock owner id readonly lockTtl?: number; // default 30000; any finite number (2.9.10), not NaN/Infinity readonly detach?: boolean; // don't auto-release on disconnect diff --git a/docs/features/cli.md b/docs/features/cli.md index 0c06c230..94e3cb80 100644 --- a/docs/features/cli.md +++ b/docs/features/cli.md @@ -187,9 +187,13 @@ are separate, so a healthy IPv6 listener is not misreported as unreachable. ### 7. Backup (`executeBackupCommand`, `src/cli/commands/backup.ts:21`) -Requires a data path from -`BUNQUEUE_DATA_PATH`→`BQ_DATA_PATH`→`DATA_PATH`→`SQLITE_PATH`; otherwise -fails. It accepts temporary S3 credentials (`S3_SESSION_TOKEN` / +Reads the server's config file (`--config`/`-c `, else discovered +`bunqueue.config.{ts,js,mjs}` in the working directory). The data path resolves as +`storage.dataPath` > `BUNQUEUE_DATA_PATH` > `BQ_DATA_PATH` > `DATA_PATH` > +`SQLITE_PATH`; without a path it fails. The selected storage driver must be +SQLite; PostgreSQL and memory are refused. Backup settings use the file's +`backup` section > `S3_*`/`AWS_*` environment > defaults, through +`resolveBackupCommandConfig`. It accepts temporary S3 credentials (`S3_SESSION_TOKEN` / `AWS_SESSION_TOKEN`) and virtual-host addressing (`S3_VIRTUAL_HOSTED_STYLE`). It validates S3 config via `S3BackupManager.validate()`. `restore` refuses without `--force` (`-f`) and @@ -308,7 +312,7 @@ Client-affecting env vars / flags (defaults in parentheses): - `--json` (off), `--help`/`-h`, `--version`/`-v`. - `NO_COLOR` (any non-empty value) disables ANSI output; `FORCE_COLOR` forces it (`0`/`false` force it off); `TERM=dumb` disables it. - Connection timeout 5000 ms (`src/cli/client.ts:171-176`); command timeout has a 30000 ms base (`src/cli/client.ts:44-48`). -- `backup` reads the server's config file (`--config`/`-c `, else `bunqueue.config.{ts,js,mjs}` in the working directory) with the server's precedence: data path ← `storage.dataPath` > `BUNQUEUE_DATA_PATH` > `BQ_DATA_PATH` > `DATA_PATH` > `SQLITE_PATH`, backup settings ← the `backup` section > `S3_*`/`AWS_*` > defaults (`resolveBackupCommandConfig`). A configuration that selects PostgreSQL or memory is refused. +- `backup` uses the server's config-file discovery, storage selection and backup precedence described in the Backup section above. Server-start flags (`bunqueue start`, merged over `bunqueue.config.ts`, CLI wins): `--tcp-port` (6789; 0-65535, 0 lets the OS pick), `--http-port` (6790; diff --git a/docs/features/client-queue-sdk.md b/docs/features/client-queue-sdk.md index defa29a5..be083ffd 100644 --- a/docs/features/client-queue-sdk.md +++ b/docs/features/client-queue-sdk.md @@ -464,7 +464,14 @@ zero. their caller, so use them when the outcome matters. - **Synchronous TCP boundaries**: `getJobs`/`getWaiting`/… (sync) return `[]`, `count()` returns `0`, `getCountsPerPriority()` returns `{}`, and `isPaused()` returns `false` in TCP mode because their signatures cannot await a round trip. Use the corresponding `Async` variants for authoritative remote results. The same rule applies to synchronous DLQ reads and fire-and-forget mutation forms; use `getDlqAsync`, `getDlqStatsAsync`, `retryDlqAsync`, `retryDlqByFilterAsync`, `purgeDlqAsync`, and `retryCompletedAsync` when the result matters. Selective `removeDlqJob` is deliberately Promise-based even without the suffix, and `removeDlqJobAsync` is its explicit alias. Limit getters, worker discovery, dependency methods, deduplication methods, and `moveToWaitingChildren` are asynchronous and now query or mutate the selected broker runtime directly. - **Detached conversion helpers**: broker-returned `Job` instances always receive a complete live operation context. Low-level callers that invoke `createPublicJob` without a context receive only detached fallback behavior and must not treat that helper as a broker client. -- **Idempotency**: `jobId`/`deduplication.id` make `add` idempotent (custom-id dedup, server-side, only while the job is live: completion or the DLQ releases the id). `forward()` uses deterministic remote ids (`fwd::`) so re-forwards don't duplicate (see [Store-and-Forward](./store-and-forward.md)). +- **Duplicate suppression**: custom `jobId` returns a live generation only while + ownership is retained. Ordinary memory/SQLite adds use a bounded cache (50,000 + IDs by default), so eviction can end protection before completion/DLQ; + PostgreSQL checks database identity. `deduplication.id` separately depends on + unique-key ownership/TTL. `forward()` uses deterministic remote IDs with the + same backend limits. Keep external effects idempotent; see + [Deduplication & Unique Jobs](./deduplication-and-unique.md) and + [Store-and-Forward](./store-and-forward.md). - **Metrics/event retention**: `getMetrics(type,start,end)` returns queue-scoped, newest-first one-minute buckets over identical embedded/TCP paths. `trimEvents(maxLength)` returns the exact number removed from that queue's diff --git a/docs/features/concurrency-and-locking.md b/docs/features/concurrency-and-locking.md index 48c00ae7..8328f481 100644 --- a/docs/features/concurrency-and-locking.md +++ b/docs/features/concurrency-and-locking.md @@ -134,7 +134,7 @@ export function checkStalledJobs(ctx: BackgroundContext): void; Lease renewal / heartbeat flow through the TCP handlers (`src/infrastructure/server/handlers/monitoring.ts`) into `QueueManager`: -- `JobHeartbeat` / `JobHeartbeatBatch` → `renewJobLock` when a `token` is present, else updates `job.lastHeartbeat` (`queue-manager/locks.ts`). +- `JobHeartbeat` / `JobHeartbeatB` → `renewJobLock` when a `token` is present, else updates `job.lastHeartbeat` (`queue-manager/locks.ts`). - `ExtendLock` / `ExtendLocks` → `extendLock` → `renewJobLock`. - Lease durations (`PULL`/`PULLB` `lockTtl`, `ExtendLock` `duration`, `ExtendLocks` `durations[i]`, `JobHeartbeat` `duration`) must be finite numbers of any sign, passed @@ -359,7 +359,10 @@ lock.createdAt`. Pull stamps `startedAt` before the lease is created, so a - **Malformed `LOCK_TIMEOUT_MS`** — the variable must be whole milliseconds >= 1 (`0` would fail every contended acquire at once). `lockTimeoutMs()` parses it with `parseDurationEnv` on first use (never at import) and caches it; QueueManager construction calls it before opening storage, so `abc`, `-1`, `0` or `1e3` stops server startup and the first embedded `Queue`/`Worker` with `Invalid LOCK_TIMEOUT_MS: "abc" (expected a whole number of milliseconds >= 1)`. A server checks it earlier, in `resolveServerConfig` (one `Fatal error:` line, before storage opens), and applies the resolved value with `configureLockTimeoutMs` before the QueueManager is built (the config file's `timeouts.lock` is ignored with a warning, as it always was); the rule and default (`LOCK_TIMEOUT_SETTING`) are defined once, in `lockTimeout.ts`. Before, `parseInt` made `abc` or `99999999999` fail every contended acquire after about 1 ms and `-1` at once. - **Resource-slot leaks** — every reclaim path (`handleRecoveryBoundExceeded`, `requeueExpiredJob`, `moveStalliedJobToDlq`, `retryStalliedJob`) calls `shard.releaseJobResources(queue, uniqueKey, groupId, ownerId)` before moving the job; omitting it wedges the queue's concurrency limiter. Passing `ownerId` also prevents a stale generation from releasing a replacement job's unique key. - **Orphan SQLite rows (#97)** — DLQ moves must `saveDlqEntry` + `deleteJob`; missing the delete leaves a `jobs` row that collides on retry with `UNIQUE constraint failed: jobs.id`. -- **Cron preventOverlap (#73/#75)** — `cron:`-prefixed jobs are discarded rather than requeued/DLQ'd on stall or lock expiry, since the scheduler re-creates them on the next tick; requeuing would cause "starts right away on reconnect". +- **Cron preventOverlap (#73/#75)** — memory/SQLite discard `cron:`-prefixed + jobs on stall/lock expiry so the scheduler can recreate them on the next tick. + PostgreSQL's discard checks dependency safety; protected IDs continue through + normal lease recovery instead. - **False-positive suppression** — single-cycle hiccups never trigger action thanks to two-phase detection plus the `gracePeriod` after job start. - **Generation-aware `createLock`** — returns `null` when the job is not processing or the existing lock belongs to the same processing generation. diff --git a/docs/features/deduplication-and-unique.md b/docs/features/deduplication-and-unique.md index 53e3c657..f7b2c304 100644 --- a/docs/features/deduplication-and-unique.md +++ b/docs/features/deduplication-and-unique.md @@ -4,7 +4,15 @@ ## Purpose -Prevents duplicate jobs from entering a queue. Two independent mechanisms cover different needs: **custom job IDs** (`jobId`) make a re-add idempotent — while the prior job is unfinished (waiting / waiting-children / **active**) the add is a no-op that returns the existing job, and once it has _completed_ the deterministic id is recycled for a fresh job (#92) — and **unique keys** (`deduplication.id`) suppress / extend / replace pending jobs within an optional TTL window. Both checks run inside the shard write lock during `pushJob` so concurrent adds of the same key cannot race. This is how store-and-forward edges, cron `preventOverlap`, and BullMQ-style `deduplication`/`debounce` options avoid creating redundant work. +Suppresses duplicate admissions within ownership limits. **Custom job IDs** +(`jobId`) return an unfinished generation while its ownership is retained; +ordinary memory/SQLite adds depend on a bounded ownership LRU (50,000 IDs by +default), whereas PostgreSQL checks live database identity. **Unique keys** +(`deduplication.id`) independently suppress / extend / replace jobs while their +key is owned, within an optional TTL. Local checks run under shard write locks, +but locking does not recover ownership evicted from the custom-ID cache. +Terminal custom IDs can be reused for a fresh generation. Keep external +effects idempotent; neither mechanism promises permanent business deduplication. ## Responsibilities & Scope @@ -130,11 +138,18 @@ before the TCP `PUSH`/`PUSHB` (for example `sdk/python/bunqueue/options.py` and silently ignored by `handleDeduplication`, which returns early without a `uniqueKey`. -SQLite columns: `jobs.unique_key`, `jobs.custom_id` (`schema.ts:34-35`). Note `idx_jobs_unique ON jobs(queue, unique_key) WHERE unique_key IS NOT NULL` (`schema.ts:58-59`) is a **non-unique** index for lookup speed — uniqueness is enforced in memory, not by the database. Custom-id idempotency is backstopped by the `jobs.id` PRIMARY KEY (the custom id becomes the row id). +SQLite columns: `jobs.unique_key`, `jobs.custom_id` (`schema.ts:34-35`). Note +`idx_jobs_unique ON jobs(queue, unique_key) WHERE unique_key IS NOT NULL` +(`schema.ts:58-59`) is a **non-unique** lookup index. The custom ID becomes the +`jobs.id` primary key, but ordinary single/batch persistence uses +`ON CONFLICT(id) DO UPDATE`; the primary key is not a live-generation +deduplication guard after ownership-cache eviction. `durable` changes commit +timing, not this admission limitation. Consequently, custom IDs are broker-wide rather than queue-scoped. A live -generation suppresses the same ID even when the next push names another queue; -the existing job and its original data win. `uniqueKey`, by contrast, remains +generation suppresses the same ID across queues only while the local ownership +cache retains it; PostgreSQL uses the authoritative namespace/id row instead. +`uniqueKey`, by contrast, remains scoped by queue. ## Business Logic / Control Flow @@ -144,7 +159,13 @@ Single push: `pushJob` takes the required write locks and runs both checks insid ### 1. Custom-ID idempotency — `handleCustomId` (`customId.ts`) 1. No `customId` → return a fresh generated id. -2. `id = jobId(input.customId)` (the custom id _is_ the job id). Look up `customIdMap`. If the prior job is still **unfinished** (and not in `completedJobs`), the re-add is an idempotent no-op (BullMQ parity) — never a second insert of the same deterministic id, which would otherwise collide on the `jobs.id` PRIMARY KEY. Three unfinished states are covered: +2. `id = jobId(input.customId)` (the custom id _is_ the job id). Look up + `customIdMap`. **Only on a cache hit**, an unfinished generation (not in + `completedJobs`) can become an idempotent skip. A cache miss has no + authoritative live-row/index rejection; the later checks only identify + terminal/dependency retirement. The existing owner's shard may be locked + via `jobIndex` without repairing that missing admission check. On a hit, + three unfinished states are covered: - **waiting / delayed / prioritized** — `jobIndex` type `queue` and present in the PriorityQueue → return the existing job. - **waiting-children** — `jobIndex` type `queue` but held in `shard.waitingDeps` (its row is already persisted, so it is _not_ in the PriorityQueue) → idempotent skip via the existing id. - **active (processing)** — `jobIndex` type `processing`; the job was popped from the queue and is still running on a worker, with its row still on disk → idempotent skip via the existing id. @@ -216,7 +237,12 @@ RAM-only, uncounted prefix while avoiding a false all-or-nothing guarantee. ### Forwarder dedup key (`src/client/forwarder.ts:97`) -Store-and-forward assigns each forwarded job the deterministic remote id `fwd::`, sent as the remote `jobId`. The server's custom-id idempotency then makes a re-forward after a crash/retry a no-op remotely while the remote job is still live; once it completes or reaches the DLQ the id is released and a re-forward creates a new job. See [Store-and-Forward](./store-and-forward.md). +Store-and-forward assigns the deterministic remote id +`fwd::`, sent as `jobId`. Suppression lasts while +remote live ownership is retained: memory/SQLite use the bounded cache; +PostgreSQL checks the database. Completion/DLQ permits a fresh generation, and +cache eviction can end local-engine protection earlier. Keep remote effects +idempotent. See [Store-and-Forward](./store-and-forward.md). ### Bunqueue auto-dedup/debounce (`dedupDebounce.ts:21`) @@ -248,7 +274,13 @@ Lifecycle / release: ## Edge Cases & Failure Modes - **TTL semantics:** `ttl` undefined ⇒ `expiresAt = null` ⇒ never expires (`calculateExpiration`, `deduplication.ts:60`). Expiry is lazy on read plus a periodic sweep. -- **Re-add of an active / waiting-children custom id is idempotent:** when the prior job has been popped from the PriorityQueue (active = `jobIndex` type `processing`; or waiting-children = type `queue` but held in `shard.waitingDeps`) its row is still on disk, so re-adding the same `jobId` must NOT take the reuse path — that would re-insert the same deterministic id and collide on the `jobs.id` PRIMARY KEY (durable → throws `UNIQUE constraint failed: jobs.id`; buffered → the write buffer silently drops the insert, turning the intended no-op into a reject). `handleCustomId` instead returns the existing job (or, when the live `Job` is unreachable without `processingShards`, an `existingId` result) and inserts nothing (`src/application/operations/customId.ts:45-65`). This is the custom-id twin of the unique-key fix #69. The guard is `!ctx.completedJobs.has(id)` plus a `queue`/`processing` location check, so completed (#92) and DLQ jobs still fall through to reuse. +- **Active / waiting-children custom IDs require retained local ownership:** on + a `customIdMap` hit, `handleCustomId` checks `!completedJobs.has(id)` plus the + queue/processing location and returns the existing job/id without insertion. + After eviction, those live checks are skipped. SQLite's upsert does not turn + that miss into uniqueness rejection. This is a correctness limitation to + investigate with a failing public-path regression, not supported live-job + replacement semantics. No evicted-live-ID runtime reproduction is claimed here. - **Custom-id reuse after completion (#92):** completed rows survive on disk; the reuse path evicts them so the recycled id does not falsely report `completed` or collide on the PK at flush (`src/application/operations/customId.ts:67-73`). - **Custom-id reuse after DLQ:** a terminal DLQ generation is retired from SQLite in the same transaction that inserts the replacement jobs row (durable or not), and then from the owning shard, `jobIndex` and its O(1) counter. Both `PUSH` and `PUSHB` share this path, so the deterministic id has exactly one observable generation and the old entry cannot keep `failed` inflated or reappear after restart. - **Orphan-row reconciliation via upsert:** a durable `jobs` row can outlive its in-memory tracking when the legacy fire-and-forget `obliterate()` races an in-flight durable insert, or when a write-buffer flush is reordered relative to that insert. (`obliterateAsync()` is the authoritative ordered API for new code.) The `customIdMap` / `jobIndex` entries may be cleared while the row survives. A later re-add of the same custom id finds no mapping → reuse path → insert. To stop the recycled id colliding on the PK (`UNIQUE constraint failed: jobs.id`), **both** insert statements use `ON CONFLICT(id) DO UPDATE` (upsert): the orphan row is overwritten in place at insert time. The `DO UPDATE SET` clause covers **all** non-id columns — including the per-execution fields `started_at` / `completed_at` / `progress` / `progress_msg` / `last_heartbeat` / `stacktrace`, which are absent from the INSERT column list so `excluded.` resolves to their DEFAULT (the fresh-job value). Without resetting these, an upsert over a previously-completed orphan would leave a brand-new `waiting` job reporting `progress=100` / a stale `completed_at` / a prior life's `stacktrace`. A brand-new id is a plain INSERT with **zero** extra cost, so the customId hot path is not taxed (`statements.ts` `insertJob`, `sqliteBatch.ts` batch insert). In the buffered batch path this also prevents one collision from failing the whole flush and dropping every innocent job batched in the same window (`reportLostJobs`). `jobs` has no foreign keys; its completion-count triggers intentionally observe the `DO UPDATE` state/queue transition, while avoiding `REPLACE`'s separate DELETE+INSERT lifecycle. Covered by `test/repro-idem-active-readd.test.ts`. @@ -286,7 +318,18 @@ Lifecycle / release: single job or batch prefix is still persisted, counted, announced and notified before the cleanup error is surfaced; a cleanup exception cannot leave committed jobs invisible to accounting. -- **Memory bounds:** `customIdMap` is an `LRUMap` capped at `maxCustomIds` (default `50_000` in `application/types/config.ts`; initialized in `queue-manager/state.ts`). Unique-key registries are swept every cleanup tick (`cleanExpiredUniqueKeys`, `cleanupTasks.ts`) and any single-queue registry exceeding **1000** entries is force-trimmed by half via insertion-order iteration — under churn a still-live key can be evicted, weakening (not breaking) dedup. LRU eviction of `customIdMap` likewise allows a re-add to slip through. +- **Memory bounds:** ordinary memory/SQLite `customIdMap` is an `LRUMap` capped + at `QueueManagerConfig.maxCustomIds` (default `50_000` in + `application/types/config.ts`; initialized in `queue-manager/state.ts`). + Eviction is not live-job-aware; after it, a re-add is not guaranteed to + preserve the original generation. This is not repaired by SQLite's primary + key or durable admission. Only the low-level QueueManager config exposes the + cap: no Queue/Worker option, typed standalone-server field, CLI flag or env + setting does. PostgreSQL checks locked database identity independently. + Atomic flow creation separately checks `jobIndex`/SQLite rows or PostgreSQL + `rejectExisting`, rejecting existing IDs; ordinary adds retain their own limit. + Unique-key registries are separately swept every cleanup tick and force-trimmed + by half above 1000 per-queue entries; that can also end suppression early. - **Debounce is metadata-only (gotcha):** `debounce` sets `job.debounceId`/`job.debounceTtl` (`operations/add/single.ts:63-64`, `operations/add/bulk.ts:63-64`, `domain/job/create.ts:103-104`) but does **not** diff --git a/docs/features/flow-producer.md b/docs/features/flow-producer.md index 561622d6..9acbc844 100644 --- a/docs/features/flow-producer.md +++ b/docs/features/flow-producer.md @@ -163,7 +163,8 @@ not trusted. It checks: `NOT NULL`); `backoff.delay` must be a finite number `>= 0` and `backoff.maxDelay` `0..86_400_000` ms. Unsupported repeat, deduplication and debounce inputs are rejected before the rules; -- a 10 MB per-job and 64 MB aggregate flow-data bound; +- a JSON string length bound of 10 × 1024 × 1024 characters per job and + 64 × 1024 × 1024 characters across the graph (not byte counts); - unique IDs, dependencies, children, and parent/metadata back-references; - duplicate edges and graph cycles; - mutually exclusive failure policies; diff --git a/docs/features/http-api.md b/docs/features/http-api.md index d908e7c7..3e8ec1a6 100644 --- a/docs/features/http-api.md +++ b/docs/features/http-api.md @@ -252,10 +252,14 @@ Resolved in `config/resolve.ts` (config file > env > default) and passed into `H | `RATE_LIMIT_WINDOW_MS` / `RATE_LIMIT_MAX_REQUESTS` / `RATE_LIMIT_CLEANUP_MS` | rate limiter | `60000` / `10000` / `60000` (`rateLimiter.ts`) | `createHttpServer` reads `rateLimiterEnvConfig()` before it subscribes or binds, so a -malformed `RATE_LIMIT_*` value (whole numbers >= 1 only) fails server creation instead -of every request; the standalone server reports it even earlier, from +rejected `RATE_LIMIT_*` value fails server creation instead of every request; +window 0 disables limiting, while compatible fallbacks warn for negative windows +(disabled), unreadable request limits (unlimited), and 0/negative/unreadable cleanup periods +(default sweep). The standalone server +reports errors even earlier, from `resolveServerConfig`, before the banner. See -[Rate Limiting & Concurrency](./rate-limiting-and-concurrency.md). +[Rate Limiting & Concurrency](./rate-limiting-and-concurrency.md#configuration) +for the complete input rules. WS/SSE tuning constants are compile-time, not env-configurable: SSE `MAX_CLIENTS=1000`, `HEARTBEAT_MS=30000`, `RETRY_MS=3000`, `EVENT_BUFFER_SIZE=1000`; WS `MAX_WS_CLIENTS=1000`, `BACKPRESSURE_BYTES=1048576`, idle `120s`, `maxPayloadLength=1 MiB`. diff --git a/docs/features/job-lifecycle.md b/docs/features/job-lifecycle.md index 2900e318..e93a9737 100644 --- a/docs/features/job-lifecycle.md +++ b/docs/features/job-lifecycle.md @@ -115,7 +115,7 @@ plus the batch helpers `ExtractedJob`/`BatchContext`/`FinalizeContext` and `groupByProcShard`/`extractJobs`/`groupByQueueShard`/`releaseResources`/ `finalizeBatchAck` in `ackHelpers.ts`. -`isTimedOut(job, now)` follows `processingDeadline` (`src/domain/job/timeoutRule.ts`): an absent, `0` or NaN `timeout` never times out, a fractional deadline rounds up, and the deadline itself is due, exactly as the timeout scheduler fails jobs. `QueueManager.pull`/`pullWithLock`/`pullBatch`/`pullBatchWithLock` reject a `timeoutMs` outside 0..60,000 ms (NaN, Infinity, negative) with the TCP `PULL` message, before claiming anything. +`isTimedOut(job, now)` follows `processingDeadline` (`src/domain/job/timeoutRule.ts`): an absent, `0` or NaN `timeout` never times out, a fractional deadline rounds up, and the deadline itself is due, exactly as the timeout scheduler fails jobs. Direct `QueueManager.pull`/`pullWithLock`/`pullBatch`/`pullBatchWithLock` use `pullTimeoutArgument`: numeric strings are converted; negative, NaN and non-number values mean no wait. Positive waits above 60,000 ms are honored, and Infinity waits until delivery or signal abort. TCP/HTTP `PULL`/`PULLB` instead require a finite timeout in 0..60,000 ms before claiming anything. Job-type helpers re-exported from the `src/domain/types/job.ts` facade: `createJob`, `generateJobId`, `jobId`, `calculateBackoff`, `canRetry`, `isReady`, `isDelayed`, `isExpired`, `isTimedOut`, `normalizeStacktrace`, `createJobLock`, `renewLock`, `isLockExpired`, and the `JobState` const enum. @@ -135,7 +135,7 @@ See [data-model](../data-model.md) for full definitions. The central shape is `J - Failure: `stacktrace` (capped at `stackTraceLimit`, default `10`). - `timeline: JobTimelineEntry[]` — capped at `MAX_TIMELINE_ENTRIES = 20` and persisted as a MessagePack BLOB on lifecycle transitions, so it survives SQLite recovery. -`JobInput` (`src/domain/types/jobs/model.ts:92-137`) is the wire/SDK input; `createJob` (`src/domain/job/create.ts:108-151`) fills defaults from `JOB_DEFAULTS` (`src/domain/job/constants.ts:5-13`). Note `removeOnComplete`/`removeOnFail` are coerced via `toBoolean` because the wire boundary is not runtime-type-safe (#90, `src/domain/job/create.ts`). Every public admission path (TCP/HTTP, flows, embedded `Queue.add`/`addBulk`, cron templates) first runs the shared `validateJobOptions` ([Job Options Validation](./job-options-validation.md)), which refuses only what cannot run, so every option value 2.9.10 ran is admitted. `createJob` first applies `normalizeJobInput` (`src/domain/job/optionNormalize.ts`, also applied by `QueueManager.push/pushBatch` and the PostgreSQL engine): a numeric string is its number, `maxAttempts` 0 is stored as 1 (it runs once, as on 2.9.10) and a fraction rounds up (so the attempts made never exceed `maxAttempts`), `maxAttempts` above 2,147,483,647 or Infinity is that maximum, and `delay`/`timeout`/`ttl`/`dedup.ttl`/`debounceTtl`/`repeat.every` beyond `MAX_JOB_DURATION_MS` (4.32e15 ms) are clamped. `createJob` still defends internal callers and stored rows: `parseBackoff` turns the object form of `backoff` into `backoffConfig` and keeps `maxDelay` only when it is a finite number in `0..MAX_BACKOFF_DELAY` (24 h), and a NaN or non-number `timestamp`, `delay`, `backoff`/`backoff.delay`, `priority` or `maxAttempts` falls back to now, no delay or the default, so no NaN reaches a `NOT NULL` SQLite column (the write buffer would drop the row). A `null` backoff is treated as absent. A negative `delay` is accepted by the validator and kept as on 2.9.10: `runAt = createdAt + jobRunDelay(delay)` (`src/domain/job/options.ts`, bounded to ±`MAX_JOB_DURATION_MS`), a run time in the past, so the job is created `waiting` (never `delayed`), stays out of the delayed index and, since ready jobs are ordered by `runAt`, goes ahead of ready jobs with a later run time. `ChangeDelay`/`MoveToDelayed` keep a negative delay the same way (`runAt = now + delay`). +`JobInput` (`src/domain/types/jobs/model.ts:92-137`) is the wire/SDK input; `createJob` (`src/domain/job/create.ts:108-151`) fills defaults from `JOB_DEFAULTS` (`src/domain/job/constants.ts:5-13`). Note `removeOnComplete`/`removeOnFail` are coerced via `toBoolean` because the wire boundary is not runtime-type-safe (#90, `src/domain/job/create.ts`). Every public admission path (TCP/HTTP, flows, embedded `Queue.add`/`addBulk`, cron templates) first runs the shared `validateJobOptions` ([Job Options Validation](./job-options-validation.md)), which refuses only what cannot run, so every option value 2.9.10 ran is admitted. `createJob` first applies `normalizeJobInput` (`src/domain/job/optionNormalize.ts`, also applied by `QueueManager.push/pushBatch` and the PostgreSQL engine): a numeric string is its number, `maxAttempts` 0 is stored as 1 (it runs once, as on 2.9.10) and a fraction rounds up (so the attempts made never exceed `maxAttempts`), `maxAttempts` above 2,147,483,647 or Infinity is that maximum, and `delay`/`timeout`/`ttl`/`dedup.ttl`/`debounceTtl`/`repeat.every` beyond `MAX_JOB_DURATION_MS` (4.32e15 ms) are clamped. `createJob` still defends internal callers and stored rows: `parseBackoff` turns the object form of `backoff` into `backoffConfig` and keeps `maxDelay` only when it is a finite number in `0..MAX_BACKOFF_DELAY` (24 h), and a NaN or non-number `timestamp`, `delay`, `backoff`/`backoff.delay`, `priority` or `maxAttempts` falls back to now, no delay or the default, so no NaN reaches a `NOT NULL` SQLite column (the write buffer would drop the row). A `null` backoff is treated as absent. A negative `delay` is accepted by the validator and kept as on 2.9.10: `runAt = createdAt + jobRunDelay(delay)` (`src/domain/job/options.ts`, bounded to ±`MAX_JOB_DURATION_MS`), a run time in the past, so the job is created `waiting` (never `delayed`) and stays out of the delayed index. Among equal-priority FIFO jobs, earlier `runAt` values go first; priority and LIFO ordering still apply. `ChangeDelay`/`MoveToDelayed` keep a negative delay the same way (`runAt = now + delay`). `JobState` enum values: `waiting`, `prioritized`, `delayed`, `active`, `completed`, `failed` (`src/domain/types/jobs/model.ts:4-11`). The additional logical states `waiting-children` (dependency gate) and `paused` (queue-level) are represented via shard membership (`waitingDeps`/`waitingChildren`) and `QueueState.paused`, not the enum. @@ -157,7 +157,11 @@ admission. 1. Compute `idx = shardIndex(queue)` and acquire every target, custom-ID owner, and parent shard write lock required by the request in ascending order. 2. **Inspect custom-ID idempotency** (`handleCustomId`, `customId.ts`). A live - generation is an idempotent skip. Reusing a completed, DLQ, or + generation is an idempotent skip only on a retained `customIdMap` hit. + Memory/SQLite ordinary adds use a bounded LRU (50,000 default); cache misses + do not perform authoritative live-generation rejection, and SQLite upserts + do not repair that limit. PostgreSQL uses separate locked database checks; + see [Deduplication & Unique Jobs](./deduplication-and-unique.md). Reusing a completed, DLQ, or payload-free dependency-completion generation produces a retirement plan; it does not delete the old generation yet. When a completed generation has left the bounded hot cache, the lookup consults SQLite so its row and stale @@ -467,7 +471,13 @@ detached CLI, and durable recovery paths. ## Edge Cases & Failure Modes -- **Idempotent push**: a live `customId` returns the existing job (no insert). A recycled terminal `customId` evicts the stale completed row or DLQ entry first, including persisted terminal state, counters, result, logs, and their bounded owner indexes, so state queries and auxiliary reads expose only the new generation; timeout markers are cleared too (#33/#75). +- **Custom-ID push**: a live `customId` returns the existing job only while + ownership is retained by the local bounded cache; PostgreSQL instead checks + database identity. Cache eviction is not terminal reuse and does not promise + preservation of the live generation. A recycled terminal `customId` retires + stale completed/DLQ state, counters, result, logs, owner indexes and timeout + markers before publishing the successor (#33/#75). See + [Deduplication & Unique Jobs](./deduplication-and-unique.md). - **Dedup of active jobs**: default strategy treats an active job (in `jobIndex`, not in queue) as a duplicate and returns its id without inserting (`src/application/operations/pushDeduplication.ts:96-110`); the returned placeholder carries the correct existing id (`src/application/operations/push.ts:76-85`). - **Pull loss prevention**: `requeueJob` restores any job that fails to move to processing. - **Expired (TTL) jobs** are silently dropped during pull (`isExpired`) and never delivered. Their SQLite row (or pending buffered INSERT) is deleted before the heap/counter/index removal, preventing restart resurrection while keeping the shard critical section synchronous. diff --git a/docs/features/job-options-validation.md b/docs/features/job-options-validation.md index cc2c058a..65eea626 100644 --- a/docs/features/job-options-validation.md +++ b/docs/features/job-options-validation.md @@ -155,7 +155,7 @@ larger finite value is accepted and stored as the maximum. | Field (wire name) | Rule | | ------------------------------ | ------------------------------------------------------------------ | | `priority` | any finite number; with `groupId`: `group.priority` integer 0..2,097,151 | -| `delay` | finite; a negative delay is a past run time (ready at once, ahead of later ready jobs); clamped to ±`MAX_JOB_DURATION_MS` | +| `delay` | finite; a negative delay is a past run time (ready at once; among equal-priority FIFO jobs, earlier run times go first); clamped to ±`MAX_JOB_DURATION_MS` | | `ttl`, `timeout` | finite, at least 0; clamped to `MAX_JOB_DURATION_MS` | | `stallTimeout` | finite (a negative value stalls at once, as 0 does) | | `maxAttempts` (`attempts`) | any number, ±Infinity included; stored as `attemptCount`: 1 or less → 1 (runs once), a fraction rounds up, at most 2,147,483,647 | @@ -189,14 +189,16 @@ Why these bounds: flows, cron templates, MCP, Cloud, PostgreSQL): `delay: runAt - Date.now()` turns negative once `runAt` has passed, and `createJob` stores `runAt = createdAt + jobRunDelay(delay)`, a run time in the past. The job is `waiting`, never `delayed` - and never in the delayed index, and because the waiting queue (and the PostgreSQL - claim order) sorts ready jobs by `runAt`, it goes AHEAD of ready jobs whose run time - is later, as on 2.9.10 (a 2.11 candidate clamped it to now, which put it behind them). + and never in the delayed index. Among equal-priority FIFO jobs, the waiting queue + (and the PostgreSQL claim order) sorts by `runAt`, so it goes ahead of ready jobs + whose run time is later, as on 2.9.10; priority and LIFO ordering still apply. + A 2.11 candidate clamped it to now, which put it behind those FIFO jobs. The public `job.delay` and `job.opts.delay` report 0 (`effectiveJobDelay`), as a job read back from the broker does. ChangeDelay/MoveToDelayed apply a negative delay the same way in the SQLite/embedded engine (`delayArgument`: `runAt = now + delay`); the PostgreSQL engine keeps its own 2.9.10 result, `now + max(0, delay)` (a negative - ChangeDelay/MoveToDelayed makes the job ready now, behind earlier ready jobs): the + ChangeDelay/MoveToDelayed makes the job ready now, behind otherwise equal-priority + FIFO jobs with earlier run times): the engines differed on 2.9.10 and keep differing. The public `moveToDelayed(timestamp)` keeps 2.9.10's client-side `max(0, timestamp - now)`, so a past timestamp means "now" (a plain decimal string timestamp is read as its number, as 2.9.10's `-` coerced it). @@ -232,8 +234,11 @@ directly, legacy cron templates, stored rows) are still made safe: - `createJob` never stores NaN or a non-number in `createdAt`, `runAt`, `backoff`, `priority` or `maxAttempts`: it falls back to now / no delay / the defaults, so SQLite never drops the row. Infinity is kept (it is a valid "never" for these fields). A - negative `delay` is stored as no delay (`effectiveJobDelay`, also used by the client - to report the added job's `delay`). + negative `delay` is stored as a past run time (`createdAt + jobRunDelay(delay)`, + bounded to ±`MAX_JOB_DURATION_MS`), so it is ready immediately. Among + equal-priority FIFO jobs, it sorts ahead of jobs with later run times; + priority and LIFO ordering still apply. Only the client-facing `delay` projection is nonnegative + (`effectiveJobDelay` reports 0). - `isReady(job)` is `!(job.runAt > now)`, the complement of `isDelayed`, as every state view already classifies jobs: a NaN run time is ready, not "waiting" forever. This also removes a synchronous infinite loop in the group scheduler (promote, not ready, @@ -278,7 +283,8 @@ embedded jobs, MCP, Cloud): engine still clamps to [0, 100]. - `Update`: JSON serializable (`validateUpdatedJobData`), with no size limit, as on 2.9.10 in both modes (an 11 MB `updateData` succeeded, and an embedded `add` never - checked the data size either). TCP `PUSH`/`PUSHB` keep their 10 MB data limit. + checked the data size either). TCP `PUSH`/`PUSHB` keep their limit of + 10 × 1024 × 1024 JSON string characters, rather than a byte cap. - PUSH `lifo` is stored as a boolean (`createJob`): `1` and `true` each sorted before the other in the heap comparator. - `ClearLogs`: `keepLogs` is applied as 2.9.10 applied it (`keepLogsArgument`): omitted, @@ -332,7 +338,7 @@ recovered. Queue jobs, FlowProducer jobs and DLQ jobs, plus object and invalid progress, in both modes. - `test/repro-job-options-negative-delay.test.ts`: a negative `delay` is a `waiting` - job pulled at once, ahead of later ready jobs (`runAt === createdAt + delay`, empty + job pulled at once, ahead of equal-priority FIFO jobs with later run times (`runAt === createdAt + delay`, empty delayed index) through embedded and TCP `add`/`addBulk`, flows, HTTP `POST /queues/:q/jobs` and `/jobs/bulk`, MCP `add_job`/`add_jobs_bulk`/`add_flow`/`add_cron` and Cloud `job:push`; NaN, diff --git a/docs/features/job-queries-and-control.md b/docs/features/job-queries-and-control.md index 11a023db..1c251f93 100644 --- a/docs/features/job-queries-and-control.md +++ b/docs/features/job-queries-and-control.md @@ -432,7 +432,8 @@ are PostgreSQL-specific and do not alter the synchronous SQLite implementation. `updateJobData` requires JSON-serializable data, with no size limit; `changeDelay`/ `moveToDelayed`/`changeWaitingDelay` take any finite delay (`delayArgument`: a negative one keeps its past run time, `now + delay`, so the job is ready at once and sorts - ahead of later ready jobs, as on 2.9.10); `clearLogs` applies `keepLogsArgument` (0 or less + ahead of equal-priority FIFO jobs with later run times, as on 2.9.10; priority + and LIFO ordering still apply); `clearLogs` applies `keepLogsArgument` (0 or less clears all, a fraction keeps its whole part) and rejects only NaN or text; `pull*` honour any `timeoutMs` (`pullTimeoutArgument`: negative or NaN is no wait, no 60 s cap) and `pullWithLock`/`extendLock` grant any finite lease. See [Job Options Validation](./job-options-validation.md). diff --git a/docs/features/mcp-server.md b/docs/features/mcp-server.md index d58ee3f9..edf318ae 100644 --- a/docs/features/mcp-server.md +++ b/docs/features/mcp-server.md @@ -102,7 +102,7 @@ Registration files (`setupTools`, `toolSetup.ts`): `jobTools.ts` 11, `jobMgmtToo **Tool parameters and semantics.** zod validates every input before either backend runs; the numeric bounds mirror the broker's job-option validation (`tools/schemas.ts:15-23`), so embedded and TCP reject the same inputs. Every queue-name field (`queue`, flow `queueName`, worker `queues[]`, the optional webhook `queue`) and every job name is non-empty. - **`jobs`.** - - `add_job` and each `add_jobs_bulk` item take `queue`, `name`, `data` plus the shared options (`jobOptionsShape`, `schemas.ts:159-183`): `priority` (integer, ±1 000 000), `delay` (at most one year; a negative delay makes the job ready at once with a past run time, `jobDelayField`), `attempts` (integer 1–1000), `backoff` (a number 0–86 400 000 used as the exponential base — retry _n_ waits about base × 2^_n_, jittered, capped at 1 h — or `{ type: 'fixed' | 'exponential', delay, maxDelay? }`), `timeout` and `stallTimeout` (0–86 400 000 ms), `jobId` (custom id, 1–1024 characters of well-formed Unicode; it becomes the job id, ids are server-wide, a re-add while that job is unfinished returns the existing id), `deduplication` (`{ id, ttl? 1 ms–1 year, extend? (requires ttl), replace? }`, scoped to the queue), `removeOnComplete`, `removeOnFail` (booleans only), `durable`, `lifo`, `tags` (strings ≤ 256 chars). `groupId` is not exposed. Both backends map these through `backend/jobOptions.ts` (`toJobInput` for embedded push and TCP `PUSHB` items, where the custom id travels as `customId`; `toPushFields` for TCP `PUSH`, where it travels as `jobId`); undefined fields are omitted. + - `add_job` and each `add_jobs_bulk` item take `queue`, `name`, `data` plus the shared options (`jobOptionsShape`, `schemas.ts:159-183`): `priority` (integer, ±1 000 000), `delay` (at most one year; a negative delay makes the job ready at once with a past run time, `jobDelayField`), `attempts` (integer 1–1000), `backoff` (a number 0–86 400 000 used as the exponential base — retry _n_ waits about base × 2^_n_, jittered, capped at 1 h — or `{ type: 'fixed' | 'exponential', delay, maxDelay? }`), `timeout` and `stallTimeout` (0–86 400 000 ms), `jobId` (custom id, 1–1024 characters of well-formed Unicode; becomes the job id; broker-wide, namespace-wide on PostgreSQL; returns the unfinished generation only while ownership is retained), `deduplication` (`{ id, ttl? 1 ms–1 year, extend? (requires ttl), replace? }`, scoped to the queue), `removeOnComplete`, `removeOnFail` (booleans only), `durable`, `lifo`, `tags` (strings ≤ 256 chars). Memory/SQLite ordinary admission uses a bounded ownership cache (50,000 IDs by default), whereas PostgreSQL checks database identity; see [Deduplication & Unique Jobs](./deduplication-and-unique.md). `groupId` is not exposed. Both backends map these through `backend/jobOptions.ts` (`toJobInput` for embedded push and TCP `PUSHB` items, where the custom id travels as `customId`; `toPushFields` for TCP `PUSH`, where it travels as `jobId`); undefined fields are omitted. - `add_jobs_bulk` is ordered, not atomic: invalid options reject the whole call, but a job the broker refuses during admission leaves the jobs before it added. - `get_job` returns the job with its `state` and its options (`backoff` always; `timeout`, `stallTimeout`, `lifo`, `removeOnComplete`, `removeOnFail`, `tags`, `deduplicationId` when set). `get_job_by_custom_id` finds only unfinished jobs (the custom id is released on completion or final failure; `get_job` still finds the finished job) and also reports `state` (TCP resolves it with an extra `GetState`, `backend/tcp/jobs.ts:92-97`). - `get_job_state`: `waiting`, `prioritized`, `delayed`, `active`, `completed`, `failed` (DLQ), `waiting-children`, or `unknown`. @@ -129,9 +129,9 @@ Registration files (`setupTools`, `toolSetup.ts`): `jobTools.ts` 11, `jobMgmtToo - `get_dlq_stats` returns `{ queue, total, byReason, pendingRetry, expired, oldestEntry, newestEntry }`; `byReason` carries every reason the engine reports. - `retry_dlq` (optional `jobId`; TCP sends it as `jobId`), `purge_dlq` and `retry_completed` (optional `jobId`) return counts. - A supplied retry `jobId` must contain non-whitespace text; empty or blank IDs are rejected before either backend is called. IDs are validated without trimming or rewriting them. Omit `jobId` explicitly for a bulk retry; bulk `retry_completed` remains subject to the configured confirmation guard. -- **`cron`.** `add_cron` takes `name` (non-empty), `queue`, `data`, `schedule` (5/6-field pattern or shortcut) or `repeatEvery` (integer ≥ 1), plus (`tools/cronOptions.ts:41-84`): `jobName`, `priority` (bounded integer), `timezone` (an IANA zone accepted by `Bun.cron.parse`; requires `schedule`, `cronOptions.ts:110-112`), `maxLimit` (integer ≥ 1), `immediately`, `skipIfNoWorker`, `preventOverlap` (default true), `skipMissedOnRestart` (default true), `deduplication` (mapped to `uniqueKey` + `dedup`), and job options `attempts`, `backoff`, `timeout`, `delay` (a negative delay is treated as 0), `stallTimeout`, `removeOnComplete`, `removeOnFail` (mapped to `jobOptions`). Re-adding a name replaces the schedule and keeps its run count; only one job per schedule is pending at a time under `preventOverlap`. `add_cron`, `list_crons` and `get_cron` return `SerializedCron` (with `jobName`, `priority`, `timezone`, `maxLimit`); the policy flags, deduplication and job options are applied but not echoed (see [Edge Cases](#edge-cases--failure-modes)). +- **`cron`.** `add_cron` takes `name` (non-empty), `queue`, `data`, `schedule` (5/6-field pattern or shortcut) or `repeatEvery` (integer ≥ 1), plus (`tools/cronOptions.ts:41-84`): `jobName`, `priority` (bounded integer), `timezone` (an IANA zone accepted by `Bun.cron.parse`; requires `schedule`, `cronOptions.ts:110-112`), `maxLimit` (integer ≥ 1), `immediately`, `skipIfNoWorker`, `preventOverlap` (default true), `skipMissedOnRestart` (default true), `deduplication` (mapped to `uniqueKey` + `dedup`), and job options `attempts`, `backoff`, `timeout`, `delay` (a negative delay keeps a past run time, ready immediately), `stallTimeout`, `removeOnComplete`, `removeOnFail` (mapped to `jobOptions`). Re-adding a name replaces the schedule and keeps its run count; only one job per schedule is pending at a time under `preventOverlap`. `add_cron`, `list_crons` and `get_cron` return `SerializedCron` (with `jobName`, `priority`, `timezone`, `maxLimit`); the policy flags, deduplication and job options are applied but not echoed (see [Edge Cases](#edge-cases--failure-modes)). - **`limits`.** `set_rate_limit` takes `limit` (integer ≥ 1) and `duration` (integer window in ms, 1 to one year, default 1000) and returns `durationMs`; the limiter is a continuously refilling token bucket. `set_concurrency` takes `limit` (integer ≥ 1). `get_queue_limits` returns `QueueLimits` (`{ queue, paused, rateLimit: { max, durationMs } | null, rateLimitTtlMs, rateLimited, concurrencyLimit, active, concurrencyMaxed }`). -- **`flows`.** `add_flow` (root and recursive `children`), `add_flow_chain` (`steps` ≥ 1) and `add_flow_bulk_then` (`parallel` ≥ 1, `final`) take a **strict** `opts` object (`flowTools.ts:28-47`): `priority`, `delay` (a negative delay is treated as 0), `attempts`, `backoff`, `timeout`, `jobId` (must be new and contain no `:`), `removeOnComplete`, `removeOnFail`, `lifo`, `stallTimeout`. `deduplication`, `tags` and `durable` are rejected (the broker rejects dedup in flows, stores no tags on flow jobs, and flows are always durable). `get_flow` takes integer `depth` ≥ 1 (default 10) and `maxChildren` ≥ 1. +- **`flows`.** `add_flow` (root and recursive `children`), `add_flow_chain` (`steps` ≥ 1) and `add_flow_bulk_then` (`parallel` ≥ 1, `final`) take a **strict** `opts` object (`flowTools.ts:28-47`): `priority`, `delay` (a negative delay keeps a past run time, ready immediately), `attempts`, `backoff`, `timeout`, `jobId` (must be new and contain no `:`), `removeOnComplete`, `removeOnFail`, `lifo`, `stallTimeout`. `deduplication`, `tags` and `durable` are rejected (the broker rejects dedup in flows, stores no tags on flow jobs, and flows are always durable). `get_flow` takes integer `depth` ≥ 1 (default 10) and `maxChildren` ≥ 1. - **`workers`.** `register_worker` (`queues` ≥ 1); `register_handler` (`url`, `method` `GET`/`POST`/`PUT`/`DELETE`, `headers?`, `body?`, `timeoutMs` 1 000–120 000, default 30 000). - **`webhooks`.** `add_webhook` accepts only `WEBHOOK_EVENTS`; `queue` is optional. - **`monitoring`.** `get_stats` shape differs by backend (see [Backend dispatch](#business-logic--control-flow)); `get_per_queue_stats` returns `waiting`, `prioritized`, `delayed`, `active`, `dlq` per queue, preserving every queue name (including `__proto__`) as an own enumerable property; `get_memory_stats` returns the `QueueManager.getMemoryStats()` collection sizes in both modes. @@ -344,6 +344,10 @@ The MCP server holds no locks of its own; concurrency safety is delegated to the | `BUNQUEUE_MCP_WORKFLOW_QUEUE` | `__wf:steps` | The Engine's step queue (`EngineOptions.queueName`), where `signal_workflow` enqueues the resume job. Ignored (stderr notice) without `BUNQUEUE_MCP_WORKFLOW_DB`. | | `BUNQUEUE_CLOUD_URL` | — | When set in embedded mode, enables `CloudAgent` MCP telemetry. Its configuration is resolved and validated before any transport starts, so an invalid `BUNQUEUE_CLOUD_*` value exits 1 before the HTTP transport binds. Other `BUNQUEUE_CLOUD_*` vars apply via the agent — see [bunqueue Cloud Dashboard Integration](./cloud-integration.md). | +The admission limit labeled 10 MB above is 10 × 1024 × 1024 JSON string +characters, not a byte cap (`validateJobData`). Job-data updates intentionally +have no size cap; apply application payload limits separately. + ## Related Docs - [architecture](../architecture.md), [data-model](../data-model.md) diff --git a/docs/features/model-based-testing.md b/docs/features/model-based-testing.md index 24eaefa1..ab62d976 100644 --- a/docs/features/model-based-testing.md +++ b/docs/features/model-based-testing.md @@ -188,8 +188,9 @@ as a command that does not throw. The generated transitions also assert the following contracts: - only legal state-graph edges occur; -- custom IDs are idempotent while live and create a new generation only after a - terminal state; +- custom IDs are idempotent while live within the generated campaign's retained + ownership working set, and terminal reuse creates a new generation; this does + not prove ordinary memory/SQLite admission safe after ownership-cache eviction; - `attempts <= maxAttempts` and `stallCount <= maxStalls`; exceeding either bound is terminal; - crash recovery increments the persisted counters, never leaves a ghost @@ -476,7 +477,7 @@ violates it or the required state is not observable. | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- | | C6 | Every persisted lifecycle transition is atomic under injected I/O failure | Existing failure tests are examples, not a transition-by-transition storage fault model | | C7 | Pagination has a no-duplicate/no-skip contract under concurrent mutation | Static ordering is deterministic, but no snapshot/cursor contract exists for mutating result sets | -| C8 | Strong custom-ID/unique-key idempotency holds for every live job | Bounded LRU/registry trimming intentionally weakens the guarantee after eviction | +| C8 | Strong custom-ID/unique-key idempotency holds for every live job | Ordinary memory/SQLite custom-ID checks depend on a bounded ownership LRU; evicted-live correctness needs its own regression/investigation. Unique-key trimming separately ends ownership. | | C9 | Internal resource conservation is directly observable for unique keys, groups, limiter tokens, temporal reverse indexes, and dependency reverse edges | `/stats` exposes aggregate subsets, so exact independent cardinality oracles need diagnostic counters | | C10 | JSON monitoring totals remain exact across the full `bigint` range | TCP/HTTP stats convert to `number` and lose precision above 2^53 | diff --git a/docs/features/persistence.md b/docs/features/persistence.md index 453c2571..5b9a6dad 100644 --- a/docs/features/persistence.md +++ b/docs/features/persistence.md @@ -672,7 +672,7 @@ This module is not lock-coordinated with the shard locks documented in [Concurre - **Data path** (resolution order, `config/resolve.ts:44-49`): `fileConfig.storage.dataPath` → `BUNQUEUE_DATA_PATH` → `BQ_DATA_PATH` → `DATA_PATH` → `SQLITE_PATH`. If unset, `QueueManager` runs with `storage = null` (pure in-memory, no persistence). Programmatic: `new Queue('q', { embedded: true, dataPath: './data/q.db' })`. - **PRAGMAs** (`schema.ts:6-14`, fixed at startup): `journal_mode=WAL`, `synchronous=NORMAL`, `cache_size=-64000` (~64 MB), `temp_store=MEMORY`, `mmap_size=268435456` (256 MB), `page_size=4096`, `busy_timeout=5000`. -- **WriteBuffer**: `writeBufferSize` (default 100 rows → force flush), `writeBufferFlushMs` (effective default 10 ms). Throughput/durability trade-off: buffered ≈ up to 10 ms of loss on process crash; `durable: true` bypasses that application buffer and commits before return at lower throughput. SQLite runs with `synchronous=NORMAL`, so power-loss durability still depends on the host, filesystem, and storage device. +- **WriteBuffer**: `writeBufferSize` (default 100 rows → force flush), `writeBufferFlushMs` (effective default 10 ms). Throughput/durability trade-off: a process crash can lose pending buffered admissions; the typical 10 ms flush interval is not an upper bound during stalls/retries. `durable: true` bypasses that application admission buffer and commits before return at lower throughput. SQLite runs with `synchronous=NORMAL`, so power-loss durability still depends on the host, filesystem, and storage device. - **`onCriticalLoss`** callback for surfacing dropped jobs to ops tooling. - **Completed retention:** `maxCompletedJobs` (default `50_000`) bounds only the hot cache. `completedRetentionMs` defaults to `null`; when configured, each diff --git a/docs/features/rate-limiting-and-concurrency.md b/docs/features/rate-limiting-and-concurrency.md index 3cc2ec66..d358b16f 100644 --- a/docs/features/rate-limiting-and-concurrency.md +++ b/docs/features/rate-limiting-and-concurrency.md @@ -60,7 +60,7 @@ External/runtime: - `deleteQueue(queue): void`, `getQueueNames(): string[]`, `getStateMap(): Map` - `class RateLimiter` (`src/domain/types/queue.ts:35-83`) — token bucket. `constructor(capacity, refillRate = capacity)`, `tryAcquire(): boolean`, `getTokens(): number`, `getTtl(maxJobs?)`. - `class ConcurrencyLimiter` (`src/domain/types/queue.ts:94-130`) — `constructor(limit, activeJobIds?)`, `tryAcquire(jobId): boolean`, `release(jobId): void`, `getActive()` (number of holders), `getLimit()`, `setLimit(limit)`. It keeps a `Set` of slot holders instead of a counter: a job that already holds a slot re-acquires it without taking another, and releasing a job that holds none changes nothing. -- `class ProtocolRateLimiter` (`src/infrastructure/server/rateLimiter.ts:129`) — `constructor(config?: Partial)`, `isAllowed(clientId): boolean`, `getRemaining(clientId): number`, `removeClient(clientId)`, `stop()`. Module helpers `getRateLimiter(config?)` (lazy singleton), `stopRateLimiter()` and `rateLimiterEnvConfig()` (the `RATE_LIMIT_*` defaults, parsed on first use and cached; `createTcpServer`/`createHttpServer` call it first so a malformed value fails startup). Programmatic fields are validated in the constructor (`windowMs` finite ms >= 1, `maxRequests` whole number >= 1, `cleanupIntervalMs` finite ms >= 1 or `0` to disable the sweep, a test-only opt-out); a missing field takes the env default. +- `class ProtocolRateLimiter` (`src/infrastructure/server/rateLimiter.ts:129`) — `constructor(config?: Partial)`, `isAllowed(clientId): boolean`, `getRemaining(clientId): number`, `removeClient(clientId)`, `stop()`. Module helpers `getRateLimiter(config?)` (lazy singleton), `stopRateLimiter()` and `rateLimiterEnvConfig()` (the `RATE_LIMIT_*` defaults, parsed on first use and cached; `createTcpServer`/`createHttpServer` call it before binding, applying the compatibility rules below). Programmatic fields are validated in the constructor (`windowMs` finite ms >= 1 or `0` to disable rate limiting, `maxRequests` whole number >= 1, `cleanupIntervalMs` finite ms >= 1 or `0` to disable the sweep, a test-only opt-out); a missing field takes the env default. ### Client-side classes @@ -131,7 +131,7 @@ interface QueueState { // src/infrastructure/server/rateLimiter.ts:11 — protocol abuse limiter config interface RateLimiterConfig { - windowMs: number; // default 60000 (RATE_LIMIT_WINDOW_MS), finite ms >= 1 + windowMs: number; // default 60000 (RATE_LIMIT_WINDOW_MS), finite ms >= 1; 0 disables maxRequests: number; // default 10000 (RATE_LIMIT_MAX_REQUESTS), whole number >= 1 cleanupIntervalMs?: number; // default 60000 (RATE_LIMIT_CLEANUP_MS); 0 disables (tests only) } @@ -371,7 +371,7 @@ Key invariant: **if `limiter.groupKey` is set, the `WorkerRateLimiter` is disabl - **`isMaxed` scope:** it reflects the global concurrency limiter, not worker-local concurrency or rate-token availability. With no configured global concurrency it is `false`. - **Memory bounds.** - `SlidingWindowDeque` advances a head pointer for O(1) amortized expiry and compacts the array when `head > 1000` (`rateLimiter.ts:95`); the cleanup interval (`safeInterval`) deletes empty per-client deques every `cleanupIntervalMs` (`rateLimiter.ts:180-205`). The maintenance timer is unreferenced: it continues while TCP/HTTP work keeps the process alive but cannot prevent a stopped broker from exiting. Request-time lazy expiry remains authoritative, so timer scheduling never changes admission correctness. - - The sweep is the only eviction for HTTP clients (TCP clients are also dropped on disconnect), so `RATE_LIMIT_CLEANUP_MS` rejects `0`: before, `0` or a non-number skipped the `if (ms)` guard, no sweep ever ran and the per-client map grew without bound. `-1`, `1e12` (`parseInt`: 1) or a period above 2^31 - 1 used to spin the sweep about every millisecond; a long period is now honoured. A `NaN` window (`RATE_LIMIT_WINDOW_MS=abc`) never expired a timestamp, so a client that reached the limit stayed blocked forever, and `RATE_LIMIT_MAX_REQUESTS=abc` disabled the limit silently. Each of these now fails startup with `Invalid NAME: "value" (expected ...)`. + - The sweep is the only eviction for HTTP clients (TCP clients are also dropped on disconnect). `RATE_LIMIT_CLEANUP_MS` of `0`, negative or unreadable keeps the 60,000 ms default with a warning; long periods are honored. Misread forms such as `1e12` fail startup. An unreadable window (`RATE_LIMIT_WINDOW_MS=abc`) also fails startup because it could prevent expiry forever. An unreadable request limit (`RATE_LIMIT_MAX_REQUESTS=abc`) preserves the compatibility behavior of disabling rate limiting, now with a warning; `0` and negative request limits remain errors. - `WorkerRateLimiter.evictExpired` advances head and compacts when more than half the token array is dead space (`workerRateLimiter.ts:108`). - `GroupConcurrencyLimiter.decrement` deletes a group's map entry once its count hits 0, so idle groups don't accumulate (`groupConcurrency.ts:74`). - **Negative/non-positive limits:** CLI rejects `limit <= 0`; TCP handlers only require a finite number, so `RateLimit limit:0` would create a token bucket that never refills enough to grant a token (effectively blocks the queue). `ConcurrencyLimiter` with limit `0` blocks all pulls. @@ -387,18 +387,21 @@ Key invariant: **if `limiter.groupKey` is set, the `WorkerRateLimiter` is disabl ## Configuration Protocol-level limiter (`rateLimiterEnvConfig()`, `src/infrastructure/server/rateLimiter.ts`). -Values are whole numbers (digits only; an empty value keeps the default). The rules +Values use the shared compatible numeric parser: forms such as `+5000`, `5000.0` +and, for duration fields, `5000ms` retain their parseInt reading, while misread exponent/unit forms are +rejected (see [Configuration](./configuration.md#upgrade-compatibility-with-2910)). An empty +value keeps the default. The rules are exported (`RATE_LIMIT_WINDOW_SETTING`, `RATE_LIMIT_MAX_REQUESTS_SETTING`, `RATE_LIMIT_CLEANUP_SETTING`) and the standalone server validates them in -`resolveServerConfig`, so a bad value prints one `Fatal error:` line before storage +`resolveServerConfig`, so a rejected value prints one `Fatal error:` line before storage opens or the banner prints. Embedded and programmatic servers keep the lazy accessor, which applies the same rules when the TCP or HTTP server is created: | Env var | Default | Accepted | Meaning | | ------------------------- | ------- | ---------------------------- | -------------------------------------- | -| `RATE_LIMIT_WINDOW_MS` | 60000 | ms >= 1 | Sliding window size per client | -| `RATE_LIMIT_MAX_REQUESTS` | 10000 | >= 1 | Max raw requests per client per window | -| `RATE_LIMIT_CLEANUP_MS` | 60000 | ms >= 1 (above 2^31 - 1 too) | Interval to evict idle client deques | +| `RATE_LIMIT_WINDOW_MS` | 60000 | ms >= 1; 0 disables; negative → 0 with warning; unreadable is an error | Sliding window size per client | +| `RATE_LIMIT_MAX_REQUESTS` | 10000 | >= 1; unreadable → unlimited with warning; 0/negative are errors | Max raw requests per client per window | +| `RATE_LIMIT_CLEANUP_MS` | 60000 | ms >= 1 (above 2^31 - 1 too); 0/negative/unreadable → default with warning | Interval to evict idle client deques | Per-queue limits are runtime-set (TCP/HTTP/CLI/SDK), persisted in `queue_state`, and have no env defaults (unset = unlimited). diff --git a/docs/features/security-tls-auth.md b/docs/features/security-tls-auth.md index 1226734f..603f3a4c 100644 --- a/docs/features/security-tls-auth.md +++ b/docs/features/security-tls-auth.md @@ -230,7 +230,7 @@ No queue locks are taken in this module. The only per-connection mutable securit - **Token compare is O(validTokens):** every check loops all configured tokens; intended for small token lists. - **Webhook delivery:** fire-and-forget with up to `WEBHOOK_MAX_RETRIES` attempts (default 3), linear backoff `WEBHOOK_RETRY_DELAY_MS * (attempt+1)` (default 1000ms), and a 10s per-request `AbortSignal.timeout` (`webhookManager.ts:142-171`). Signature is computed once and reused across retries. - **Cloud signing:** signs the post-compression body; the receiver must verify over the same compressed bytes. On `401`/`403` the error is logged loudly rather than silently buffered (`httpSender.ts:122-128`). -- **SSRF validator is hostname-string based:** it does not resolve DNS, so a public hostname that resolves to a private IP (DNS rebinding) is not caught; only literal private/loopback/metadata hosts (IPv4 and the IPv6 literal forms listed above) and the listed protocols are blocked. +- **SSRF validator is hostname-string based:** it does not resolve DNS, so a public hostname that resolves to a private IP (DNS rebinding) is not caught; only literal private/loopback/metadata hosts (IPv4 and the IPv6 literal forms listed above) and the listed protocols are blocked. Delivery does not validate redirect destinations as an egress policy. Use trusted registrations and network egress controls. ## Configuration diff --git a/docs/features/store-and-forward.md b/docs/features/store-and-forward.md index 34120629..6bd5a680 100644 --- a/docs/features/store-and-forward.md +++ b/docs/features/store-and-forward.md @@ -6,7 +6,7 @@ This module groups two unrelated client-side concerns that share the "compatibility / integration" theme. -1. **Store-and-Forward** (`forwarder.ts`): drains jobs from a local (typically embedded SQLite) queue to a remote bunqueue server over TCP/TLS. Built for the edge/gateway pattern — a device buffers work locally while offline, then ships it to a central broker when the uplink is up. With a surviving SQLite volume, a remote failure leaves the forward in the local retry/backoff/DLQ path. Re-forwards use a deterministic remote `jobId`; deduplication is bounded by the remote custom-id retention window. +1. **Store-and-Forward** (`forwarder.ts`): drains jobs from a local (typically embedded SQLite) queue to a remote bunqueue server over TCP/TLS. Built for the edge/gateway pattern — a device buffers work locally while offline, then ships it to a central broker when the uplink is up. With a surviving SQLite volume, a remote failure leaves the forward in the local retry/backoff/DLQ path. Re-forwards use a deterministic remote `jobId`; suppression depends on the remote backend's live-ownership/cache limits, not a guaranteed time window. 2. **BullMQ v5 compatibility shims** (`bullmqCompat.ts`): a thin adapter exposing BullMQ-named read APIs (`getPrioritized`, `getWaitingChildren`, `getJobDependencies`, …) on the [Client SDK: Queue](./client-queue-sdk.md), so code written against BullMQ can read prioritized jobs, waiting-children, and parent/child dependency views with minimal changes. The asynchronous read paths are authoritative in embedded and TCP modes; synchronous compatibility snapshots retain the explicit transport boundaries described in [Public API Completeness](./public-api-completeness.md). @@ -175,7 +175,15 @@ The Forwarder holds no locks of its own. Concurrency is the `Worker`'s `concurre ## Edge Cases & Failure Modes -- **Idempotent re-forwards (bounded).** Every push carries `jobId = fwd::`. The server dedupes custom job ids, so a re-forward after a crash/retry does not duplicate the job remotely (`forwarder.ts:11-15`, `forwarder.ts:97`). **Invariant/gotcha:** dedupe only holds *while the remote job is live* (waiting, delayed or active). The server releases a custom id when the job completes (`src/application/operations/ack/completion.ts:43`) or moves to the DLQ (`src/application/operations/ack/failure.ts:29`), and the bounded `customIdMap` LRU can evict it earlier (see [Deduplication & Unique Jobs](./deduplication-and-unique.md)). A re-forward of the same `localId` after that creates a duplicate remote job. Local job-id reuse across DB resets would likewise collide. +- **Re-forward suppression is ownership-bounded.** Every push carries + `jobId = fwd::`. A memory/SQLite destination returns the + live generation only while its bounded ownership cache retains the ID + (50,000 entries by default); after eviction, preserving that generation is + not guaranteed. PostgreSQL checks locked database identity independently. + Completion/DLQ permits a new generation in either backend. Keep external + effects idempotent and reconcile uncertain forwards; see + [Deduplication & Unique Jobs](./deduplication-and-unique.md). Local job-ID + reuse across database resets must also be considered. - **Listener-error isolation.** A throwing `forwarded` listener is caught and ignored so it cannot fail an already-succeeded forward (`forwarder.ts:103-107`). - **`error` event suppression.** Worker `failed`/`error` are only re-emitted as `Forwarder` `error` when `listenerCount('error') > 0` (`forwarder.ts:123,126`). This prevents `EventEmitter`'s default "throw on unhandled error" from crashing the process on a transient uplink failure — but it also means errors are silently dropped if you never attach an `error` listener. Failed forwards remain local jobs; they survive process restart only when the source uses a surviving SQLite volume, subject to the normal buffered/durable write boundary and retry/DLQ retention. - **Auto-batch disabled on the remote.** The remote queue forces `autoBatch:{ enabled:false }` (`forwarder.ts:88`); forwards are individual pushes, not coalesced. diff --git a/docs/features/tcp-protocol.md b/docs/features/tcp-protocol.md index dcffce9f..4a1da7e9 100644 --- a/docs/features/tcp-protocol.md +++ b/docs/features/tcp-protocol.md @@ -156,14 +156,17 @@ selects descending createdAt/job-id order before offset/limit pagination. group getter/control commands expose server-authoritative depth, active count, fixed-window TTL/configuration, concurrency, pause state, and immediate manual deadlines. Grouped `PUSH`/`PUSHB`/`PUSHF` inputs carry normalized `groupId`, -ascending BullMQ Pro priority, and optional atomic `groupMaxSize`; a failed -capacity check rejects the complete batch/graph through the ordinary error -response. +ascending BullMQ Pro priority, and optional per-job atomic `groupMaxSize`. +Memory/SQLite `PUSHB` retains an accepted prefix if a later member fails its +capacity check, despite an error response; preflight validation rejects before +mutation. PostgreSQL batch insertion is transactional. `PUSHF` rejects the +complete graph on a capacity error. `PUSHF` carries `{ cmd: 'PUSHF', jobs: AtomicFlowJobInput[] }`, where every entry already has its final string `id`, queue, job input, and graph links. The broker validates strict runtime types, symmetric parent/child metadata, -acyclicity, 10,000 jobs, 10 MB per job and 64 MB aggregate flow data before +acyclicity, 10,000 jobs, 10 × 1024 × 1024 JSON string characters per job and +64 × 1024 × 1024 characters across the graph before mutation. Success is `{ ok: true, data: { jobs: Job[] } }`; the snapshots cover exactly the requested IDs. With SQLite configured, all rows commit in one immediate transaction before the graph is visible to workers. See @@ -200,7 +203,12 @@ outside the `Response` union because it does not answer a command. ## Business Logic / Control Flow -**Server creation (`createTcpServer`, `tcp.ts:25-35`)**: before anything binds, resolve `idleTimeoutMs` and `maxWriteQueueBytes` (an explicit config value is validated; otherwise `tcpIdleTimeoutMs()` / `tcpMaxWriteQueueBytes()` parse `TCP_IDLE_TIMEOUT_MS` / `TCP_MAX_WRITE_QUEUE_BYTES`) and read `rateLimiterEnvConfig()` (`RATE_LIMIT_*`). A malformed value throws (`Invalid TCP_IDLE_TIMEOUT_MS: "1e12" (expected a whole number of milliseconds >= 0)`, or a `RangeError` naming `TcpServerConfig.idleTimeoutMs`), so the server fails at startup instead of on the first connection or request. The env rules are exported (`TCP_IDLE_TIMEOUT_SETTING`, `TCP_MAX_WRITE_QUEUE_SETTING` in `tcp/constants.ts`) and the standalone server validates them earlier, in `resolveServerConfig`, before storage opens or the banner prints. The env values are strict: digits only, an empty value keeps the default, and `1e12`, `-1`, `abc` or `64MB` are rejected (`parseInt` used to read them as 1 ms, "disabled" or a 64-byte cap). +The `validateJobData` cap described above as 10 MB counts JSON string length: +10 × 1024 × 1024 characters, not bytes. `Update` intentionally has no job-data +size cap; applications must enforce their own payload limits. The 64 MiB frame +limit is a separate transport bound. + +**Server creation (`createTcpServer`, `tcp.ts:25-35`)**: before anything binds, resolve `idleTimeoutMs` and `maxWriteQueueBytes` (an explicit config value is validated; otherwise `tcpIdleTimeoutMs()` / `tcpMaxWriteQueueBytes()` parse `TCP_IDLE_TIMEOUT_MS` / `TCP_MAX_WRITE_QUEUE_BYTES`) and read `rateLimiterEnvConfig()` (`RATE_LIMIT_*`). A rejected value throws (`Invalid TCP_IDLE_TIMEOUT_MS: "1e12" (expected a whole number of milliseconds >= 0)`, or a `RangeError` naming `TcpServerConfig.idleTimeoutMs`), so the server fails at startup instead of on the first connection or request. The env rules are exported (`TCP_IDLE_TIMEOUT_SETTING`, `TCP_MAX_WRITE_QUEUE_SETTING` in `tcp/constants.ts`) and the standalone server validates them earlier, in `resolveServerConfig`, before storage opens or the banner prints. Empty env values keep defaults; compatible parseInt forms such as `5000ms` work for milliseconds. Negative or unreadable idle/write-queue values disable that setting with a warning; misread forms such as `1e12` or `64MB` for a byte count fail. See [Configuration](./configuration.md#upgrade-compatibility-with-2910) and [Rate Limiting & Concurrency](./rate-limiting-and-concurrency.md#configuration) for each fallback. **Connection open (`TcpConnectionRegistry.init`, `tcp/connections.ts:31-59`)**: allocate `clientId = uuid()`, build `HandlerContext` with `authenticated = authTokens.size === 0` (auto-auth when no tokens configured) and `signal = AbortSignal.any([connection signal, server drain signal])`, and attach per-connection `TcpConnectionData`: a fresh `FrameParser`, `Semaphore(50)`, `SocketWriteQueue(maxWriteQueueBytes)`, abort controller, null `stallTimer`, and null event queue. Register in `connections` map and emit `client:connected`. `init` is idempotent (returns early if `socket.data` exists) and is called from **both** the `open` handler and lazily at the top of the `data` handler (`tcp.ts:38-44`): under native TLS, Bun can deliver `data` before `open` has run, and the handler previously destructured a null `socket.data`, escalating a TypeError to the process-level unhandledRejection handler (a pre-auth remote DoS on an exposed TLS port, #108). Lazy init preserves that first frame instead of dropping it. @@ -336,7 +344,17 @@ connections (`closeAll`) and stops the listener; see the shutdown order in constraint, relation, driver, host, SQLite, and network diagnostics to `Internal server error` before framing. Non-throwing storage-status payloads apply the same projection; SQLite disk-full retains its actionable detail. -- **Job release on disconnect with retry:** `releaseClientJobsWithRetry` (`tcp/clientRelease.ts:4-23`) retries up to 3× with exponential backoff (100/200/400ms). If all retries fail, `TcpConnectionRegistry.close` falls back to `forceReleaseClientJobs` (`tcp/connections.ts:116-124`), which unconditionally clears client tracking (prevents a `connections`/ownership Map leak) and resets heartbeats so the stall detector recovers any orphaned `active` jobs on its next tick — chosen over leaking the jobs. Both paths act only on deliveries the connection still owns: a job recovered from it (stall, orphan, lock expiry, timeout) is detached at recovery, and a later delivery to another worker is never released or expired by this connection's close (see [Webhooks, Events & Job Logs](./webhooks-and-events.md)). +- **Job release on disconnect with retry:** `releaseClientJobsWithRetry` + (`tcp/clientRelease.ts:4-23`) retries cleanup up to 3 times with exponential + backoff (100/200/400ms), independently of processing attempts. Exhaustion + invokes the backend's `forceReleaseClientJobs`: memory/SQLite clear owned + tracking and expire heartbeat/start state for stall recovery; PostgreSQL + queue another transactional, token-fenced release of the retained lease + session. If release cannot complete, eventual lease expiry remains available. + Immediate release excludes renewed leases and stale delivery owners. Local + eligible `cron:` deliveries are discarded; PostgreSQL uses ordinary release + for dependency-protected cron IDs that cannot safely be discarded. See + [Webhooks, Events & Job Logs](./webhooks-and-events.md). - **Idempotency** is a command-level concern (`jobId`/`uniqueKey`/`dedup`), not a transport one; the transport delivers exactly the frames it parsed. `reqId` is purely for client-side response correlation and is echoed back verbatim, including on errors. - **TLS startup invariant:** `readTlsMaterial` runs _before_ `Bun.listen` binds the port, so a bad cert/key path fails fast at startup instead of leaving a half-started listener. - **Replaceable TLS listener:** with TLS the listener binds with `reusePort: true` (after a plain exclusive probe bind on a configured port other than `0`, so a busy port still fails with `Failed to listen at ` and a second broker cannot share it). `prepareTlsReload(material)` binds a second listener on the bound port with the new pair and the **same** `socketHandlers` object, so every connection, accepted by either listener, lives in the one `TcpConnectionRegistry`; `commit` points `server` at it and stops the old one with `stop(false)` (its connections stay open), `abort` stops the new one. `beginDrain()` and `stop()` refuse any later prepare. A plain listener binds exactly as before (no probe, no SO_REUSEPORT) and has no reload. diff --git a/docs/features/tcp-server-handlers.md b/docs/features/tcp-server-handlers.md index dacf0d8a..1fb172b6 100644 --- a/docs/features/tcp-server-handlers.md +++ b/docs/features/tcp-server-handlers.md @@ -139,9 +139,14 @@ listener when idle, and clears it during server shutdown. Auth (`handleAuth`, `handler.ts:30`): iterate configured tokens, comparing with `constantTimeEqual` (timing-safe). On match, set `ctx.authenticated = true` and return `ok()`; on no match, emit `auth:failed` and return `error('Invalid token')`. +The handler admission cap sometimes labeled 10 MB below counts +10 × 1024 × 1024 JSON string characters, not bytes. `Update` has no job-data +size cap. Atomic flows also count JSON characters for their aggregate bound; +the TCP frame limit remains a separate 64 MiB byte limit. + Core paths: -- `handlePush` (`core.ts`): validates queue name, data size (≤10MB), and numeric +- `handlePush` (`core.ts`): validates queue name, JSON data length (≤10 × 1024 × 1024 string characters), and numeric option bounds. Memory/SQLite validates each `dependsOn` id against `jobIndex`, `completedJobs`, or `depCompletions`. PostgreSQL first queries the authoritative namespace so a remote parent is accepted even while the local @@ -156,12 +161,15 @@ Core paths: different ID rolls the entire batch back. SQLite keeps its synchronous local validation. On validation failure, the handler returns an indexed `jobs[i]: ...` error when available; on success it returns `resp.batch(ids)`. + Memory/SQLite runtime admission errors (such as a later group-cap rejection) + retain the accepted prefix; they do not roll the batch back. - `handlePushFlow` (`flow.ts`): passes the fully resolved multi-queue graph to `QueueManager.pushFlow`. The application validator checks the complete batch before mutation, including strict string/array/boolean wire types, internal parent metadata, duplicate/missing/asymmetric edges, cycles, mutually - exclusive failure policies, 10,000 jobs, 10 MB per job and 64 MB total flow - data. Success uses `resp.data({ jobs })`; any validation, ownership, or + exclusive failure policies, 10,000 jobs, 10 × 1024 × 1024 JSON string characters + per job and 64 × 1024 × 1024 characters across the graph. Success uses + `resp.data({ jobs })`; any validation, ownership, or persistence error rejects the whole command. - `handlePull` (`src/infrastructure/server/handlers/core.ts`): caps `timeout` to `[0, 60000]`, validates optional group defaults, and forwards them to the @@ -296,7 +304,16 @@ Concurrency relevant to this layer: - The transport processes frames of a single connection in parallel under a per-connection `Semaphore(50)` (`tcp/constants.ts:1`, constructed at `tcp/connections.ts:42` and acquired at `src/infrastructure/server/tcp.ts:85`), so multiple `handleCommand` calls can be in flight on the same `ctx`. Handlers are therefore expected to be safe against concurrent invocation — they are, because they hold no per-call state and `QueueManager` serializes the underlying mutations. - `ctx.authenticated` is the single shared mutable field; once `handleAuth` flips it to `true` it is monotonic for the connection's lifetime, so the concurrent reads in the auth gate are benign. -- Client-job ownership: `registerClientJob` (on pull) / `unregisterClientJob` (on applied ack/fail) keep a per-`clientId` set so that on disconnect the TCP `close` handler can call `releaseClientJobs` (with retry + force-release fallback) to requeue leased jobs. Ignored retired generations do not clear tracking for a possibly newer lease with the same ID. Ownership is per delivery: registering a delivery detaches its previous owner, every recovery path detaches the silent owner, and disconnect release acts only on a delivery the connection still owns (see [Webhooks, Events & Job Logs](./webhooks-and-events.md)). See [Worker Registry & Management](./workers-management.md). +- Client-job ownership: `registerClientJob` (on pull) / `unregisterClientJob` + (on applied ack/fail) keep a per-`clientId` set for disconnect cleanup. TCP + retries cleanup with a backend-specific force-release fallback: local engines + expire heartbeat/start state, whereas PostgreSQL queues transactional, + token-fenced release of the retained session. Release only touches eligible + current deliveries and excludes renewed leases; local protected cron jobs + are discarded, while PostgreSQL dependency-protected cron IDs use ordinary + release. Ignored retired generations do not clear a newer lease's tracking. + See [Webhooks, Events & Job Logs](./webhooks-and-events.md) and + [Worker Registry & Management](./workers-management.md). - Lock tokens (`token`) are an opaque ownership credential minted by `pullWithLock`/`pullBatchWithLock` and verified by `ack`/`fail`/`extendLock`/`renewJobLock`; the handlers only pass them through. ## Edge Cases & Failure Modes @@ -352,7 +369,7 @@ auth, telemetry, timeout, SQLite path, TLS, Cloud, and S3 fields plus [PostgreSQL 16–18 Multi-Broker Persistence](./postgres-multibroker.md), and [Security: TLS, Auth, CORS](./security-tls-auth.md). -Input-validation limits enforced by the handlers (from `protocol.ts`): queue name ≤256 chars and `^[a-zA-Z0-9_\-.:]+$`; job data ≤10MB; `PULL`/`PULLB` timeout `[0,60000]`; `PULLB` count `[1,1000]`; `WaitJob` timeout `[0,600000]`. Job options go through the shared `validateJobOptions` (the same validator and rules as atomic flows, embedded `Queue.add`/`addBulk` and cron templates), which refuses only what no job can run with and admits every value 2.9.10 accepted in either mode ([Job Options Validation](./job-options-validation.md)): a non-number, `NaN` or an infinity is refused (a plain numeric string counts as its number); `priority` is any finite number (grouped: `group.priority` integer `[0,2097151]`); a negative job `delay` keeps its past run time (ready at once, ahead of later ready jobs, as on 2.9.10); `timeout` and `ttl` cannot be negative; `maxAttempts` of 1 or less runs once (`Infinity` is stored as 2,147,483,647); `stallTimeout`, `stackTraceLimit`, `keepLogs`, `sizeLimit`, `dedup.ttl` and `debounceTtl` only need to be finite; `timestamp` within ±4.32e15; `groupMaxSize` a positive safe integer; `repeat.every` positive. Durations above `MAX_JOB_DURATION_MS` (4.32e15 ms) are clamped by `normalizeJobInput`, which also stores the normalized values. `PUSH` passes the whole command. `backoff` accepts either a number (ms) or the object form `{ type, delay?, maxDelay? }` (`validateBackoffField`) — any `type` (`fixed` is fixed, anything else runs as exponential), a missing `delay` is the 1000 ms default, a given `delay` must be a finite number `>= 0`, and an optional `maxDelay` (per-job retry-delay cap; `null` = absent) must be a finite number in `[0, 1day]` (`MAX_BACKOFF_DELAY`); `PUSH`, `PUSHB` (per job, via `validatePushBatchJobs`, error `jobs[i]: ...`) and `PUSHF` validate the applicable rules. `PUSHF` additionally caps the full graph as described above. `ChangeDelay`/`MoveToDelayed` require a finite `delay` (`validateDelayArgument`; the engine keeps a negative one as a past run time, `now + delay`); `PULL`/`PULLB` `lockTtl`, `ExtendLock` `duration`, `ExtendLocks` `durations[i]` and a non-zero `JobHeartbeat` `duration` must be finite numbers, any sign, passed through as on 2.9.10 (`validateLockDuration`; `NaN`, an infinity or a string is refused before any job is claimed or lease changed). `ChangePriority` takes any finite priority (missing = 0) and normalizes `lifo` to a boolean; `Progress` is never refused (`normalizeProgress`); `Update` needs only JSON-serializable data (no size limit); `ClearLogs` applies `keepLogsArgument`. `Cron` template options, spawned `priority`/`dedup.ttl` and `repeatEvery` (at most `MAX_JOB_DURATION_MS`) are validated by `assertValidCronInput` with the same rules, and the thrown message is returned as the protocol error. +Input-validation limits enforced by the handlers (from `protocol.ts`): queue name ≤256 chars and `^[a-zA-Z0-9_\-.:]+$`; job data ≤10MB; `PULL`/`PULLB` timeout `[0,60000]`; `PULLB` count `[1,1000]`; `WaitJob` timeout `[0,600000]`. Job options go through the shared `validateJobOptions` (the same validator and rules as atomic flows, embedded `Queue.add`/`addBulk` and cron templates), which refuses only what no job can run with and admits every value 2.9.10 accepted in either mode ([Job Options Validation](./job-options-validation.md)): a non-number, `NaN` or an infinity is refused (a plain numeric string counts as its number); `priority` is any finite number (grouped: `group.priority` integer `[0,2097151]`); a negative job `delay` keeps its past run time (ready at once; among equal-priority FIFO jobs, earlier run times go first, as on 2.9.10; priority and LIFO ordering still apply); `timeout` and `ttl` cannot be negative; `maxAttempts` of 1 or less runs once (`Infinity` is stored as 2,147,483,647); `stallTimeout`, `stackTraceLimit`, `keepLogs`, `sizeLimit`, `dedup.ttl` and `debounceTtl` only need to be finite; `timestamp` within ±4.32e15; `groupMaxSize` a positive safe integer; `repeat.every` positive. Durations above `MAX_JOB_DURATION_MS` (4.32e15 ms) are clamped by `normalizeJobInput`, which also stores the normalized values. `PUSH` passes the whole command. `backoff` accepts either a number (ms) or the object form `{ type, delay?, maxDelay? }` (`validateBackoffField`) — any `type` (`fixed` is fixed, anything else runs as exponential), a missing `delay` is the 1000 ms default, a given `delay` must be a finite number `>= 0`, and an optional `maxDelay` (per-job retry-delay cap; `null` = absent) must be a finite number in `[0, 1day]` (`MAX_BACKOFF_DELAY`); `PUSH`, `PUSHB` (per job, via `validatePushBatchJobs`, error `jobs[i]: ...`) and `PUSHF` validate the applicable rules. `PUSHF` additionally caps the full graph as described above. `ChangeDelay`/`MoveToDelayed` require a finite `delay` (`validateDelayArgument`; the engine keeps a negative one as a past run time, `now + delay`); `PULL`/`PULLB` `lockTtl`, `ExtendLock` `duration`, `ExtendLocks` `durations[i]` and a non-zero `JobHeartbeat` `duration` must be finite numbers, any sign, passed through as on 2.9.10 (`validateLockDuration`; `NaN`, an infinity or a string is refused before any job is claimed or lease changed). `ChangePriority` takes any finite priority (missing = 0) and normalizes `lifo` to a boolean; `Progress` is never refused (`normalizeProgress`); `Update` needs only JSON-serializable data (no size limit); `ClearLogs` applies `keepLogsArgument`. `Cron` template options, spawned `priority`/`dedup.ttl` and `repeatEvery` (at most `MAX_JOB_DURATION_MS`) are validated by `assertValidCronInput` with the same rules, and the thrown message is returned as the protocol error. ## Related Docs diff --git a/docs/features/webhooks-and-events.md b/docs/features/webhooks-and-events.md index 01710f90..9ff3dd61 100644 --- a/docs/features/webhooks-and-events.md +++ b/docs/features/webhooks-and-events.md @@ -34,7 +34,15 @@ metrics. engine. PostgreSQL mode persists job logs and lifecycle events, but webhook definitions/delivery remain process-local. See [Persistence](./persistence.md) and [PostgreSQL 16–18 Multi-Broker Persistence](./postgres-multibroker.md). -- Stall recovery — `clientTracking` only _triggers_ recovery (resets heartbeats); the stall detector in [Background Tasks](./background-tasks.md) reclaims orphaned jobs. +- Memory/SQLite disconnect release — `clientTracking` immediately releases eligible deliveries + still owned by the disconnecting client under shard/processing locks, persisting + the uncharged requeue. Renewed leases stay active to protect pooled workers; + expiry/stall recovery handles them if heartbeats stop. Only the force-release + fallback after cleanup failures expires heartbeat/start state. Stall + classification belongs to [Background Tasks](./background-tasks.md). PostgreSQL + uses transactional, token-fenced release; its force-release fallback queues + another attempt for the retained lease session, with eventual lease expiry + available if release cannot complete. ## Dependencies @@ -277,6 +285,9 @@ for the moves without a job event (see [Client SDK: Queue](./client-queue-sdk.md ### Client tracking & disconnect release +The `clientTracking` phases below describe memory/SQLite; the PostgreSQL +transactional/session implementation is described in Concurrency & Locking. + - `registerClientJob` (on PULL) and `unregisterClientJob` (on ACK/FAIL) maintain `clientJobs: Map>`, deleting the set when empty, together with its reverse index `clientJobOwners: Map` (`clientOwnership.ts`). Every mutation of the two maps goes through that module, so a job is in `clientJobs.get(c)` exactly when its owner record names `c`, and every operation is O(1) per job. - **Delivery identity.** The owner record stores the `jobIndex` entry current at registration. Only a pull installs a `{ type: 'processing' }` entry and every pull installs a fresh object, so the record identifies one delivery: a later delivery of the same job id never matches it. `ownsCurrentDelivery(clientId, jobId)` is true only when the record names the client and the job's current `jobIndex` entry is that same processing object. - **One owner per delivery.** `registerClientJob` detaches any previous owner of the job before recording the new one. `unregisterClientJob` removes only the caller's own claim, so a late outcome handled on one connection cannot detach a newer delivery owned by another. @@ -290,17 +301,26 @@ for the moves without a job event (see [Client SDK: Queue](./client-queue-sdk.md ## Concurrency & Locking +The following lock/heartbeat mechanics describe the memory/SQLite +`clientTracking` implementation; PostgreSQL's transaction/session path follows +below. + - `releaseClientJobs` follows the project lock hierarchy: **shardLocks before processingLocks** (`clientTracking.ts`). Reads (phase 1) are lock-free; the release decision is repeated and the mutations happen under both locks. - The `renewalCount > 0` guard prevents a **double-execute** race: with a pooled client, heartbeats travel on a different connection than the one that pulled, so the pulling socket closing does not mean the worker died — such jobs are left for lock-expiry/stall detection to reclaim (`isReleasable`, `clientTracking.ts`). - `forceReleaseClientJobs` is intentionally lock-free and acts only on deliveries the client still owns (`ownsCurrentDelivery`): for each it drops `jobLocks[jobId]` (no stale token survives the disconnect) and sets `lastHeartbeat = 0` and `startedAt = 0` so the stall detector's grace gate passes on its next eligible tick. A job recovered from the client and delivered again keeps the new owner's lock and timers. It accepts that a concurrent stall/lock-expiry path may mutate the same job — worst case the write lands on an object no longer in the map (wasted, never corrupting) (`clientTracking.ts`). - `releaseClientJobs` drops the client (`dropClient`: its `clientJobs` set and only its own owner records) in a `finally` block even on mid-flight lock failure, preventing an unbounded leak across disconnects that hit lock timeouts (`clientTracking.ts`). -The PostgreSQL adapter keeps the same connection-facing contract with durable -lease ownership. Disconnect release row-locks the exact generation and only -requeues a still-live, never-renewed token. A heartbeat received by another -broker increments `lease_renewals` and transfers `lease_broker_id`, so loss of -the original pull socket cannot double-deliver the job. Broker shutdown releases -its remaining untransferred leases; expired protected cron leases are discarded. +The PostgreSQL adapter uses durable lease ownership. Disconnect release +row-locks the exact generation and only releases a still-live, never-renewed +token. Protected cron deliveries are discarded when destruction is safe; +dependency-protected IDs instead follow ordinary transactional release. A +heartbeat received by another broker increments `lease_renewals` and transfers +`lease_broker_id`, so loss of the original pull socket cannot double-deliver the +job. The PostgreSQL force-release override queues `runClientRelease` for the +retained session, rather than expiring local heartbeat/start timestamps; +repeated attempts remain token-fenced. If they cannot complete, lease expiry +still provides recovery. Broker shutdown releases its remaining untransferred +leases, using the same dependency-protected cron discard check. - `EventsManager`/`WebhookManager` hold no locks; `broadcast` and `broadcastBatch` are synchronous and webhook delivery is async @@ -308,6 +328,9 @@ its remaining untransferred leases; expired protected cron leases are discarded. ## Edge Cases & Failure Modes +- **Egress boundary:** registration checks URL hostname text, without DNS + resolution; delivery does not validate redirect destinations as an egress + policy. Use trusted webhook registrations and network egress controls. - **SSRF protection:** `validateWebhookUrl` (on by default; disabled via `validateUrls: false`) rejects non-http(s) schemes, URLs > 2048 chars, localhost variants, private IPv4 (`10.*`, `172.16–31.*`, `192.168.*`), link-local `169.254.*`, `0.*`, `127.*`, **IPv4-mapped / IPv4-compatible IPv6** whose embedded IPv4 is loopback/private (`extractMappedIpv4` unwraps the dotted `::ffff:127.0.0.1`, the URL-parser-normalized hex form `[::ffff:7f00:1]`, **and** the deprecated `::`-prefixed compatible form `[::127.0.0.1]`/`[::7f00:1]` before the octet check), **IPv6 ULA `fc00::/7` and link-local `fe80::/10`** plus the unspecified `::` (`checkBlockedIpv6`), and cloud-metadata hosts (`169.254.169.254`, `metadata.google.internal`, `*.internal`) (`webhookValidation.ts:42`). - **Dead-event rejection:** `AddWebhook` rejects events not in `WEBHOOK_EVENTS`, so a webhook can't be created against an event that would silently never fire (`handlers/monitoring/webhooks.ts:13-29`). - **Delivery is best-effort / fire-and-forget:** failures are logged and counted @@ -323,7 +346,10 @@ its remaining untransferred leases; expired protected cron leases are discarded. are caught and ignored; a failed telemetry batch performs isolated scalar retries before returning. - **Job-log bounds:** per-job cap is `maxLogsPerJob = 100` (`queue-manager/state.ts`); the `jobLogs` LRU itself holds at most `maxJobLogs = 10_000` distinct jobs (`application/types/config.ts`), evicting whole-job entries. Adding logs to a job not in `jobIndex` returns `false`. -- **Cron `preventOverlap` invariant:** disconnect release must discard (not requeue) `cron:*` jobs, or they re-run immediately on reconnect (#73). +- **Cron `preventOverlap`:** memory/SQLite disconnect release discards eligible + `cron:*` deliveries rather than requeueing them (#73). PostgreSQL performs the + same discard only when destruction is safe; dependency-protected IDs follow + ordinary transactional release instead. ## Configuration diff --git a/docs/protocol.md b/docs/protocol.md index 0c7d415d..b09c59c4 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -181,8 +181,13 @@ Client MUSTs: `PUSHB` entries are typed `JobInput`, whose field is **`customId`**. A bulk client MUST rename `jobId → customId` per entry or custom ids are silently lost. -- `jobId` is idempotent: re-pushing an existing custom id returns the - existing job's `id` instead of enqueuing a duplicate. +- `jobId` returns the unfinished generation while ownership is retained. + Ordinary memory/SQLite admission depends on a bounded ownership LRU (50,000 + IDs by default); after eviction, preserving the original live generation is + not guaranteed. SQLite uses upserts, so its primary key does not add that + guard, and `durable` does not remove the limit. PostgreSQL uses locked live-row + checks instead. After completion or DLQ, an ID can create a fresh generation. + See [Custom-ID limits](https://bunqueue.dev/guide/queue/deduplication/#custom-id-limits). - The broker deduplicates on `uniqueKey` alone; `dedup` is only the `ttl`/`extend`/`replace` policy and is ignored without a `uniqueKey`. A client `deduplication {id, ttl?, extend?, replace?}` option MUST send `id` @@ -195,8 +200,13 @@ Client MUSTs: an integer from 0 through 2,097,151, served in ascending order (`0` first). Without `groupId`, the ordinary queue priority keeps its higher-first range. - `groupMaxSize` is a positive safe integer. The broker checks pending depth - and admission atomically; if any job in a `PUSHB` or `PUSHF` graph would - exceed the limit, it rejects the complete command without partial writes. + and admission atomically per job. Memory/SQLite `PUSHB` retains an already + accepted prefix on runtime rejection (for example a later member exceeding + its group cap), despite returning an error; preflight validation rejects + before mutation. Reconcile accepted jobs before retrying; stable custom IDs + alone are insufficient after memory/SQLite ownership eviction. Keep external + effects idempotent. PostgreSQL batch insertion is transactional. `PUSHF` rejects the + complete graph without partial writes if any member exceeds its cap. ### 6.2 Query @@ -436,8 +446,10 @@ entries. ### 6.7 Webhooks, rate limits, monitoring -- `AddWebhook` (`url`, `events`) → **`{data: {webhookId}}`**. Localhost and - private-network URLs are rejected (SSRF guard). Event names look like +- `AddWebhook` (`url`, `events`) → **`{data: {webhookId}}`**. Prohibited literal + hostnames/IPs are rejected at registration; DNS resolution and redirect + destinations are not checked as an egress policy. Use trusted registrations + and network egress controls. Event names look like `job.completed`. - `RemoveWebhook` takes **`webhookId`**; `SetWebhookEnabled` takes **`id`**. Yes, they differ; both are normative. diff --git a/docs/src/content/docs/api/http.md b/docs/src/content/docs/api/http.md index d2a11092..96b42468 100644 --- a/docs/src/content/docs/api/http.md +++ b/docs/src/content/docs/api/http.md @@ -114,7 +114,7 @@ Some endpoints, such as `DELETE /jobs/:id`, queue control, DLQ, rate-limit/concu - **Queue names**: 1-256 characters, alphanumeric + `-_.:` - **Numeric fields**: Validated by the same validator as the TCP commands, embedded `Queue.add` and flows, which rejects only what no job can run with: a non-number, `NaN`, an infinity in any field except `maxAttempts` (where an infinity means the maximum attempt count), a negative `timeout`, `ttl`, `backoff` or `backoff.delay` (a missing `backoff.delay` is the 1000 ms default), and, with `groupId`, a `priority` that is not an integer from 0 to 2,097,151. A negative `delay` makes the job ready at once with a past run time (as in 2.9.10), a move-to-delayed/delay `delay` is required, and a lock `duration` must be a finite number. Every value 2.9.10 accepted keeps its 2.9.10 result. -- **Job data**: Max 10MB per pushed job payload (updating an existing job's data has no size limit, as in 2.9.10) +- **Job data**: Max 10 × 1024 × 1024 JSON string characters per pushed job payload (a character count, not a byte limit; updating an existing job's data has no size limit, as in 2.9.10). - **Job IDs**: UUID v7 format (auto-generated) or custom string (via `jobId` field) --- @@ -230,7 +230,7 @@ curl -X POST http://localhost:6790/queues/emails/jobs \ | Field | Type | Default | Description | | ------------------ | -------------------- | ------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `data` | `any` | - | Job payload. Any JSON-serializable value. Max 10MB. A push without `data` is accepted and creates a job with no payload. | +| `data` | `any` | - | Job payload. Any JSON-serializable value. Max 10 × 1024 × 1024 JSON string characters. A push without `data` is accepted and creates a job with no payload. | | `name` | `string` | `"default"` | Job name, stored as the top-level `job.name`. | | `priority` | `number` | `0` | Higher value = processed sooner. Any finite number; with `groupId` it is the group priority and must be an integer from 0 to 2,097,151. | | `delay` | `number` | `0` | Milliseconds before the job becomes available for processing. A negative delay makes it ready at once (past run time). | @@ -239,13 +239,13 @@ curl -X POST http://localhost:6790/queues/emails/jobs \ | `ttl` | `number` | - | Time-to-live from creation in milliseconds (0 or more). Job is discarded if not processed within this window. | | `timeout` | `number` | - | Processing timeout in milliseconds (0 or more; 0 means none). The broker fails the active attempt at its absolute deadline; a later outcome from that lease generation is ignored. | | `uniqueKey` | `string` | - | Deduplication key. If a job with the same `uniqueKey` already exists in the queue, no new job is created: the response is `200` with the existing job's `id`. | -| `jobId` | `string` | - | Broker-wide custom job ID. If a live job with this ID already exists in any queue, the push is idempotent and returns the existing ID. | +| `jobId` | `string` | - | Custom ID, broker-wide (PostgreSQL: namespace-wide). Returns the live generation while ownership is retained; memory/SQLite use a bounded cache, whereas PostgreSQL checks the database. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). | | `tags` | `string[]` | `[]` | Metadata tags for filtering and querying. | | `groupId` | `string` | - | Job-group identifier. Claims use ascending group priority with FIFO ties and round-robin across groups; execution is concurrent unless a Worker supplies a group concurrency cap. | | `lifo` | `boolean` | `false` | Last-in-first-out ordering. When true, the job is processed before other jobs at the same priority. | | `removeOnComplete` | `boolean` | `false` | Automatically remove the job from memory after completion. Saves memory for fire-and-forget jobs. | | `removeOnFail` | `boolean` | `false` | Automatically remove the job after final failure (after all retries exhausted). | -| `durable` | `boolean` | `false` | SQLite: bypass the write buffer and commit before returning (slower, but no 10 ms buffer-loss window). PostgreSQL admissions are already transactional and do not use the SQLite buffer. | +| `durable` | `boolean` | `false` | SQLite: bypass the application admission buffer and commit before returning. PostgreSQL admissions are already transactional and do not use the SQLite buffer. | | `dependsOn` | `string[]` | `[]` | Job IDs that must complete before this job becomes available. The job enters `waiting-children` state until all dependencies are met. | | `repeat` | `object` | - | Repeat configuration: `{ every: ms, limit: n }` for interval-based, or `{ pattern: "cron expression" }` for cron-based (optional `tz`, `startDate`, `endDate`, `immediately`). | @@ -262,9 +262,13 @@ each job object whole. { "ok": true, "id": "019ce9d7-6983-7000-946f-48737be2b0f9" } ``` -The `id` is a UUID v7 (time-ordered, sortable). If `jobId` was provided and a -live job with that broker-wide ID already exists, including in another queue, -the existing job's ID is returned (idempotent). +Generated `id` values are UUID v7 (time-ordered, sortable). A custom `jobId` +returns the live generation while ownership is retained, including across +queues. Memory/SQLite use a bounded cache (50,000 IDs by default); eviction can +end this protection before completion, and SQLite upserts/durable admission do +not repair it. PostgreSQL checks live database identity. See +[Custom-ID limits](/guide/queue/deduplication/#custom-id-limits); keep external +effects idempotent and reconcile uncertain admissions before retrying. **Error responses:** @@ -273,7 +277,7 @@ the existing job's ID is returned (idempotent). | `400` | `Invalid JSON body` | Request body is not valid JSON | | `404` | `Not found` | Empty queue name (no route matches) | | `400` | `Queue name contains invalid characters` | Queue name has chars outside `a-zA-Z0-9_-.:` | -| `400` | `Job data too large (max 10MB)` | Serialized data exceeds 10MB | +| `400` | `Job data too large (max 10MB)` | JSON data exceeds 10 × 1024 × 1024 string characters | | `400` | `priority must be a number` | Non-numeric priority (`"high"`) | | `400` | `delay must be a number` | Non-numeric delay (a negative one runs at once) | | `400` | `timeout must be at least 0` | Negative processing timeout | @@ -301,7 +305,7 @@ curl -X POST http://localhost:6790/queues/emails/jobs/bulk \ }' ``` -Each item in `jobs` supports the same fields as a single push and is validated with the same rules (option bounds and `dependsOn` existence; a `dependsOn` entry may also reference the `customId` of any job in the same batch, in any order). Two differences: the custom ID field is `customId` (not `jobId`), and the `attempts` alias is not accepted (use `maxAttempts`). Validation is all-or-nothing: if any job fails validation, the whole batch is rejected with an error naming the offending index (`jobs[i]: ...`) and nothing is pushed. A runtime admission error after validation, such as a full group (`group.maxSize`) or an unresolved custom-ID dependency, can leave the jobs accepted before it pushed, so check which IDs exist before you resubmit. +Each item in `jobs` supports the same fields as a single push and is validated with the same rules (option bounds and `dependsOn` existence; a `dependsOn` entry may also reference the `customId` of any job in the same batch, in any order). Two differences: the custom ID field is `customId` (not `jobId`), and the `attempts` alias is not accepted (use `maxAttempts`). Validation is all-or-nothing: if any job fails validation, the whole batch is rejected with an error naming the offending index (`jobs[i]: ...`) and nothing is pushed. On memory/SQLite, a runtime admission error after validation, such as a full group (`group.maxSize`) or an unresolved custom-ID dependency, can preserve the accepted prefix. Reconcile accepted work and outcomes before resubmitting; stable custom IDs alone do not make retries safe after ownership-cache eviction. Keep external effects idempotent. PostgreSQL batch insertion is transactional. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). **Response** (`200`): @@ -1602,7 +1606,7 @@ curl -X POST http://localhost:6790/webhooks \ | Field | Type | Required | Description | | -------- | ---------- | -------- | ------------------------------------------------------------------------------------------------------------------- | -| `url` | `string` | Yes | HTTP or HTTPS endpoint URL. Validated against SSRF (localhost, private IPs, cloud metadata blocked). | +| `url` | `string` | Yes | HTTP or HTTPS endpoint URL. Registration rejects prohibited literal hostnames/IPs (localhost, private IPs, cloud metadata). DNS resolution and redirect destinations are not validated; use trusted registrations and network egress controls. | | `events` | `string[]` | Yes | Event types: `job.pushed`, `job.started`, `job.completed`, `job.failed`, `job.progress`. Other values are rejected. | | `queue` | `string` | No | Filter to specific queue. Omit for all queues. | | `secret` | `string` | No | HMAC signing secret for verifying webhook authenticity. | @@ -2069,7 +2073,18 @@ When `AUTH_TOKENS` is configured, the upgrade request must carry `Authorization: #### Connection Cleanup -When a WebSocket disconnects, all jobs owned by that client (pulled but not ACKed) are automatically released back to the queue. This prevents jobs from being stuck when a worker disconnects unexpectedly. +When a WebSocket disconnects, cleanup immediately releases eligible deliveries +still owned by that client without charging a processing attempt. Renewed leases +remain active because another pooled connection may still heartbeat them; +lease expiry/stall recovery reclaims them if heartbeats stop. Detached pulls +are not tracked for disconnect release. Memory/SQLite discard eligible `cron:` +prevent-overlap deliveries so the next scheduled tick can recreate them. +PostgreSQL also attempts that discard, but dependency-protected IDs use ordinary +transactional release instead. + +The WebSocket close path logs a cleanup failure. It does not use the TCP +cleanup retry loop or force-release fallback; jobs left active remain subject +to lease expiry/stall recovery. #### Complete Dashboard Example diff --git a/docs/src/content/docs/api/tcp/control.md b/docs/src/content/docs/api/tcp/control.md index a20117d4..7afb7689 100644 --- a/docs/src/content/docs/api/tcp/control.md +++ b/docs/src/content/docs/api/tcp/control.md @@ -169,8 +169,9 @@ Move an active job back to the delayed state. A missing or non-finite `delay` fails the command (`delay is required`, `delay must be a finite number`, ...) and leaves the job unchanged. As in 2.9.10, a -negative `delay` makes the job ready at once with a past run time (ahead of later ready -jobs; a PostgreSQL broker uses "now", as on 2.9.10) and a very large one is applied (clamped at +negative `delay` makes the job ready at once with a past run time (ahead of +equal-priority FIFO jobs with later run times; priority and LIFO ordering still +apply; a PostgreSQL broker uses "now", as on 2.9.10) and a very large one is applied (clamped at about 136,900 years). **Response:** diff --git a/docs/src/content/docs/api/tcp/cron.md b/docs/src/content/docs/api/tcp/cron.md index 32ee89f5..5723372a 100644 --- a/docs/src/content/docs/api/tcp/cron.md +++ b/docs/src/content/docs/api/tcp/cron.md @@ -57,7 +57,8 @@ Create or update a cron/repeating job schedule. The spawned-job options are validated with the same rules and messages as [`PUSH`](/api/tcp/jobs/#push), so a cron cannot admit a job a `PUSH` would refuse, and every template 2.9.10 stored is still accepted: only what no job can run with fails -(a NaN or non-numeric value, a negative `timeout` or `maxAttempts`, ...). Template +(a NaN or non-numeric value, a negative `timeout` or `backoff.delay`, ...). A +`maxAttempts` at or below one, including a negative number, runs the job once. Template fields are reported with a `jobOptions.` prefix (`jobOptions.timeout must be at least 0`), `priority` and `dedup.ttl` without it, and a `repeatEvery` above 4,320,000,000,000,000 ms fails with diff --git a/docs/src/content/docs/api/tcp/index.md b/docs/src/content/docs/api/tcp/index.md index e2467e1a..9d17d3ff 100644 --- a/docs/src/content/docs/api/tcp/index.md +++ b/docs/src/content/docs/api/tcp/index.md @@ -203,7 +203,21 @@ check `event.eventType` on the client side as well. ## Connection Lifecycle -When a TCP connection closes, the server automatically releases all jobs that were being processed by that client back to their queues. This uses retry logic with exponential backoff (up to 3 attempts) to ensure jobs are not left in an inconsistent state. +When a TCP connection closes, the server immediately releases eligible deliveries +still owned by that client without charging a processing attempt. Renewed leases +remain active because a pooled worker may heartbeat them on another connection; +lease expiry/stall recovery reclaims them if heartbeats stop. Detached pulls +(`detach: true`) are not tracked for disconnect release. Memory/SQLite discard +eligible `cron:` prevent-overlap deliveries so the next scheduled tick can +recreate them. PostgreSQL also attempts that discard, but dependency-protected +IDs use ordinary transactional release instead. + +The cleanup operation retries with exponential backoff, up to 3 attempts. That +budget retries cleanup, not job processing. If cleanup exhausts its retries, a +force-release fallback differs by backend: memory/SQLite expire heartbeat/start +state for stall recovery; PostgreSQL queues another transactional, token-fenced +release of the retained lease session. If release cannot complete, eventual +lease expiry remains the recovery path. ## Rate Limiting @@ -241,7 +255,10 @@ Queue names must satisfy the following constraints: ## Job Data Limits -Job data payloads are limited to **10 MB** when serialized. +New `PUSH`/`PUSHB` and flow admissions limit each JSON-serialized job payload to +10 × 1024 × 1024 string characters. This is a character count, not a byte cap; +`Update` intentionally has no job-data size limit. Enforce application payload +limits separately from the transport's 64 MiB frame limit. ## Command Summary diff --git a/docs/src/content/docs/api/tcp/jobs.md b/docs/src/content/docs/api/tcp/jobs.md index 66bee953..623ff8db 100644 --- a/docs/src/content/docs/api/tcp/jobs.md +++ b/docs/src/content/docs/api/tcp/jobs.md @@ -29,7 +29,7 @@ Add a single job to a queue. cmd: 'PUSH', queue: string, // Queue name (required, max 256 chars, alphanumeric/underscore/dash/dot/colon) name: string, // Job name metadata (required for protocol v3 clients) - data: any, // Untouched user payload (required, max 10 MB) + data: any, // Untouched user payload (required, max 10 × 1024 × 1024 JSON characters) priority?: number, // Ungrouped: any finite number, higher first. With groupId: 0 first, then ascending (0..2097151) delay?: number, // Delay in ms before processing (default: 0; negative = ready, past run time) maxAttempts?: number, // Max attempts (default: 3; 1 or less runs once, fraction rounds up, max 2147483647) @@ -37,7 +37,7 @@ Add a single job to a queue. ttl?: number, // Time-to-live in ms (>= 0) timeout?: number, // Processing timeout in ms (>= 0; 0 = none) uniqueKey?: string, // Deduplication key - jobId?: string, // Custom job ID (idempotent) + jobId?: string, // Custom job ID (retained-ownership dedup; see Custom-ID limits below) dependsOn?: string[], // Job IDs this job depends on tags?: string[], // Metadata tags groupId?: string, // Job group identifier @@ -84,9 +84,17 @@ violation fails the command with `{ ok: false, error }` naming the field, for ex accepted in either mode is accepted with its 2.9.10 result; durations above about 136,900 years (4.32e15 ms) are clamped. Durations may be fractional. A negative `delay` keeps its past run time, as in 2.9.10: the job is ready at once (never `delayed`) and -runs ahead of ready jobs with a later run time. Embedded `Queue.add`, `addBulk`, flows and job schedulers apply the same +runs ahead of equal-priority FIFO jobs with a later run time; priority and LIFO +ordering still apply. Embedded `Queue.add`, `addBulk`, flows and job schedulers apply the same rules (their messages name the SDK option, such as `attempts`). +Custom `jobId` suppression depends on retained live ownership: memory/SQLite +use a bounded cache (50,000 IDs by default), while PostgreSQL checks the database. +After local cache eviction, preserving the original generation is not guaranteed; +neither SQLite's upsert nor `durable` removes that limit. See +[Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). Atomic flow +creation has a separate reject-existing contract. + The `backoff` field also accepts an object form: `{ type: 'fixed' | 'exponential', delay?: number, maxDelay?: number }` (any other `type` runs as exponential, as in 2.9.10). `maxDelay` caps each computed retry delay for that job (default: 1 hour when omitted). A missing `delay` is the 1000 ms default base; a given one must be a finite number of at least 0; `maxDelay` must be a finite number between 0 and 86,400,000 ms (1 day); `maxDelay: null` is treated as omitted, and any other invalid value fails the command. The same rule applies to every job in `PUSHB` and `PUSHF`. **Response:** @@ -139,8 +147,14 @@ violation the whole batch is rejected with an error naming the offending index When `groupId` is present, `priority` is the intra-group priority and must be an integer from `0` through `2,097,151`; lower values run first. `groupMaxSize` -makes pending-depth admission atomic. If one member would exceed its group cap, -the complete `PUSHB` is rejected without partial writes. +makes the pending-depth check and admission atomic for each job. In memory/SQLite, +a runtime rejection such as a later member exceeding its group cap preserves +the already accepted prefix, even though `PUSHB` returns an error. Preflight +validation failures reject before mutation. Reconcile accepted jobs before +retrying a failed batch; stable custom IDs alone do not make retries safe after +memory/SQLite ownership-cache eviction. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). +Keep external effects idempotent. PostgreSQL uses transactional +batch insertion. `PUSHF` graph creation is fully atomic. **Response:** @@ -182,7 +196,8 @@ calls. The complete graph is validated before mutation: strict runtime types, duplicate/missing/asymmetric edges, cycles, policy conflicts, 10,000 jobs, -10 MB per job and 64 MB aggregate data. With configured SQLite, all job rows +10 × 1024 × 1024 JSON string characters per job and 64 × 1024 × 1024 +characters across the graph. With configured SQLite, all job rows commit in one immediate transaction before any leaf becomes visible. In PostgreSQL mode commits the graph, dependency edges, and ordered durable events in one database transaction before publication. In memory-only mode, @@ -486,7 +501,8 @@ Change the delay of a delayed job (recomputes `runAt`). `delay` (ms from now) is required and must be a finite number (a numeric string counts as its number). As with the `PUSH` `delay` and as in 2.9.10, a negative value makes the -job ready at once with a past run time (ahead of later ready jobs); a PostgreSQL broker +job ready at once with a past run time (ahead of equal-priority FIFO jobs with later +run times; priority and LIFO ordering still apply); a PostgreSQL broker applies it as "now", as it did on 2.9.10. NaN or an infinity fails the command (`delay must be a finite number`). `token` is required when the job is active and currently leased. Worker diff --git a/docs/src/content/docs/api/tcp/workers.md b/docs/src/content/docs/api/tcp/workers.md index ef4509cc..42c5d6dd 100644 --- a/docs/src/content/docs/api/tcp/workers.md +++ b/docs/src/content/docs/api/tcp/workers.md @@ -129,7 +129,10 @@ List all registered workers and their stats. ### AddWebhook -Register a webhook to receive event notifications. URLs are validated to prevent SSRF (localhost, private IPs, and cloud metadata endpoints are blocked). +Register a webhook to receive event notifications. Registration rejects prohibited +literal hostnames/IPs, including localhost, private IPs and cloud metadata +endpoints. DNS resolution and redirect destinations are not validated as an +egress policy; use trusted registrations and network egress controls. **Request:** diff --git a/docs/src/content/docs/api/types.md b/docs/src/content/docs/api/types.md index 68d89803..e4a77da3 100644 --- a/docs/src/content/docs/api/types.md +++ b/docs/src/content/docs/api/types.md @@ -234,7 +234,8 @@ interface Job { /** * Change the delay of a job (any finite number of ms; a negative delay makes the job - * ready at once with a past run time, ahead of later ready jobs, as in 2.9.10). Throws for a NaN, infinite or non-numeric delay. A job that + * ready at once with a past run time, ahead of equal-priority FIFO jobs with later + * run times, as in 2.9.10; priority and LIFO ordering still apply). Throws for a NaN, infinite or non-numeric delay. A job that * can no longer be changed (gone, active or finished) is left as is, as in 2.9.10; a * Worker processor's own `changeDelay` throws instead, since its delivery must end. */ @@ -483,8 +484,9 @@ interface JobOptions { /** * Delay in milliseconds before job becomes available (default: 0). A negative delay - * makes the job ready at once with a run time in the past, so it runs ahead of later - * ready jobs, as in 2.9.10. The maximum is `MAX_JOB_DURATION_MS` (4,320,000,000,000,000 + * makes the job ready at once with a run time in the past, so it runs ahead of + * equal-priority FIFO jobs with later run times, as in 2.9.10. Priority and LIFO + * ordering still apply. The maximum is `MAX_JOB_DURATION_MS` (4,320,000,000,000,000 * ms, about 136,900 years); a longer finite delay is clamped to it. */ delay?: number; @@ -506,8 +508,9 @@ interface JobOptions { timeout?: number; /** - * Custom job ID for idempotent/deduplication. - * If a job with this ID already exists, the existing job is returned. + * Custom job ID; returns the live generation while ownership is retained. + * Memory/SQLite use a bounded ownership cache (50,000 IDs by default); + * PostgreSQL checks database identity. See custom-ID limits in the deduplication guide. */ jobId?: string; diff --git a/docs/src/content/docs/architecture/application-layer.md b/docs/src/content/docs/architecture/application-layer.md index dd1235b1..233b5559 100644 --- a/docs/src/content/docs/architecture/application-layer.md +++ b/docs/src/content/docs/architecture/application-layer.md @@ -115,6 +115,12 @@ and the user-facing [storage guide](/guide/databases/). ## PUSH Operation Flow +This memory/SQLite path uses the bounded `customIdMap` LRU (50,000 IDs by +default). Its live-generation suppression requires a cache hit; eviction can +end protection while a job remains live, and SQLite upserts do not add a +uniqueness rejection. PostgreSQL uses separate locked database admission. +See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). +
PUSH flowmemory / SQLite push(queue, input)
1. Generate UUIDv7 ID, 2. check customId idempotency customIdMap, if exists return existing job
diff --git a/docs/src/content/docs/architecture/model-based-testing.md b/docs/src/content/docs/architecture/model-based-testing.md index 353704fe..5efb0297 100644 --- a/docs/src/content/docs/architecture/model-based-testing.md +++ b/docs/src/content/docs/architecture/model-based-testing.md @@ -87,7 +87,10 @@ The oracle checks these contracts together: before its dependencies. - **Failure policies:** fail-parent, ignore, continue, and remove-dependency modes produce their declared parent transition. -- **Resources:** custom IDs and unique keys remain exclusive; concurrency and +- **Resources:** within the generated campaign's retained ownership working + set, custom IDs and unique keys remain exclusive. This does not prove + ordinary memory/SQLite adds safe after cache eviction; see + [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). Concurrency and rate tokens admit no extra job or leak when rate acquisition rejects after a concurrency acquisition; expired jobs leave memory, counters, index, write buffer, and SQLite exactly once. diff --git a/docs/src/content/docs/architecture/persistence.md b/docs/src/content/docs/architecture/persistence.md index f5e1a5bd..1970473a 100644 --- a/docs/src/content/docs/architecture/persistence.md +++ b/docs/src/content/docs/architecture/persistence.md @@ -94,7 +94,7 @@ Benchmarks](/guide/benchmarks/) for distributions and TCP results.
(queue, state, created_at, id) stable state pagination
(run_at) delayed job scheduler
(queue, unique_key) deduplication
-
(custom_id) idempotency
+
(custom_id) lookup; live dedup depends on ownership cache
(parent_id) parent job lookup
diff --git a/docs/src/content/docs/changelog.md b/docs/src/content/docs/changelog.md index d8ca372d..6f6a7a7a 100644 --- a/docs/src/content/docs/changelog.md +++ b/docs/src/content/docs/changelog.md @@ -455,6 +455,20 @@ head: ### Documentation +- Record the completed documentation audit with its green source CI, isolated + Linux validation, rendered-link checks and review evidence. Keep live custom-ID + reuse after cache eviction as a separate reproduce-first runtime work item. + +- Correct current guides, protocol and agent references against the implementation: + use authoritative async methods for TCP DLQ and cleanup operations, distinguish + accepted-prefix bulk admission from atomic flow creation, and describe actual + retry timing, bounded custom-ID ownership, cooperative timeouts and disconnect + recovery. Clarify configuration fallbacks, payload and webhook validation + boundaries, SQLite durability, persistence fields and backup configuration. + Repair the workflow summary's missing classifier result and stale API names. + Track live-ID reuse after cache eviction for a separate runtime reproduction. + These corrections document existing behavior; they do not change the runtime. + - Release metadata refreshed after the 2.9.22 publication (`docs/src/data/releaseMeta.json`): 2.9.20, 2.9.21 and 2.9.22 now carry their tag, GitHub release, npm, Docker and contributor data, and 2.9.19 its npm diff --git a/docs/src/content/docs/examples.mdx b/docs/src/content/docs/examples.mdx index 4e2bdb2c..de5d75b8 100644 --- a/docs/src/content/docs/examples.mdx +++ b/docs/src/content/docs/examples.mdx @@ -479,7 +479,13 @@ Timezones and schedule management in the [cron guide](/guide/cron/). ## Deduplicate jobs with jobId -Adding a job with the `jobId` of an unfinished job (waiting, delayed, waiting for children, or active) returns the existing job instead of creating a duplicate, so re-runs after a restart do not queue the same work twice. Once that job has completed or moved to the DLQ, the same `jobId` creates a fresh job, so it does not by itself guarantee "exactly one welcome email per user". +Adding an unfinished `jobId` returns its live generation while ownership is +retained. Memory/SQLite use a bounded cache (50,000 IDs by default); eviction +can end protection before completion. PostgreSQL checks database identity. +After completion or DLQ the ID may create a fresh job. See +[Custom-ID limits](/guide/queue/deduplication/#custom-id-limits): reconcile +uncertain admissions and keep welcome-email or other external effects +idempotent in the application. diff --git a/docs/src/content/docs/faq.mdx b/docs/src/content/docs/faq.mdx index b635e24b..60e83e56 100644 --- a/docs/src/content/docs/faq.mdx +++ b/docs/src/content/docs/faq.mdx @@ -40,7 +40,7 @@ historical SDK API remains available explicitly from `bunqueue-client/legacy`. ### What are the requirements? -Bun 1.4.0 or newer (enforced via the package `engines` field) on macOS, Linux, or Windows via WSL. An SSD helps write throughput. Install steps are on the [installation page](/guide/installation/). +Bun 1.4.0 or newer (enforced via the package `engines` field) on macOS, Linux, or Windows. An SSD helps write throughput. Install steps are on the [installation page](/guide/installation/). ### How heavy is the install? @@ -68,8 +68,8 @@ bunqueue start ### How does persistence work? Jobs are written to SQLite in WAL mode (write-ahead logging, a journal that lets -reads and writes overlap). By default writes are buffered for up to 10 ms and -flushed in batches. If that process-crash window is unacceptable for a job, add +reads and writes overlap). By default writes use a typical 10 ms flush interval; +stalls or retries can keep them buffered longer. If that process-crash exposure is unacceptable for a job, add it with `{ durable: true }` to commit before `add()` returns. This closes bunqueue's application buffer; power-loss durability still depends on SQLite's host, filesystem, and storage. Current native @@ -112,48 +112,53 @@ Three knobs, in order of impact: raise worker `concurrency` (parallel jobs per w ### How does deduplication work? -Pass a `jobId`. Adding the same `jobId` while the earlier job is still unfinished (waiting, delayed, waiting for children, or active) returns the existing job instead of creating a duplicate. Once that job has completed or moved to the DLQ, the same `jobId` creates a fresh job. This makes webhook handlers and restart-recovery code safe to re-run while the work is pending: +Pass a `jobId` to return the unfinished generation while its ownership is +retained. Memory/SQLite use a bounded cache (50,000 IDs by default); after +eviction, a re-add is not guaranteed to preserve the original live job. +PostgreSQL checks live identity in the database. After completion or DLQ, the +ID may create a fresh job. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits) +and keep webhook/payment effects idempotent in your application: ```typescript -await queue.add('charge', data, { jobId: `order-${orderId}` }); // idempotent +await queue.add('charge', data, { jobId: `order-${orderId}` }); // retained-ownership dedup; see limits above ``` ```typescript -await queue.add('charge', data, { jobId: `order-${orderId}` }); // idempotent +await queue.add('charge', data, { jobId: `order-${orderId}` }); // retained-ownership dedup; see limits above ``` ```python -queue.add("charge", data, job_id=f"order-{order_id}") # idempotent +queue.add("charge", data, job_id=f"order-{order_id}") # retained-ownership dedup; see limits above ``` ```php -$queue->add('charge', $data, ['jobId' => "order-{$orderId}"]); // idempotent +$queue->add('charge', $data, ['jobId' => "order-{$orderId}"]); // retained-ownership dedup; see limits above ``` ```go -queue.Add("charge", data, bunqueue.JobOptions{"jobId": "order-" + orderID}) // idempotent +queue.Add("charge", data, bunqueue.JobOptions{"jobId": "order-" + orderID}) // retained-ownership dedup; see limits above ``` ```rust -// idempotent +// Retained-ownership dedup; see limits above queue.add("charge", data, JobOptions { job_id: Some(format!("order-{order_id}")), ..Default::default() @@ -164,7 +169,7 @@ queue.add("charge", data, JobOptions { ```elixir -# idempotent +# Retained-ownership dedup; see limits above {:ok, _job} = Bunqueue.Queue.add(queue, "charge", data, jobId: "order-#{order_id}") ``` @@ -239,7 +244,7 @@ Workers send heartbeats (periodic "still alive" pings). If one goes silent, stal ### What is the dead letter queue? -The DLQ is the holding area for jobs that exhausted their retries, threw an unrecoverable error, or stalled too many times. Nothing is silently dropped: you can inspect entries, retry them (`queue.retryDlq()`), or purge them. It also supports auto-retry and expiration policies. See [DLQ](/guide/dlq/). +The DLQ is the holding area for jobs that exhausted their retries, threw an unrecoverable error, or stalled too many times. Nothing is silently dropped: you can inspect entries, retry them (`await queue.retryDlqAsync()`), or purge them. It also supports auto-retry and expiration policies. See [DLQ](/guide/dlq/). ### Can I control processing order? @@ -317,7 +322,7 @@ The job was already removed: it completed with `removeOnComplete: true`, failed ### High memory usage -Usually accumulation: completed jobs kept around, a growing DLQ, or oversized job payloads (keep payloads small, store big blobs elsewhere and pass a reference). Add jobs with `removeOnComplete: true`, purge the DLQ periodically (`queue.purgeDlq()`), and clean old jobs with `queue.clean(3600000, 1000)` (grace period in ms, max jobs to remove). +Usually accumulation: completed jobs kept around, a growing DLQ, or oversized job payloads (keep payloads small, store big blobs elsewhere and pass a reference). Add jobs with `removeOnComplete: true`, purge the DLQ periodically (`await queue.purgeDlqAsync()`), and clean old jobs with `await queue.cleanAsync(3600000, 1000, 'completed')` (grace period in ms, max jobs to remove). These async methods work in embedded and TCP modes. ## Contributing diff --git a/docs/src/content/docs/guide/benchmarks.mdx b/docs/src/content/docs/guide/benchmarks.mdx index cbb8b989..5cb2ee9e 100644 --- a/docs/src/content/docs/guide/benchmarks.mdx +++ b/docs/src/content/docs/guide/benchmarks.mdx @@ -50,7 +50,7 @@ These labels are not interchangeable: | Internal in-memory | Direct `QueueManager` batch API, no `dataPath`, no SQLite write | | Embedded on-disk | Public `Queue` API and a fresh SQLite `dataPath` | | TCP on-disk | Public client over localhost MessagePack to a broker with its own SQLite file | -| Buffered | Normal write-behind persistence, with the documented ≤10 ms hard-crash window | +| Buffered | Normal write-behind persistence; a crash can lose pending admissions. Typical flush interval: 10 ms, longer during stalls/retries | | Durable (SQLite) | `durable:true`; the timed call waits for synchronous SQLite persistence | | Workflow | A complete multi-node execution, not one queue operation | diff --git a/docs/src/content/docs/guide/cli.md b/docs/src/content/docs/guide/cli.md index 07286d51..916e8c2b 100644 --- a/docs/src/content/docs/guide/cli.md +++ b/docs/src/content/docs/guide/cli.md @@ -93,12 +93,17 @@ Job IDs are UUID v7 strings (time-ordered). Push accepts options for priority, r ```bash bunqueue push emails '{"to":"vip@example.com"}' --priority 10 # higher = sooner bunqueue push notifications '{"msg":"hi"}' --delay 5000 # run in 5s -bunqueue push orders '{"orderId":"ORD-123"}' --job-id order-ORD-123 # idempotent ID +bunqueue push orders '{"orderId":"ORD-123"}' --job-id order-ORD-123 # retained-ownership dedup bunqueue push emails '{"to":"a@b.c"}' --max-attempts 5 --backoff 2000 # retry config bunqueue push notifications '{"userId":"1"}' -u user-1-notify # unique key (dedup) bunqueue push aggregate '{"type":"sum"}' --depends-on job-1,job-2 # wait for other jobs ``` +Custom IDs use the selected broker's ownership rules: memory/SQLite keep a +bounded cache (50,000 IDs by default), whereas PostgreSQL checks the database. +See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits); keep +external effects idempotent. + | Option | Short | Default | Description | | ---------------------- | ----- | ------- | ----------------------------------- | | `--priority` | `-P` | `0` | Higher = processed first | diff --git a/docs/src/content/docs/guide/configuration.md b/docs/src/content/docs/guide/configuration.md index 100947ba..f792464e 100644 --- a/docs/src/content/docs/guide/configuration.md +++ b/docs/src/content/docs/guide/configuration.md @@ -165,7 +165,8 @@ defineConfig({ number ≥ 1); it does not delete SQLite rows. Set `completedRetentionMs` to opt into age-based durable cleanup (up to 1,000 oldest eligible rows per 10-second cleanup tick). The default is `null`, so completed rows remain until -`queue.clean(...)`, `obliterate`, or another explicit policy removes them. +`await queue.cleanAsync(graceMs, limit, 'completed')`, `obliterate`, or another +explicit policy removes them. The async cleanup method works in embedded and TCP modes. Results still needed by live dependency consumers are protected until the consumer leaves the graph. Finite non-negative values are rounded down to whole milliseconds and `null` disables retention. A negative, non-finite or diff --git a/docs/src/content/docs/guide/databases.mdx b/docs/src/content/docs/guide/databases.mdx index 96d75dbd..3160dc30 100644 --- a/docs/src/content/docs/guide/databases.mdx +++ b/docs/src/content/docs/guide/databases.mdx @@ -305,7 +305,7 @@ application-level deduplication for external effects. ## SQLite retention and schema upgrades `maxCompletedJobs` is a hot-cache/recovery bound, not a database-retention -policy. `queue.clean(..., 'completed')` is SQLite-authoritative and deletes the +policy. `await queue.cleanAsync(graceMs, limit, 'completed')` is SQLite-authoritative and deletes the oldest eligible retained rows even after they leave that cache. For automatic retention, set `storage.completedRetentionMs`, `BUNQUEUE_COMPLETED_RETENTION_MS`, or `--completed-retention-ms`; it is disabled @@ -316,8 +316,11 @@ exists, including after restart; the cleanup tick unregisters the name after the last row is committed away. Invalid direct retention values cannot make that deletion immediate: negative, non-finite, and unsafe values disable the policy, while finite non-negative fractions are floored to milliseconds. In the -server's own configuration (config file, environment or CLI flag) such a value -is rejected instead, and the server does not start. +server's configuration, invalid config-file values and negative or unreadable +environment/CLI values preserve the no-retention fallback with a warning. +Environment/CLI forms that would be misread (for example `1e12`) are rejected; +see [Configuration](/guide/configuration/) and [Environment Variables](/guide/env-vars/) +for each input's rules. Deleting retained rows makes their SQLite pages reusable, which bounds future growth for a steady workload, but it does not automatically shrink a database @@ -373,7 +376,7 @@ bunqueue embedded and forward jobs to one central, durable bunqueue server: const local = new Queue('ingest', { embedded: true, dataPath: '/var/lib/bunqueue/spool.db', - defaultJobOptions: { durable: true }, // close SQLite's 10ms hard-crash window + defaultJobOptions: { durable: true }, // bypass SQLite's application admission buffer }); const forwarder = local.forward({ diff --git a/docs/src/content/docs/guide/deployment.mdx b/docs/src/content/docs/guide/deployment.mdx index f6a6ad2a..d4d45b9d 100644 --- a/docs/src/content/docs/guide/deployment.mdx +++ b/docs/src/content/docs/guide/deployment.mdx @@ -663,7 +663,7 @@ recommendation. See [PostgreSQL benchmark evidence](/guide/benchmarks/#postgresq - **Two embedded processes, one file, is silent corruption.** See the warning above; always go through the server for multi-process. - **Copying the database while the server runs can produce a torn copy.** SQLite runs in WAL mode (writes go to a `.db-wal` sidecar file first). Stop the server, then copy `bunq.db` and `bunq.db-wal` together, or just use the built-in S3 backup, which snapshots safely while live. - **`AUTH_TOKENS` is not set by default.** Anyone who can reach the ports can push and pull jobs. Set tokens (`openssl rand -hex 32`) and either enable [native TLS](/guide/tls/) or keep the ports on a private network. See [Security](/security/). -- **SQLite jobs that cannot tolerate bunqueue's process-crash buffer should be `durable: true`.** SQLite buffers ordinary writes for up to 10 ms; `durable` commits before the add returns. Host/filesystem/media durability remains operational. PostgreSQL mutations are already transactional and do not use this buffer. Details in [Production Operations](/guide/production/). +- **SQLite jobs that cannot tolerate bunqueue's process-crash buffer should be `durable: true`.** SQLite buffers ordinary writes with a typical 10 ms flush interval, which can be longer during stalls/retries; `durable` commits before the add returns. Host/filesystem/media durability remains operational. PostgreSQL mutations are already transactional and do not use this buffer. Details in [Production Operations](/guide/production/). - **Do not improvise a mixed-version PostgreSQL rollout.** Stop the old broker fleet before the first new binary migrates the schema, verify one new broker, then start the rest at the same version. An old binary may reject the newer diff --git a/docs/src/content/docs/guide/dlq/index.mdx b/docs/src/content/docs/guide/dlq/index.mdx index d83bd1ce..710c6d47 100644 --- a/docs/src/content/docs/guide/dlq/index.mdx +++ b/docs/src/content/docs/guide/dlq/index.mdx @@ -72,7 +72,7 @@ for (const entry of entries) { } // Put everything back in the queue for another try -await queue.retryDlq(); +await queue.retryDlqAsync(); ``` diff --git a/docs/src/content/docs/guide/dlq/operations.mdx b/docs/src/content/docs/guide/dlq/operations.mdx index 5d7d0963..2030d1af 100644 --- a/docs/src/content/docs/guide/dlq/operations.mdx +++ b/docs/src/content/docs/guide/dlq/operations.mdx @@ -134,8 +134,8 @@ const n = await queue.retryDlqAsync(); // retry and get the ```typescript -await queue.retryDlq(); // retry everything -await queue.retryDlq('job-123'); // retry one job +await queue.retryDlqAsync(); // retry everything +await queue.retryDlqAsync('job-123'); // retry one job await queue.retryJobs({ count: 100 }); // retry only the first 100 entries ``` diff --git a/docs/src/content/docs/guide/flow/index.mdx b/docs/src/content/docs/guide/flow/index.mdx index 9dd4c0ce..f152cfc0 100644 --- a/docs/src/content/docs/guide/flow/index.mdx +++ b/docs/src/content/docs/guide/flow/index.mdx @@ -266,8 +266,9 @@ before any broker can claim a leaf. Without either persistent backend, embedded mode is intentionally memory-only: creation is still atomically visible to workers, but a process crash cannot recover it. -- A flow may contain at most 10,000 jobs, at most 10 MB of data per job and - 64 MB across the batch. A root is depth 0; descendants may be at most 100 +- A flow may contain at most 10,000 jobs, at most 10 × 1024 × 1024 + JSON string characters of data per job and 64 × 1024 × 1024 characters + across the batch (character counts, not byte limits). A root is depth 0; descendants may be at most 100 edges below it. - `jobId` is allowed, but cannot be empty or contain `:`. Reusing any existing or retained flow ID—including durable job/DLQ rows, completion or timeout diff --git a/docs/src/content/docs/guide/glossary.md b/docs/src/content/docs/guide/glossary.md index e2a15681..fee38016 100644 --- a/docs/src/content/docs/guide/glossary.md +++ b/docs/src/content/docs/guide/glossary.md @@ -118,7 +118,10 @@ Capping how many jobs run per time window, to protect a downstream service like ### Deduplication and idempotency -Giving a job a custom `jobId` so adding it again while the first one is still waiting or running does nothing. This makes `add()` safe to call more than once for the same logical task. See the [Queue API](/guide/queue/). +Giving a job a custom `jobId` suppresses repeat adds while its live ownership is +retained. Memory/SQLite use a bounded cache (50,000 IDs by default); PostgreSQL +checks database identity. This is separate from application idempotency for +external effects. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). ## Composing jobs @@ -171,7 +174,9 @@ The compact binary format used on the TCP wire, smaller and faster to parse than An edge pattern: a small embedded queue stores jobs on a persistent local volume, then forwards them to a central server when it can reach it. A network outage does not drop jobs while the local process and volume survive; use SQLite -`durable: true` when even its 10ms hard-crash window is unacceptable. See +`durable: true` when losing pending buffered admissions on a crash is +unacceptable. SQLite's typical 10 ms flush interval can be longer during +stalls/retries. See [IoT & Edge](/guide/iot-edge/). :::tip[Related] diff --git a/docs/src/content/docs/guide/hono.md b/docs/src/content/docs/guide/hono.md index c3a734f0..f9ddecc0 100644 --- a/docs/src/content/docs/guide/hono.md +++ b/docs/src/content/docs/guide/hono.md @@ -47,7 +47,7 @@ const app = new Hono(); app.post('/api/send-email', async (c) => { const body = await c.req.json(); const job = await emails.add('send', body, { - attempts: 3, // retry up to 3 times on failure + attempts: 3, // up to 3 total attempts (2 retries) backoff: 5000, // exponential from a 5s base: ~10s, then ~20s between retries }); return c.json({ queued: true, jobId: job.id }); diff --git a/docs/src/content/docs/guide/iot-edge.md b/docs/src/content/docs/guide/iot-edge.md index f6a87cb0..236eb98d 100644 --- a/docs/src/content/docs/guide/iot-edge.md +++ b/docs/src/content/docs/guide/iot-edge.md @@ -123,12 +123,14 @@ What `forward()` guarantees: push fails, the job fails locally, retries with backoff, and after its attempts lands in the local DLQ. When the uplink returns, `local.retryDlq()` re-enqueues it. This assumes the gateway process or persistent volume survives; use - `durable: true` for local jobs that cannot tolerate SQLite's 10ms hard-crash - window. -- **Re-forwards deduplicate while the remote job is live.** Each forwarded job + `durable: true` for local jobs that cannot tolerate losing pending buffered + admissions on a crash. SQLite's typical 10 ms flush interval can be longer + during stalls/retries. +- **Re-forwards deduplicate while remote ownership is retained.** Each forwarded job carries a deterministic remote job id, `fwd::`, and - the server treats a second push with that id as a no-op while the remote job - is waiting, delayed or active. Once it completes or reaches the DLQ the id is + a memory/SQLite server returns the live generation only while its bounded + ownership cache retains the ID (50,000 entries by default). PostgreSQL checks + live database identity instead. Once it completes or reaches the DLQ the id is released, so a re-forward after that creates a new remote job. End-to-end processing is at-least-once: make remote processors idempotent. - **Priority is preserved.** Pass `durable: true` in the forward options to @@ -143,7 +145,8 @@ The embedded queue persists to SQLite in WAL mode: committed frames use an append-only sidecar and SQLite recovery preserves transactional consistency across ordinary process or power interruption. Hardware, filesystem, and volume durability still belong to the operator. By default writes are batched -for up to 10 ms; for readings you cannot afford to lose inside that window, +at a typical 10 ms flush interval (longer during stalls/retries); for readings +you cannot afford to lose while buffered, mark them durable, meaning committed before `add()` returns: ```typescript @@ -180,9 +183,11 @@ See [Cron & Scheduled Jobs](/guide/cron/) for cron expressions and timezones. ## Gotchas -- **Forward dedup only covers live jobs.** The server releases a custom job id +- **Forward dedup has backend and lifetime limits.** The server releases a custom job id when the remote job completes or moves to the DLQ, whether or not the job - itself is kept, and its bounded id cache can evict entries earlier. A + itself is kept. Memory/SQLite's 50,000-entry default ownership cache can evict + live IDs earlier; PostgreSQL's database identity check is independent of that + cache. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). A re-forward that arrives after the original finished is accepted as a new job, so strict exactly-once effects require downstream idempotency. - **`forwarder.on('error')` is observability, not control flow.** Failed forwards are already handled by the local retry and DLQ path; the event just tells you the uplink is unhappy. diff --git a/docs/src/content/docs/guide/mcp.mdx b/docs/src/content/docs/guide/mcp.mdx index b0613a22..541dfb9b 100644 --- a/docs/src/content/docs/guide/mcp.mdx +++ b/docs/src/content/docs/guide/mcp.mdx @@ -439,7 +439,7 @@ The server also offers 3 prompts, `bunqueue_health_report`, `bunqueue_debug_queu
Job options: retries, timeouts, custom ids, deduplication -`add_job` and each item of `add_jobs_bulk` accept `priority`, `delay`, `attempts`, `backoff` (a number, used as an exponential base, or `{ type: 'fixed' | 'exponential', delay, maxDelay }`), `timeout`, `jobId`, `deduplication` (`{ id, ttl, extend, replace }`, per queue), `removeOnComplete`, `removeOnFail`, `durable`, `lifo`, `tags` and `stallTimeout`. Adding a `jobId` again while that job is unfinished returns the existing job. The flow tools take the same options, except `deduplication`, `tags` and `durable`. `get_job` shows the options a job carries. +`add_job` and each item of `add_jobs_bulk` accept `priority`, `delay`, `attempts`, `backoff` (a number, used as an exponential base, or `{ type: 'fixed' | 'exponential', delay, maxDelay }`), `timeout`, `jobId`, `deduplication` (`{ id, ttl, extend, replace }`, per queue), `removeOnComplete`, `removeOnFail`, `durable`, `lifo`, `tags` and `stallTimeout`. Re-adding a custom `jobId` returns its unfinished generation while ownership is retained: memory/SQLite use a bounded cache (50,000 IDs by default), whereas PostgreSQL checks the database. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits); keep external effects idempotent. The flow tools take the same options, except `deduplication`, `tags` and `durable`, and independently reject existing flow IDs rather than collapsing duplicates. `get_job` shows the options a job carries.
diff --git a/docs/src/content/docs/guide/migration.mdx b/docs/src/content/docs/guide/migration.mdx index 43efbd61..743b4b01 100644 --- a/docs/src/content/docs/guide/migration.mdx +++ b/docs/src/content/docs/guide/migration.mdx @@ -168,7 +168,7 @@ Then, in your code: - **Classes**: `Queue`, `Worker`, `QueueEvents`, `FlowProducer`, same constructors minus the Redis connection. - **Job options**: `priority`, `delay`, `attempts`, `backoff` (including the `{ type, delay }` object form), `jobId`, `removeOnComplete`, `removeOnFail`, `repeat`. -- **`jobId` deduplication**: adding the same `jobId` while that job is still live (waiting, prioritized, delayed, waiting-children or active) returns the existing job instead of creating a duplicate. Once it has completed or reached the DLQ, the ID can be reused for a fresh job. +- **`jobId` deduplication**: a live generation is returned while ownership is retained. Ordinary memory/SQLite adds use a bounded cache (50,000 IDs by default); eviction can end this protection before completion. PostgreSQL checks live database identity. After completion or DLQ, the ID may be reused for a fresh job. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits); keep external effects idempotent. - **Events**: `worker.on('completed' | 'failed' | 'progress', ...)` fire with the same signatures. - **`QueueEvents` payloads**: `completed` carries `{ jobId, returnvalue }`, `failed` carries `{ jobId, failedReason }` and `progress` carries `{ jobId, data }`; `waiting`, `active`, `stalled`, `delayed`, `removed`, `duplicated`, `waiting-children` and `drained` are emitted under the same names. A `failed` event also fires for an attempt that will be retried, with `terminal: false`. In server mode pass the same `connection` as the Queue (or `embedded: false`): with neither, `QueueEvents` listens to the embedded queue of its own process. - **Job schedulers**: `queue.upsertJobScheduler(id, { pattern }, { name, data, opts })` (or `{ every }` for a fixed interval), `getJobScheduler()`, `getJobSchedulers()` and `removeJobScheduler()` keep their BullMQ v5 names. See [Job schedulers](/guide/queue/schedulers/). @@ -249,7 +249,7 @@ await queue.add('task', data, { repeat: { every: 3600000 } }); // fixed interval ### removeOnComplete is boolean only -BullMQ accepts `removeOnComplete: { age, count }` for retention rules. bunqueue accepts only `true` or `false`. Use `queue.clean()` for age-based cleanup. +BullMQ accepts `removeOnComplete: { age, count }` for retention rules. bunqueue accepts only `true` or `false`. Use `await queue.cleanAsync(graceMs, limit, 'completed')` for age-based cleanup in embedded or TCP mode. ### Sandboxed processors diff --git a/docs/src/content/docs/guide/production.mdx b/docs/src/content/docs/guide/production.mdx index 44853406..41ede375 100644 --- a/docs/src/content/docs/guide/production.mdx +++ b/docs/src/content/docs/guide/production.mdx @@ -24,12 +24,12 @@ import { Aside } from '@astrojs/starlight/components'; | Failure | What happens to the job | | ---------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Worker crashes mid-job | If its TCP connection closes, the server puts the jobs it held back in the queue at once. A worker that hangs with its connection open loses the job when its lease expires. Either way it may run twice, never zero times. | -| Server killed hard (`kill -9`) | Jobs written to disk survive and recover on restart. Jobs accepted in the last ~10 ms may be lost, unless they were added with `durable: true`. | +| Worker crashes mid-job | A TCP disconnect immediately releases eligible deliveries still owned by that client. Renewed leases remain active until expiry/stall recovery; detached pulls are not tracked. Processing may run again. See [TCP recovery](/api/tcp/#connection-lifecycle). | +| Server killed hard (`kill -9`) | Committed jobs survive ordinary process termination and recover on restart. Pending buffered admissions may be lost; the typical flush interval is 10 ms, longer during stalls/retries. `durable: true` bypasses this application buffer. | | Server restarted gracefully (SIGTERM, deploys) | After shutdown completes successfully, buffered SQLite writes are flushed before exit. A flush error or forced termination follows the hard-kill row instead. | | Job fails repeatedly | After its retry attempts are exhausted it moves to the DLQ (dead letter queue, a parking lot for failed jobs you can inspect and retry). See [DLQ](/guide/dlq/). | | Disk fills up | The server stays up and `/health` reports `degraded` with HTTP 503. Durable pushes fail explicitly instead of pretending to succeed. | -| The whole machine is lost | You restore the last S3 snapshot. Your maximum data loss is the backup interval. | +| The whole machine is lost | Restore the latest successful, restorable S3 snapshot. Loss extends back to that snapshot; monitor its age against your target backup interval. |
@@ -54,8 +54,9 @@ import { Aside } from '@astrojs/starlight/components'; Once a job is durably admitted, the processing guarantee is **at-least-once**: it can run more than once, but the broker does not silently discard that generation. SQLite's default write buffer precedes durable admission, so a hard -crash can still lose jobs accepted inside its documented 10 ms window unless -they use `durable: true`. No queue can atomically combine your external side +crash can still lose pending buffered admissions (a typical 10 ms flush interval, +not an upper bound during stalls/retries) unless they use `durable: true`. +No queue can atomically combine your external side effect with its own acknowledgment, so a crash can always land between them. What bunqueue guarantees around those edges: @@ -64,7 +65,7 @@ What bunqueue guarantees around those edges: If the job's processing `timeout` already finalized that generation, the ACK/FAIL is instead an explicit idempotent no-op. Two generations can never both complete the same job. -- Re-adding a job with the same custom `jobId` is a no-op, even under heavy concurrency. Producers that retry on timeout should always set one. +- Re-adding a custom `jobId` returns its unfinished generation while ownership is retained. Memory/SQLite use a bounded cache (50,000 IDs by default), so eviction can end this protection while a job is still live; PostgreSQL checks live identity in the database. See [Custom-ID limits](/guide/queue/deduplication/#custom-id-limits). After completion or DLQ, the ID may create a fresh job. Reconcile uncertain admissions before retrying, and retain application-side idempotency keys for external effects. - Duplicate ACKs are harmless.