fix rate-limit window/ttl, cron maxLimit budget, flaky TCP tests; add awaitable client APIs

- setGlobalRateLimit(max, duration) honors the window end-to-end (wire
  duration field, limiter refill math, HTTP passthrough); previously the
  duration argument was silently dropped everywhere
- queue.rateLimit(ms) expires broker-side via new RateLimit ttl field:
  works over TCP (was permanent) and survives client exit; lazy expiry,
  persisted with remaining time across restarts (migrations 15-16, one
  ALTER per migration so interrupted upgrades heal)
- cron scheduler: skip decisions (skipIfNoWorker, overlap guard) now run
  BEFORE the executions increment, so skipped fires no longer consume
  the maxLimit budget (worst case burned the whole cap with zero
  deliveries; also the root cause of the maxLimit=3 container flake)
- new awaitable Queue APIs closing the fire-and-forget ordering race
  over the multi-connection pool: obliterateAsync, pauseAsync,
  resumeAsync, drainAsync, retryDlqAsync, purgeDlqAsync,
  setStallConfigAsync, setDlqConfigAsync, setGlobalRateLimitAsync,
  removeGlobalRateLimitAsync, setGlobalConcurrencyAsync,
  removeGlobalConcurrencyAsync, getDlqJobsAsync (remote DLQ listing)
- fix flaky TCP suites test-flow-advanced / test-frameparser-pipelining
  (obliterate racing subsequent pushes) and the 20k recovery test
  timeout on slow CI runners

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: egeominotti

docs/data-model.md

@@ -231,18 +231,24 @@ See [Deduplication & Unique Jobs](./features/deduplication-and-unique.md).
 Defined in `src/domain/types/queue.ts`.
 
 ```typescript
-export interface QueueState {       // queue.ts:7-13
+export interface QueueState {       // queue.ts:7-19
   readonly name: string;
   paused: boolean;
-  rateLimit: number | null;         // token-bucket capacity (jobs/sec)
+  rateLimit: number | null;         // token-bucket capacity
+  rateLimitDuration: number | null; // window ms (null = 1000ms default)
+  rateLimitExpiresAt: number | null; // epoch ms auto-expiry (null = permanent)
   concurrencyLimit: number | null;  // max active jobs
   activeCount: number;
 }
 ```
 
-`createQueueState` (`queue.ts:16-24`) defaults everything off (`paused:false`,
-limits `null`, `activeCount:0`). This is the row persisted in `queue_state`
-(see schema) for control-state recovery (#100).
+`createQueueState` defaults everything off (`paused:false`, limits `null`,
+`activeCount:0`). This is the row persisted in `queue_state` (see schema) for
+control-state recovery (#100). `rateLimitDuration` makes the limit mean
+"`rateLimit` per `duration` ms" (refill rate = `limit / (duration/1000)`
+tokens/sec); `rateLimitExpiresAt` is checked lazily on acquire and on limit
+reads — an expired limit clears itself broker-side, and recovery skips
+already-expired rows (restoring live ones with their remaining TTL).
 
 Two runtime limiter classes back the config:
 
@@ -728,19 +734,23 @@ CREATE TABLE IF NOT EXISTS cron_jobs (
 
 Row type `DbCron` at `statements.ts:155-172`.
 
-### `queue_state` (schema.ts:123-128)
+### `queue_state` (schema.ts:128-135)
 
 ```sql
 CREATE TABLE IF NOT EXISTS queue_state (
     name TEXT PRIMARY KEY,
     paused INTEGER NOT NULL DEFAULT 0,
     rate_limit INTEGER,
-    concurrency_limit INTEGER
+    concurrency_limit INTEGER,
+    rate_limit_duration INTEGER,   -- window ms (migration 15)
+    rate_limit_expires_at INTEGER  -- epoch ms auto-expiry (migration 16)
 );
 ```
 
 Persists queue control-state for recovery (#100); row type `DbQueueState`
-(`statements.ts:175-180`).
+(`statements.ts:175-183`). On boot, recovery skips rate-limit rows whose
+`rate_limit_expires_at` is already in the past and restores still-live TTL'd
+limits with their remaining time.
 
 ### `migrations` (schema.ts:132-137)
 
@@ -751,9 +761,9 @@ CREATE TABLE IF NOT EXISTS migrations (
 );
 ```
 
-### Migrations (schema.ts:140-196)
+### Migrations (schema.ts:140-210)
 
-`SCHEMA_VERSION = 14`. The migrate routine (`sqlite.ts:255-278`) reads
+`SCHEMA_VERSION = 16`. The migrate routine (`sqlite.ts:255-278`) reads
 `MAX(version)`; if below current, runs the full `SCHEMA` (idempotent
 `CREATE … IF NOT EXISTS`) then applies each incremental `ALTER`/`CREATE INDEX`
 above the stored version (wrapped in try/catch since columns may already exist),
@@ -772,6 +782,8 @@ then records `SCHEMA_VERSION`.
 | 12      | `cron_jobs.job_options` BLOB (per-cron retry/cleanup policy, #86) |
 | 13      | `jobs.stacktrace` BLOB (persist last failure stack, #74)          |
 | 14      | Stable `getJobs` indexes on `(queue, created_at, id)` and `(queue, state, created_at, id)` |
+| 15      | `queue_state.rate_limit_duration` (rate-limit window)            |
+| 16      | `queue_state.rate_limit_expires_at` (rate-limit TTL auto-expiry; split from 15 so each ALTER retries idempotently) |
 
 (Versions 2–4 are unused gaps; only the keys present in `MIGRATIONS` run.)
 

docs/features/client-queue-sdk.md

@@ -52,15 +52,19 @@ Query (`queue.ts:226`): `getJob`, `getJobState`, `getChildrenValues`, `getJobs(o
 
 Counts (`queue.ts:278`): `getJobCounts()` / `getJobCountsAsync()`, `getWaitingCount`, `getActiveCount`, `getCompletedCount`, `getFailedCount`, `getDelayedCount`, `count()` / `countAsync()`, `getCountsPerPriority()` / `getCountsPerPriorityAsync()`. Sync `count()`/`getCountsPerPriority()` return `0`/`{}` over TCP (`counts.ts:135`, `counts.ts:150`).
 
-Control (`queue.ts:313`): `pause()`, `resume()`, `drain()`, `obliterate()` (all sync, fire-and-forget), `isPaused()` / `isPausedAsync()`, `waitUntilReady()`.
+Control (`queue.ts:313`): `pause()`, `resume()`, `drain()`, `obliterate()` (all sync, fire-and-forget), `pauseAsync()`, `resumeAsync()`, `drainAsync()` (resolves with the removed count), `obliterateAsync()`, `isPaused()` / `isPausedAsync()`, `waitUntilReady()`.
+
+The async control variants resolve only after the server has processed the command; `drainAsync()`/`retryDlqAsync()`/`purgeDlqAsync()` also return the server count that the fire-and-forget forms discard (they always return 0 over TCP).
+
+`obliterateAsync()` resolves only after the server has processed the wipe. The fire-and-forget `obliterate()` gives no ordering guarantee over the multi-connection TCP pool (default 4 sockets, round-robin): a `PUSH` sent right after it can travel on a different socket, be processed first, and then be wiped by the late-arriving obliterate, even with a sleep in between, if the server event loop is busy. Await `obliterateAsync()` before enqueuing follow-up jobs on the same queue (`control.ts:44`).
 
 Management (`queue.ts:336`): `remove(id)` (sync) / `removeAsync(id)`, `retryJob(id)`, `retryJobs(opts?)`, `clean(grace, limit, type?)` / `cleanAsync(...)`, `promoteJobs(opts?)`, `promoteJob(id)`, `updateJobProgress`, `getJobLogs`, `addJobLog`, `clearJobLogs`, `updateJobData`, `changeJobDelay`, `changeJobPriority`, `extendJobLock`.
 
 `promoteJobs({ count? })` delegates to the manager/server bulk operation in both embedded and TCP modes. The operation selects delayed jobs from the live shard queue in stable `(createdAt, id)` order rather than from the eventually consistent SQLite `GetJobs` view; `count: 0` promotes none. Each promotion updates the priority queue, delayed counter/temporal tracking, persisted `run_at`, and queue waiter notification before the call resolves.
 
 Move / BullMQ-v5 (`queue.ts:502`): `moveJobToCompleted`, `moveJobToFailed`, `moveJobToWait`, `moveJobToDelayed`, `moveJobToWaitingChildren`, `waitJobUntilFinished`. `moveJobToFailed(id, error)` forwards the error's stacktrace (#74) and honours `UnrecoverableError` (skip retry) via the shared `failWire` helper, matching the worker failure path — previously both were silently dropped on this and the job-proxy paths. `moveJobToDelayed(id, timestamp)` takes an **absolute** timestamp; embedded routes waiting/active jobs via `changeWaitingDelay`/`changeDelay`, while the TCP path (`jobMove.ts`) sends the `MoveToDelayed` command with a **relative** `delay = max(0, timestamp - now)` (not the raw timestamp) and surfaces a server `ok:false` as a thrown error. Works for both waiting and active jobs.
 
-Stall (`queue.ts:396`): `setStallConfig`, `getStallConfig`, `getStallConfigAsync`. DLQ, rate-limit, scheduler, dedup, dependency, BullMQ-compat (`getPrioritized`, `getWaitingChildren`, …), worker/metrics (`getWorkers`, `getWorkersCount`, `getMetrics`, `trimEvents`), `forward(options)`.
+Stall (`queue.ts:396`): `setStallConfig` / `setStallConfigAsync`, `getStallConfig`, `getStallConfigAsync`. DLQ (`setDlqConfig` / `setDlqConfigAsync`, `getDlq` (embedded-only entries with metadata), `getDlqJobsAsync(count?)` (dead jobs as public Job objects, works over TCP via the `Dlq` command), `retryDlq` / `retryDlqAsync`, `purgeDlq` / `purgeDlqAsync`), rate-limit (`setGlobalRateLimit(max, duration?)` honoring the window in both modes, plus `setGlobalRateLimitAsync` / `removeGlobalRateLimitAsync` / `setGlobalConcurrencyAsync` / `removeGlobalConcurrencyAsync`, and `rateLimit(expireTimeMs)` with broker-side TTL expiry), scheduler, dedup, dependency, BullMQ-compat (`getPrioritized`, `getWaitingChildren`, …), worker/metrics (`getWorkers`, `getWorkersCount`, `getMetrics`, `trimEvents`), `forward(options)`.
 
 Connection: `disconnect()` (flushes + waits for in-flight batcher, then closes) and `close()` (`queue.ts:607`, `queue.ts:616`).
 
@@ -113,7 +117,7 @@ See [data-model](../data-model.md) for full definitions. Key shapes (all in `typ
 - **Durable bypass**: `opts.durable` jobs skip the `AddBatcher` and are sent as individual `PUSH` (immediate disk write) instead of being batched (`queue.ts:216`).
 - **Batcher overflow**: when `pending.length >= maxPending` (default `10000`), the oldest ~10% are spliced and rejected with `"Add buffer overflow - oldest entries dropped"` (`addBatcher.ts:69`). `stop()` rejects all remaining entries with `"AddBatcher stopped"`.
 - **Error propagation**: `add`/`addBulk` throw on `!response.ok`, ensuring the batcher rejects queued callers (e.g. auth failure) rather than resolving them with `undefined` jobs (`add.ts:168`, `add.ts:421`).
-- **Sync-over-TCP no-ops**: `getJobs`/`getWaiting`/… (sync) return `[]`, `count()` returns `0`, `getCountsPerPriority()` returns `{}`, and `isPaused()` returns `false` in TCP mode (each guards with `if (!ctx.embedded) return …`, e.g. `query.ts:222`, `counts.ts:135`, `counts.ts:150`, `control.ts:41`) — silent empties, not errors. Use the `Async` variants for correct TCP results. (`getWorkers()` is the inverse: an async method that returns `[]` in *embedded* mode, since worker registration only exists server-side, `workers.ts:23`.) The same pattern extends to the DLQ and rate-limit surfaces: the sync DLQ getters `getDlq()`/`getDlqStats()` return `[]`/zeroed stats over TCP, and sync `retryDlq()`/`purgeDlq()`/`retryCompleted()` fire the command but return `0` (`dlq.ts:51`; use `retryCompletedAsync()` for the real count). The four rate-limit getters `getGlobalConcurrency()`, `getGlobalRateLimit()`, `getRateLimitTtl()`, `isMaxed()` are stubs that resolve `null`/`null`/`0`/`false` in *both* modes (`rateLimit.ts:33`).
+- **Sync-over-TCP no-ops**: `getJobs`/`getWaiting`/… (sync) return `[]`, `count()` returns `0`, `getCountsPerPriority()` returns `{}`, and `isPaused()` returns `false` in TCP mode (each guards with `if (!ctx.embedded) return …`, e.g. `query.ts:222`, `counts.ts:135`, `counts.ts:150`, `control.ts:41`) — silent empties, not errors. Use the `Async` variants for correct TCP results. (`getWorkers()` is the inverse: an async method that returns `[]` in *embedded* mode, since worker registration only exists server-side, `workers.ts:23`.) The same pattern extends to the DLQ and rate-limit surfaces: the sync DLQ getters `getDlq()`/`getDlqStats()` return `[]`/zeroed stats over TCP, and sync `retryDlq()`/`purgeDlq()`/`retryCompleted()` fire the command but return `0` (`dlq.ts:51`; use `retryDlqAsync()`/`purgeDlqAsync()`/`retryCompletedAsync()` for the real count). The four rate-limit getters `getGlobalConcurrency()`, `getGlobalRateLimit()`, `getRateLimitTtl()`, `isMaxed()` are stubs that resolve `null`/`null`/`0`/`false` in *both* modes (`rateLimit.ts:33`).
 - **TCP-unsupported methods reject explicitly**: `Job.moveToWaitingChildren()` over TCP rejects with "not supported in TCP mode" (`jobProxy.ts:203`, `jobProxy.ts:481`); `queue.moveJobToWaitingChildren` returns `false` over TCP (`jobMove.ts:152`). `removeDeduplicationKey()` always rejects ("no server primitive available") in both modes (`jobProxy.ts:234`, `jobConversion.ts:126`).
 - **Idempotency**: `jobId`/`deduplication.id` make `add` idempotent (custom-id dedup, server-side, retention-window-bounded). `forward()` uses deterministic remote ids (`fwd:<queue>:<localId>`) so re-forwards don't duplicate (see [Store-and-Forward](./store-and-forward.md)).
 - **`retryJob` state machine** (embedded, `management.ts:40`): `failed` → `retryDlq` (throws if not in DLQ), `active` → `moveActiveToWait`, `waiting`/`prioritized`/`delayed` → no-op, anything else throws. TCP path issues `MoveToWait` and throws on `ok !== true`.

docs/features/rate-limiting-and-concurrency.md

@@ -39,9 +39,10 @@ External/runtime:
 
 - `class LimiterManager` (`src/domain/queue/limiterManager.ts:10`)
   - `getState(name): QueueState`, `isPaused(name): boolean`, `pause(name)`, `resume(name)`
-  - `setRateLimit(queue, limit: number): void` — creates `new RateLimiter(limit)`
-  - `clearRateLimit(queue): void`
-  - `tryAcquireRateLimit(queue): boolean` — `true` if no limiter is set
+  - `setRateLimit(queue, limit, durationMs?, ttlMs?): void` — creates `new RateLimiter(limit, limit / ((duration ?? 1000)/1000))`, so the limit means "`limit` per `duration` ms" (default: per second). `ttlMs` stamps `rateLimitExpiresAt = now + ttl` for broker-side auto-expiry. Non-finite / non-positive `duration`/`ttl` degrade to the defaults (1s window, permanent).
+  - `clearRateLimit(queue): void` — also nulls `rateLimitDuration`/`rateLimitExpiresAt`
+  - `expireRateLimitIfNeeded(queue): void` — lazy TTL check; called from the acquire path and from limit reads, so no timer exists and an expired limit can never throttle a pull
+  - `tryAcquireRateLimit(queue): boolean` — `true` if no limiter is set (runs the TTL check first)
   - `setConcurrency(queue, limit): void` — reuses/updates an existing `ConcurrencyLimiter`
   - `clearConcurrency(queue): void`
   - `tryAcquireConcurrency(queue): boolean` — `true` if no limiter is set
@@ -58,21 +59,22 @@ External/runtime:
 
 ### Client SDK (Queue) methods (`src/client/queue/rateLimit.ts`, surfaced in `src/client/queue/queue.ts:438`)
 
-- `setGlobalRateLimit(max: number, duration?: number)` — NOTE: `duration` is **ignored** (`src/client/queue/rateLimit.ts:38`); `max` is sent as the token-bucket capacity (jobs/sec).
-- `removeGlobalRateLimit()`, `setGlobalConcurrency(concurrency)`, `removeGlobalConcurrency()`
-- `rateLimit(expireTimeMs)` — temporary throttle (see Edge Cases).
-- `getGlobalRateLimit()`, `getGlobalConcurrency()`, `getRateLimitTtl()`, `isMaxed()` — all are **stubs** that resolve to `null`/`0`/`false` (`src/client/queue/rateLimit.ts:33`, `:56`, `:75`, `:80`).
+- `setGlobalRateLimit(max: number, duration?: number)` — `max` jobs per `duration` ms (default 1000). The window is honored in both embedded and TCP modes; fire-and-forget over TCP.
+- `setGlobalRateLimitAsync(max, duration?)` / `removeGlobalRateLimitAsync()` / `setGlobalConcurrencyAsync(n)` / `removeGlobalConcurrencyAsync()` — awaitable variants: resolve once the server has applied the change (no set-then-pull race).
+- `removeGlobalRateLimit()`, `setGlobalConcurrency(concurrency)`, `removeGlobalConcurrency()` — fire-and-forget forms.
+- `rateLimit(expireTimeMs)` — temporary throttle (`limit: 1` + broker-side `ttl`); throws on non-finite or non-positive input. The expiry lives on the broker, so it works identically embedded/TCP and survives client exit (see Edge Cases).
+- `getGlobalRateLimit()`, `getGlobalConcurrency()`, `getRateLimitTtl()`, `isMaxed()` — all are **stubs** that resolve to `null`/`0`/`false` (`src/client/queue/rateLimit.ts`).
 
 ### TCP commands (`src/domain/types/command.ts:295`, handlers `src/infrastructure/server/handlers/advanced.ts:239`)
 
-- `RateLimit { queue, limit }` — `limit` validated as finite number, else error `"limit must be a finite number"`.
+- `RateLimit { queue, limit, duration?, ttl? }` — `limit` validated as finite number, else error `"limit must be a finite number"`. `duration` = window ms (default 1000), `ttl` = broker-side auto-expiry ms; invalid values for either degrade to the defaults instead of failing.
 - `RateLimitClear { queue }`
 - `SetConcurrency { queue, limit }` — same finite-number validation.
 - `ClearConcurrency { queue }`
 
 ### HTTP endpoints (`src/infrastructure/server/httpRouteQueueConfig.ts:84`)
 
-- `PUT /queues/:queue/rate-limit` — body `{ limit }` → `RateLimit`
+- `PUT /queues/:queue/rate-limit` — body `{ limit, duration?, ttl? }` → `RateLimit`
 - `DELETE /queues/:queue/rate-limit` → `RateLimitClear`
 - `PUT /queues/:queue/concurrency` — body `{ concurrency }` or `{ limit }` → `SetConcurrency`
 - `DELETE /queues/:queue/concurrency` → `ClearConcurrency`
@@ -175,8 +177,8 @@ Key invariant: **if `limiter.groupKey` is set, the `WorkerRateLimiter` is disabl
 ## Edge Cases & Failure Modes
 
 - **Default = unlimited per queue.** No `RateLimiter`/`ConcurrencyLimiter` exists until explicitly set; `tryAcquire*` returns `true` when absent (`limiterManager.ts:63`, `:90`). The only always-on throttle is the protocol limiter at 10000 req/60s per client (the known "rate limit defaults to Infinity" caveat refers to per-queue limits being off by default).
-- **`setGlobalRateLimit(max, duration)` drops `duration`** (`rateLimit.ts:38`) and the server reinterprets `max` as a token-bucket capacity/refill (jobs/sec) — semantically different from BullMQ's `{max per duration}`. The worker-side `WorkerRateLimiter` is the one that honors `{max, duration}`.
-- **`Queue.rateLimit(expireTimeMs)` asymmetry** (`rateLimit.ts:63`): in embedded mode it sets the queue limit to `1` then auto-clears via `setTimeout(expireTimeMs)`; in TCP mode it sends `RateLimit limit:1` but **never schedules a clear** — the throttle stays at 1 job/sec until manually cleared. Treat TCP-mode temporary rate limit as sticky.
+- **`setGlobalRateLimit(max, duration)` honors `duration` end-to-end** (client → wire `duration` field → `LimiterManager` refill rate), matching BullMQ's `{max per duration}` semantics in both modes. Servers older than 2.8.35 ignore the field and fall back to the 1s bucket. The worker-side `WorkerRateLimiter` independently honors its own `{max, duration}`.
+- **`Queue.rateLimit(expireTimeMs)` expires broker-side** (`rateLimit.ts`): both modes set `limit: 1` with a broker-side `ttl`; there is no client timer, so the expiry survives client exit and behaves identically embedded/TCP. Lazy expiry: the limit clears on the first pull or limit read past the deadline. During the window jobs still trickle at 1/sec (token refill), matching the previous approximation. Invalid `expireTimeMs` (non-finite or ≤ 0) throws. A TTL'd limit persisted to `queue_state` is restored with its remaining time on restart and never resurrects once expired.
 - **Stub getters:** `getGlobalRateLimit`, `getGlobalConcurrency`, `getRateLimitTtl`, `isMaxed` always return `null`/`0`/`false`; do not rely on them to read back limits.
 - **Memory bounds.**
   - `SlidingWindowDeque` advances a head pointer for O(1) amortized expiry and compacts the array when `head > 1000` (`rateLimiter.ts:36`); the cleanup interval deletes empty per-client deques every `cleanupIntervalMs` (`rateLimiter.ts:130`).