From bca26a9408cea04e443808f8af12351e67ae149a Mon Sep 17 00:00:00 2001 From: egeominotti <2592071+egeominotti@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:07:23 +0100 Subject: [PATCH 1/2] test: verify fenced retries and complete job conservation --- docs/features/postgres-multibroker.md | 20 + docs/src/content/docs/changelog.md | 25 +- docs/testing.md | 35 + feature_list.json | 10 + progress.md | 70 +- test/postgres-public-api-queue-worker.test.ts | 42 +- test/postgres-public-delivery-oracle.test.ts | 274 +++++ test/race-conditions-benchmarks.test.ts | 124 +++ test/race-conditions-clients.test.ts | 152 +++ test/race-conditions-contention.test.ts | 110 ++ test/race-conditions-cross-queue.test.ts | 83 ++ test/race-conditions-locks.test.ts | 112 ++ test/race-conditions-mixed.test.ts | 110 ++ test/race-conditions-stalls.test.ts | 127 +++ test/race-conditions.test.ts | 995 ++---------------- ...pro-postgres-public-progress-retry.test.ts | 202 ++++ .../postgres-public-delivery-oracle.ts | 216 ++++ .../support/postgres-public-delivery-trace.ts | 124 +++ test/support/race-conditions-chaos.ts | 118 +++ test/support/race-conditions-clock.ts | 27 + test/support/race-conditions-fixture.ts | 26 + 21 files changed, 2066 insertions(+), 936 deletions(-) create mode 100644 test/postgres-public-delivery-oracle.test.ts create mode 100644 test/race-conditions-benchmarks.test.ts create mode 100644 test/race-conditions-clients.test.ts create mode 100644 test/race-conditions-contention.test.ts create mode 100644 test/race-conditions-cross-queue.test.ts create mode 100644 test/race-conditions-locks.test.ts create mode 100644 test/race-conditions-mixed.test.ts create mode 100644 test/race-conditions-stalls.test.ts create mode 100644 test/repro-postgres-public-progress-retry.test.ts create mode 100644 test/support/postgres-public-delivery-oracle.ts create mode 100644 test/support/postgres-public-delivery-trace.ts create mode 100644 test/support/race-conditions-chaos.ts create mode 100644 test/support/race-conditions-clock.ts create mode 100644 test/support/race-conditions-fixture.ts diff --git a/docs/features/postgres-multibroker.md b/docs/features/postgres-multibroker.md index c7ef8121..a15dd36d 100644 --- a/docs/features/postgres-multibroker.md +++ b/docs/features/postgres-multibroker.md @@ -1403,6 +1403,26 @@ completion-retention suite repeats the authoritative lookup with a zero-sized local result cache. These files are part of the normal `test:postgres` command and therefore run in every PostgreSQL 16–18 CI matrix entry; they are not an opt-in soak. + +The public Queue/Worker lifecycle checker retains every processor invocation's +token, attempt, stage and monotonic start/end timestamps. With no applied failure, +each ID must execute once. A repeated invocation is accepted only after the +specifically covered pre-log Progress error: a matching token-bearing Worker +failure, durable failed/retried events, the next attempt and its failure timeline +must agree. Repeated tokens, overlapping invocations, recovery/release events, +missing predecessors and other processor errors fail the checker. A repeatable-read +snapshot additionally requires one terminal row, one completion tombstone and +one durable completed event per original ID, with the exact result. + +`test/repro-postgres-public-progress-retry.test.ts` holds a real PostgreSQL row +lock after public Worker claim, observes the blocked Progress request, and waits +for its configured SQL lock timeout. It proves rollback at attempt zero, an +applied failure, a new token at attempt one, then one result and log. This +regression distinguishes legal retry from an unexplained duplicate; it does not +identify the cause of historical CI43559077874, whose logs lacked that evidence. +`test/postgres-public-delivery-oracle.test.ts` checks that malformed delivery and +completion histories are rejected. Runtime retry, lease and heartbeat settings +are unchanged. Concurrent brokers claim each job once under both ordinary and 16-consumer high-contention campaigns, stale owners are fenced, broker/client shutdown is safe, remotely renewed leases survive disconnect, shared limits and workers work, missed cron diff --git a/docs/src/content/docs/changelog.md b/docs/src/content/docs/changelog.md index 9d543c5d..170f931f 100644 --- a/docs/src/content/docs/changelog.md +++ b/docs/src/content/docs/changelog.md @@ -236,15 +236,6 @@ head: - Record the completed test-quality work item with its isolated gate totals, controlled-defect and case-preservation proofs, skeptic approvals and green source PR CI evidence. The final record changes documentation and state only. -- Track an intermittent PostgreSQL 17 public-worker processing-count failure - observed during PR validation. Its single failed-job rerun passed; a separate - reproduce-first work item retains the assertion and records the evidence - without claiming a cause or a fix. -- Track a separate intermittent legacy chaos-accounting test failure in unit - CI. Its partial count omits delayed work, but the missing state/operation - trace leaves the observed cause unproved. The single unit-job rerun passed; - a reproduce-first follow-up preserves the oracle and requires full job - conservation rather than a relaxed threshold. - The CPU-intensive workers guide test "a non-yielding handler starves heartbeats and gets recovered" now observes the recovery it claims @@ -530,6 +521,22 @@ head: ### Bug fixes +- Fix the legacy chaos test's incomplete job accounting. Delayed retries were + omitted and an empty pull ended the drain before retries became due, so a + valid 49-completed/51-delayed population failed a half-count threshold. + Seeded concurrent operations now check every original ID, exclusive state + membership and all counters exactly; regressions cover the delayed partition, + a missing retained job and a retry deadline crossing between state reads. + Split the race suites into smaller files with individually owned temporary + SQLite directories. +- Fix the PostgreSQL public Worker test's absolute invocation-count assertion: + a rolled-back Progress command can legitimately trigger the default job retry. + A real SQL row-lock regression now proves that path. Tests require distinct + tokens, sequential attempts, matching applied failures and durable history, + nonoverlapping processing and exactly one committed completion with the right + result. Unexplained repeats still fail with per-attempt diagnostics. This + corrects the oracle; the original intermittent CI cause remains unclassified. + - **Harness command cancellation covers child startup.** A command can start before `spawn()` returns. SIGTERM in that window previously terminated the supervisor before its signal handlers were registered, leaving the detached diff --git a/docs/testing.md b/docs/testing.md index 0c18d173..53513280 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -212,6 +212,41 @@ Run benchmarks only on native macOS, never in an OrbStack Machine or container. Docker Compose is only for functional tests requiring external services and must use a unique project name and disposable volumes. +### Concurrent job conservation + +The race-condition suites use a separate temporary SQLite directory for each +test. `test/race-conditions.test.ts` records 200 operations from a test-owned +seeded generator without changing runtime retry jitter. Conservation compares +every original ID against mutually exclusive waiting, prioritized, delayed, +active, completed, failed/DLQ and waiting-children listings and their counters. +An empty pull does not mean delayed retries have disappeared, and a pull's wait +argument is not a processing timeout. The suite checks the retained population +after all operations settle instead of sleeping briefly and requiring only half +the jobs to appear in a partial count. + +A controlled 51-delayed/49-completed history protects that boundary. A separate +case deliberately removes one original job through the persistence path and +proves that the checker rejects the missing ID even with 99 completions. Set +`BUNQUEUE_RACE_TRACE=1` for the seeded operation/outcome trace, attempts, retry +deadlines and complete state snapshot. The seed reproduces operation choices; +timestamps and completion order remain part of the captured evidence. +The snapshot guard reads retry deadlines independently of the state-filtered +listings: a job becoming due between waiting and delayed reads can be absent +from both. A controlled-clock regression covers that boundary; only a crossed +deadline repeats the snapshot, while assertion failures propagate immediately. + +The PostgreSQL public Worker lifecycle test correlates tokens, attempts, applied +Worker failures and durable transition events. Its absolute processor-once +assertion was invalid when a Progress command rolled back and the default retry +policy applied. The replacement requires one invocation when no failure applied, +and permits the covered pre-log Progress retry only with a complete predecessor +history. Distinct tokens, nonoverlap, contiguous attempts, exact results and one +durable completion per ID are mandatory. A real row-lock/SQL-timeout regression +proves the retry; checker mutation cases reject unexplained repeats and duplicate +completions. See [PostgreSQL validation](features/postgres-multibroker.md#validation). +The specific historical CI failure remains unclassified because its original +logs did not retain the necessary per-attempt evidence. + ### Behavioral test oracles and controlled observation Removing a case needs a coverage map: the surviving test must use the same diff --git a/feature_list.json b/feature_list.json index f63df079..37dff1a8 100644 --- a/feature_list.json +++ b/feature_list.json @@ -32,6 +32,16 @@ "evidence": "2026-11-13; lands in the commit that adds this entry, rebased onto 7bc922db. bun run test:sandbox green on the gated tree bbd565ca: unit 11746 / 0 / 399, tcp 553 / 0, embedded 349 / 0, no anomalies (agent-skills worktree, artifacts/test-sandbox/2026-11-13T00-20-42.544Z/); the skills and harness contract tests run inside the unit container. Earlier runs: green on 4e9adf35 (unit 11720 / 0 / 398); on daf7db88 one unrelated failure, test/cronScheduler.test.ts:819 at 23:59:13 UTC (work item cron-test-midnight-boundary). Host: bun test test/agent-skills test/agent-harness 99 / 0, bun run typecheck and bun run format:check clean, ./init.sh green; every script passes bash -n, --help and --dry-run under bash 3.2. After the rebase, bun run test:sandbox green again on the rebased tree 871688b7 (unit 11746 / 0 / 399, tcp 553 / 0, embedded 349 / 0, no anomalies; artifacts/test-sandbox/2026-11-13T00-32-55.365Z/); this sentence is the only change after that run. The rebase adds only the files of 7bc922db (release metadata refresh, itself sandbox-gated) and merges the changelog and state files. Skeptic: FAIL on 4e9adf35 (Machines left behind after an interrupt, renames hidden from gate-delta.sh), CONDITIONAL on daf7db88 (late create recorded as deleted, orb list under pipefail, reaped pids), all fixed with tests, then PASS on bbd565ca. bun run test:model and bun run test:sandbox:sdk not triggered (no src/ or sdk/ change). Not yet done, tracked for the next item: the trigger evaluations have not been run against real agents, and machine-gate.sh has only run against a fake OrbStack and recorded logs.", "issue": null }, + { + "id": "ci-delivery-accounting-fixes", + "name": "Resolve PostgreSQL delivery and chaos accounting CI anomalies", + "description": "Investigate and resolve postgres-public-api-processing-flake and race-chaos-accounting-flake with isolated deterministic reproductions, complete job-state evidence and regression coverage. Preserve processing exclusivity and full conservation assertions; distinguish fixture defects from engine defects before changing behavior.", + "dependencies": [], + "status": "in-progress", + "branch": "fix-ci-delivery-accounting", + "evidence": "PARTIAL 2026-11-14, base42225c31, branchfix-ci-delivery-accounting; code implemented, exact-head CI pending. Real SQLite RED with original default backoff/waits/drain: all100IDs49completed+51delayed, old>=50 fails49; strict conservation GREEN. Read-clock boundary RED expected1got0, fixed by independent retry deadlines; missing-job mutation rejected. All18original race paths/options retained,17other bodies token-identical; final native family21/0/564, focused Linux17/0/559. Real PG17 row-lock RED proves applied Progress-failure retry token0/attempt0 -> newtoken/attempt1, result14/log1, durablefailed1/completed1, then oldonce1fails2. Strict checker21cases (19negative/2positive). FINAL owned-Linux PG16.16/17.12/18.7 each571pass/0fail/1opt-insoakskip,572tests126files. Artifacts artifacts/ci-delivery-accounting/. Historical CI43559077874 cause remains unclassified; no runtime/lease/SDK changes. Full bun run test:sandbox PASS on treee2c50d5737b2dff3d2efc2c4ad3e96318cb54a4a: unit11931/0/404,TCP553/0(69suites),embedded349/0(43),0errors/anomalies/OOM,423validsamples,6.79min; artifacts/test-sandbox/2026-11-14T10-45-13.122Z/. Fresh native Ubuntu24.04.5arm64 Machine gate PASS same tree: Bun1.4.4+38afee6c8,4CPU/16GiB,unit12062/0/404,TCP553/0,embedded349/0,tinyfs9/0,type890+2163/format2373,15m40s,matching snapshot/logs copied/Machine deleted; artifacts/machines/20261114T104513Z/manifest.md. ./init.sh green11s. Model/SDK standalone gates not triggered (tests/docs/state only; model runs inside unit gate). Complete21-file skeptic review found no blocking code issue; final evidence-delta approval/commit/push/CI remain pending.", + "issue": null + }, { "id": "ci-python39-runner-pin", "name": "Drop Python 3.9 from the Python SDK and remove the runner pin", diff --git a/progress.md b/progress.md index 35b72fe3..3d49a220 100644 --- a/progress.md +++ b/progress.md @@ -22,7 +22,8 @@ overview: [docs/agents/README.md](docs/agents/README.md). tree `f3c1bdf4`. Source PR #191 CI [43557827964](https://github.com/egeominotti/bunqueue/actions/runs/43557827964) attempt1 GREEN, unit12042/0/399 and all product/PostgreSQL16–18/SDK/docs/type/ - format jobs passed. Final state-record CI and authorized integration follow. + format jobs passed. PR #191 merged as `42225c31`, main CI43563916357 GREEN; + the original main checkout was aligned. - `harness-audit-hardening`, `postgres-runner-guard-destructuring` and `machine-gate-interrupt-log-gaps`: integrated in `9b69dcba` + `45790ac7`; full sandbox, fresh Ubuntu and source CI run `43526774536` GREEN. Final @@ -61,15 +62,23 @@ overview: [docs/agents/README.md](docs/agents/README.md). ### What's In Progress +- `ci-delivery-accounting-fixes`, branch `fix-ci-delivery-accounting`, starts + from merged main `42225c31`. Both oracle defects have isolated RED/GREEN + evidence; strict delivery history and complete conservation checks are + implemented. PostgreSQL16–18, full sandbox and fresh Ubuntu gates passed; + final skeptic evidence-delta approval, commit/push and exact-head CI are next. + The three audit implementations are done, backed by green source CI. Their original staged worktrees remain preserved. Their integration is complete; other sessions' active work remains tracked in `feature_list.json`. ### What's Next -1. Validate/push the final CI-flake tracking record, watch its exact-head PR #191 - CI, perform the authorized normal squash merge, then watch main CI and - fast-forward the original main checkout. No version/publication. +1. Finish `ci-delivery-accounting-fixes`: review the complete PostgreSQL16–18 + results, sandbox and fresh Ubuntu gates, obtain the mandatory skeptic PASS + before each authorized commit/push, then watch exact-head CI and integrate. + PR #191 already merged as `42225c31`; its main CI43563916357 is green and + the original main checkout was aligned. No version/publication action. 2. `postgres-completion-recheck-state-index`: the prune recheck's consumer-state probe on PostgreSQL 16 and 17 with stale statistics. 3. Owner decisions: `postgres-completion-read-model-plan` (trade-off measured, @@ -78,17 +87,14 @@ other sessions' active work remains tracked in `feature_list.json`. ## Blockers / Risks -- PR #191's metadata commit `1037ddd3` had a legacy random chaos-accounting - failure in Unit Tests, run43560970951 attempt1 (49 counted, expected at least - 50). Its single unit-job rerun passed in attempt2, watcher exit3 FLAKY. - `race-chaos-accounting-flake` is not-started; delayed work is omitted from - that partial count, but actual CI state/IDs/seed were not captured. Preserve - the oracle and reproduce before asserting a cause or changing behavior. -- PR #191's record commit `dd147ccf` had one PostgreSQL 17 processing-count - failure in run43559077874 attempt1; the single failed-job rerun passed in - attempt2 (`ci-watch.sh` exit3 FLAKY). The new not-started item - `postgres-public-api-processing-flake` requires reproduction and per-job - failure/lease diagnostics; the cause is unproved and the assertion remains. +- The legacy partial-count predicate from CI43560970951 has been reproduced + with all100IDs retained (49completed/51delayed), then replaced with exact + conservation checks under `ci-delivery-accounting-fixes`; its source CI is + pending. The original CI snapshot/seed were not retained. +- The PostgreSQL invocation-count oracle from CI43559077874 rejects a proven + legal Progress-failure retry. Strict token/attempt/durable-history checks now + distinguish that path from unexplained repeats, but the specific historical + CI cause remains unclassified because per-attempt evidence was absent. - `main` requires pull requests; only an account with bypass rights can push to it directly. The release push (CI's version gate tags, releases and publishes from `main`) therefore needs an account with write access to `main`, or a pull @@ -140,6 +146,40 @@ other sessions' active work remains tracked in `feature_list.json`. Newest first. Prepend one entry per session; for multi-session work fill the [`session-handoff.md`](session-handoff.md) sections into the entry. +### 2026-11-14 — delivery and accounting oracle defects reproduced + +- User requested investigation and fixes for both recorded CI anomalies, within + the tests-only scope. Existing commit/push/merge authorization and use of + `egeominotti` persist. Branch `fix-ci-delivery-accounting`, base main42225c31; + managed worktree reused, original checkout and prior branch/artifacts preserved. +- Chaos RED on isolated Ubuntu24.04arm64/Bun1.4.4: the exact old `>=50` predicate + receives49 despite all100IDs retained as49completed+51delayed. Reproduced with + both controlled long backoff and the original default-backoff/100ms wait/ + first-empty200ms pull. Seeded operations now enforce all-state/per-ID + conservation; a missing-job mutation and a deadline-crossing regression guard + the checker. All18original cases remain,17other bodies token-identical. +- PostgreSQL17 controlled RED: a real row lock blocks Progress until SQL timeout, + then matching applied FAIL/new-token attempt1 completes once with result14 and + one log; the old invocation-count assertion rejects the two legal attempts. + Strict histories retain token/attempt/stage/timing and durable failure/retry/ + completion evidence;21 checker cases include19 negative and2 positive cases. The specific + historical CI cause remains unknown;150 spontaneous24-job rounds had no repeat. +- Focused Linux race validation17/0/559assertions. Final PostgreSQL16.16/17.12/ + 18.7 matrices each571/0/1skip,572tests126files, on the final monotonic trace. + Independent review caught the race snapshot guard gap; its RED/GREEN regression + and fix are included. Sandbox PASS on treee2c50d57:11931/0/404unit,553/0TCP, + 349/0embedded,0errors/anomalies/OOM,6.79min,423validresource samples. Fresh + Ubuntu24.04.5arm64/Bun1.4.4+38afee6c8/4CPU16GiB PASS same tree:12062/0/404unit, + 553/0TCP,349/0embedded,tinyfs9/0,type890+2163/format2373,15m40s; all logs + copied and Machine deleted. Manifests under artifacts/test-sandbox/ + 2026-11-14T10-45-13.122Z/ and artifacts/machines/20261114T104513Z/. + Final evidence-only delta, skeptic approval and source CI remain pending. +- Artifacts: `artifacts/ci-delivery-accounting/`; current ledger/reviews in + `.superpowers/sdd/2026-11-14-ci-delivery-accounting/`. Development Machine is + owned and isolated; its initial diagnostic full runs had setup failures + (remote Bun PATH; test-role CREATEDB permission), corrected before final runs. + Runtime, SDKs, retry/lease settings and publication state are unchanged. + ### 2026-11-14 — unseeded chaos accounting finding tracked - CI43560970951 on metadata-only `1037ddd38f4e52da51dc461ef62b5d22a0122608` diff --git a/test/postgres-public-api-queue-worker.test.ts b/test/postgres-public-api-queue-worker.test.ts index f4f2c579..c0c1e847 100644 --- a/test/postgres-public-api-queue-worker.test.ts +++ b/test/postgres-public-api-queue-worker.test.ts @@ -1,6 +1,12 @@ import { afterAll, afterEach, beforeAll, describe, expect, test } from 'bun:test'; -import { PostgresPublicApiHarness, waitFor } from './support/postgres-public-api-harness'; +import { + PostgresPublicApiHarness, + waitFor, + POSTGRES_PUBLIC_API_CLIENT_CLOSE_TIMEOUT_MS, + POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS, +} from './support/postgres-public-api-harness'; import { expectedPostgresVersionPrefix } from './support/postgres-version'; +import { PublicDeliveryTrace } from './support/postgres-public-delivery-trace'; const postgresUrl = Bun.env.BUNQUEUE_TEST_POSTGRES_URL; let harness: PostgresPublicApiHarness | null = null; @@ -8,16 +14,16 @@ let harness: PostgresPublicApiHarness | null = null; beforeAll(async () => { if (!postgresUrl) return; harness = await PostgresPublicApiHarness.start(postgresUrl, expectedPostgresVersionPrefix); -}); +}, POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS); afterEach(async () => { await harness?.closeClients(); -}); +}, POSTGRES_PUBLIC_API_CLIENT_CLOSE_TIMEOUT_MS); afterAll(async () => { await harness?.close(); harness = null; -}); +}, POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS); function fixture(): PostgresPublicApiHarness { if (!harness) throw new Error('PostgreSQL public API harness is not running'); @@ -34,6 +40,7 @@ describe(`PostgreSQL ${expectedPostgresVersionPrefix} public Queue and Worker AP const observer = current.queue<{ value: number }>(name, 3); const events = current.queueEvents(name, 2); const executions = new Map(); + const trace = new PublicDeliveryTrace(); const waiting = new Set(); const completed = new Map(); const progressed = new Set(); @@ -47,13 +54,23 @@ describe(`PostgreSQL ${expectedPostgresVersionPrefix} public Queue and Worker AP name, 1, async (job) => { + const invocation = trace.start(job); executions.set(job.id, (executions.get(job.id) ?? 0) + 1); - await job.updateProgress({ phase: 'running' }); - await job.log(`processed-${job.data.value}`); - return job.data.value * 2; + try { + await job.updateProgress({ phase: 'running' }); + invocation.stage = 'log'; + await job.log(`processed-${job.data.value}`); + return trace.returned(invocation, job.data.value * 2); + } catch (error) { + trace.threw(invocation, error); + throw error; + } }, { concurrency: 8 } ); + worker.on('failed', (job, error) => trace.failed(job, error)); + worker.on('error', (error) => trace.unexpected('error', error)); + worker.on('stalled', (id) => trace.unexpected('stalled', id)); await worker.waitUntilReady(); const jobs = await producer.addBulk( @@ -74,7 +91,12 @@ describe(`PostgreSQL ${expectedPostgresVersionPrefix} public Queue and Worker AP expect(progressed).toEqual(ids); expect(new Set(completed.keys())).toEqual(ids); expect(executions.size).toBe(jobs.length); - expect([...executions.values()].every((count) => count === 1)).toBe(true); + await trace.assert( + postgresUrl!, + current.namespace, + name, + new Map(jobs.map((job, value) => [job.id, value * 2])) + ); expect(await jobs[7].waitUntilFinished(events, 5_000)).toBe(14); expect(await observer.getJobState(jobs[7].id)).toBe('completed'); expect(await observer.getJobCountsAsync()).toMatchObject({ @@ -82,6 +104,10 @@ describe(`PostgreSQL ${expectedPostgresVersionPrefix} public Queue and Worker AP completed: jobs.length, failed: 0, waiting: 0, + delayed: 0, + prioritized: 0, + paused: 0, + 'waiting-children': 0, }); let retained = await observer.getJobsAsync({ asc: true, end: -1, state: 'completed' }); diff --git a/test/postgres-public-delivery-oracle.test.ts b/test/postgres-public-delivery-oracle.test.ts new file mode 100644 index 00000000..0332fa45 --- /dev/null +++ b/test/postgres-public-delivery-oracle.test.ts @@ -0,0 +1,274 @@ +import { expect, test } from 'bun:test'; +import { + assertPublicDeliveryHistory, + type DeliveryHistory, +} from './support/postgres-public-delivery-oracle'; + +// Independent, explicit DTO history: no production encoder or collector builds the oracle's input. +function validRetry(): DeliveryHistory { + return { + invocations: [ + { + id: 'job', + token: 'first', + attempt: 0, + start: 1, + end: 2, + stage: 'progress', + outcome: 'threw', + error: 'Error: Internal server error', + }, + { + id: 'job', + token: 'second', + attempt: 1, + start: 4, + end: 5, + stage: 'return', + outcome: 'returned', + result: 14, + }, + ], + failures: [{ id: 'job', token: 'first', at: 3, error: 'Error: Internal server error' }], + unexpected: [], + events: [ + { + id: 10, + jobId: 'job', + type: 'pulled', + at: 100, + attempt: 0, + timeline: [{ state: 'active', timestamp: 100 }], + }, + { id: 11, jobId: 'job', type: 'retried', at: 110, attempt: 1, timeline: [] }, + { + id: 12, + jobId: 'job', + type: 'failed', + at: 110, + attempt: 1, + error: 'Internal server error', + timeline: [], + }, + { + id: 13, + jobId: 'job', + type: 'pulled', + at: 120, + attempt: 1, + timeline: [ + { state: 'active', timestamp: 100 }, + { state: 'failed', timestamp: 110, attempt: 1, error: 'Internal server error' }, + { state: 'delayed', timestamp: 110, attempt: 2 }, + { state: 'active', timestamp: 120 }, + ], + }, + { id: 14, jobId: 'job', type: 'completed', at: 130, attempt: 1, result: 14, timeline: [] }, + ], + jobs: [ + { + id: 'job', + state: 'completed', + attempt: 1, + result: 14, + timeline: [ + { state: 'active', timestamp: 100 }, + { state: 'failed', timestamp: 110, attempt: 1, error: 'Internal server error' }, + { state: 'delayed', timestamp: 110, attempt: 2 }, + { state: 'active', timestamp: 120 }, + { state: 'completed', timestamp: 130 }, + ], + }, + ], + completions: [{ id: 'job', result: 14 }], + }; +} + +const expected = new Map([['job', 14]]); + +test('accepts only a fully evidenced pre-log Progress failure retry', () => { + expect(() => assertPublicDeliveryHistory(validRetry(), expected)).not.toThrow(); +}); + +test('requires one invocation when no failure occurred', () => { + const healthy: DeliveryHistory = { + invocations: [ + { + id: 'job', + token: 'only', + attempt: 0, + start: 1, + end: 2, + stage: 'return', + outcome: 'returned', + result: 14, + }, + ], + failures: [], + unexpected: [], + events: [ + { id: 10, jobId: 'job', type: 'pulled', at: 100, attempt: 0, timeline: [] }, + { id: 11, jobId: 'job', type: 'completed', at: 110, attempt: 0, result: 14, timeline: [] }, + ], + jobs: [ + { + id: 'job', + state: 'completed', + attempt: 0, + result: 14, + timeline: [ + { state: 'active', timestamp: 100 }, + { state: 'completed', timestamp: 110 }, + ], + }, + ], + completions: [{ id: 'job', result: 14 }], + }; + assertPublicDeliveryHistory(healthy, expected); + healthy.invocations.push({ ...healthy.invocations[0], token: 'unexplained', start: 3, end: 4 }); + expect(() => assertPublicDeliveryHistory(healthy, expected)).toThrow('invocation/pull mismatch'); +}); + +const invalid: [string, (history: DeliveryHistory) => void, string][] = [ + [ + 'same-token repeat', + (h) => { + h.invocations[1].token = 'first'; + }, + 'repeated delivery token', + ], + [ + 'empty token', + (h) => { + h.invocations[0].token = ''; + }, + 'Missing or repeated delivery token', + ], + [ + 'unexplained new-token delivery', + (h) => { + h.events = h.events.filter((e) => e.type !== 'failed' && e.type !== 'retried'); + }, + 'missing durable failure', + ], + [ + 'missing applied Worker failure', + (h) => { + h.failures = []; + }, + 'applied Worker failure', + ], + [ + 'wrong Worker failure token', + (h) => { + h.failures[0].token = 'other'; + }, + 'applied Worker failure', + ], + [ + 'failure observed after next invocation', + (h) => { + h.failures[0].at = 6; + }, + 'applied Worker failure', + ], + [ + 'skipped processor attempt', + (h) => { + h.invocations[1].attempt = 2; + }, + 'noncontiguous attempt', + ], + [ + 'wrong durable attempt', + (h) => { + h.events[3].attempt = 0; + }, + 'noncontiguous attempt', + ], + [ + 'overlapping processor invocations', + (h) => { + h.invocations[0].end = 5; + }, + 'overlapping processor invocations', + ], + [ + 'durable failure after next pull', + (h) => { + h.events[2].id = 20; + }, + 'durable failure must precede next pull', + ], + [ + 'missing failure timeline', + (h) => { + h.events[3].timeline = []; + }, + 'durable failure timeline', + ], + [ + 'duplicate durable completion', + (h) => { + h.events.push({ ...h.events[4], id: 15 }); + }, + 'duplicate or missing durable completion', + ], + [ + 'error after logging begins', + (h) => { + h.invocations[0].stage = 'log'; + }, + 'unexpected processor failure or stage', + ], + [ + 'arbitrary processor error', + (h) => { + h.invocations[0].error = 'Error: unexpected'; + }, + 'unexpected processor failure or stage', + ], + [ + 'stalled recovery', + (h) => { + h.unexpected.push({ kind: 'stalled' }); + }, + 'Unexpected Worker error', + ], + [ + 'durable recovery', + (h) => { + h.events.splice(1, 0, { ...h.events[1], type: 'stalled' }); + }, + 'unexpected durable transition', + ], + [ + 'wrong terminal result', + (h) => { + h.jobs[0].result = 99; + }, + 'incorrect terminal job', + ], + [ + 'wrong completion result', + (h) => { + h.completions[0].result = 99; + }, + 'incorrect durable completion result', + ], + [ + 'missing terminal job', + (h) => { + h.jobs = []; + }, + 'Terminal job count mismatch', + ], +]; + +for (const [name, corrupt, message] of invalid) { + test(`rejects ${name}`, () => { + const history = validRetry(); + corrupt(history); + expect(() => assertPublicDeliveryHistory(history, expected)).toThrow(message); + }); +} diff --git a/test/race-conditions-benchmarks.test.ts b/test/race-conditions-benchmarks.test.ts new file mode 100644 index 00000000..8e86db59 --- /dev/null +++ b/test/race-conditions-benchmarks.test.ts @@ -0,0 +1,124 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Benchmarks', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('BENCHMARK: 1000 concurrent push operations', async () => { + const COUNT = 1000; + + const start = performance.now(); + const pushes = Array.from({ length: COUNT }, (_, i) => qm.push('bench-push', { data: { i } })); + await Promise.all(pushes); + const duration = performance.now() - start; + + console.log(`\n📊 1000 concurrent pushes: ${duration.toFixed(2)}ms`); + console.log(` Throughput: ${(COUNT / (duration / 1000)).toFixed(0)} ops/sec`); + + const stats = qm.getStats(); + expect(stats.waiting).toBe(COUNT); + }); + + test('BENCHMARK: 500 concurrent pull operations', async () => { + const COUNT = 500; + + // First push + for (let i = 0; i < COUNT; i++) { + await qm.push('bench-pull', { data: { i } }); + } + + const start = performance.now(); + const pulls = Array.from({ length: COUNT }, () => qm.pull('bench-pull', 100)); + const results = await Promise.all(pulls); + const duration = performance.now() - start; + + const pulled = results.filter((r) => r !== null).length; + + console.log(`\n📊 500 concurrent pulls: ${duration.toFixed(2)}ms`); + console.log(` Pulled: ${pulled} jobs`); + console.log(` Throughput: ${(pulled / (duration / 1000)).toFixed(0)} ops/sec`); + + expect(pulled).toBe(COUNT); + }); + + test('BENCHMARK: Mixed concurrent operations (push/pull/ack)', async () => { + const ITERATIONS = 300; + + const start = performance.now(); + const operations: Promise[] = []; + + for (let i = 0; i < ITERATIONS; i++) { + // Push + operations.push(qm.push('bench-mixed', { data: { i } })); + + // Pull and ack + operations.push( + qm.pull('bench-mixed', 50).then(async (job) => { + if (job) { + await qm.ack(job.id, {}); + return true; + } + return false; + }) + ); + } + + await Promise.all(operations); + const duration = performance.now() - start; + + console.log(`\n📊 ${ITERATIONS * 2} mixed operations: ${duration.toFixed(2)}ms`); + console.log(` Throughput: ${((ITERATIONS * 2) / (duration / 1000)).toFixed(0)} ops/sec`); + + // Clean up remaining + while (true) { + const job = await qm.pull('bench-mixed', 10); + if (!job) break; + await qm.ack(job.id, {}); + } + + const stats = qm.getStats(); + expect(stats.waiting).toBe(0); + expect(stats.active).toBe(0); + }); + + test('BENCHMARK: Client disconnect/release under load', async () => { + const CLIENTS = 20; + const JOBS_PER_CLIENT = 50; + + // Setup: push and pull jobs for each client + for (let c = 0; c < CLIENTS; c++) { + for (let j = 0; j < JOBS_PER_CLIENT; j++) { + await qm.push('bench-release', { data: { c, j } }); + } + } + + for (let c = 0; c < CLIENTS; c++) { + for (let j = 0; j < JOBS_PER_CLIENT; j++) { + const job = await qm.pull('bench-release', 0); + if (job) { + qm.registerClientJob(`client-${c}`, job.id); + } + } + } + + const start = performance.now(); + const releases = Array.from({ length: CLIENTS }, (_, c) => qm.releaseClientJobs(`client-${c}`)); + const results = await Promise.all(releases); + const duration = performance.now() - start; + + const totalReleased = results.reduce((a, b) => a + b, 0); + + console.log(`\n📊 ${CLIENTS} concurrent client releases: ${duration.toFixed(2)}ms`); + console.log(` Total jobs released: ${totalReleased}`); + console.log(` Throughput: ${(totalReleased / (duration / 1000)).toFixed(0)} jobs/sec`); + + expect(totalReleased).toBe(CLIENTS * JOBS_PER_CLIENT); + }); +}); diff --git a/test/race-conditions-clients.test.ts b/test/race-conditions-clients.test.ts new file mode 100644 index 00000000..29025258 --- /dev/null +++ b/test/race-conditions-clients.test.ts @@ -0,0 +1,152 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import type { JobId } from '../src/domain/types/job'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('1. Concurrent Client Disconnections (releaseClientJobs)', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should safely release jobs when multiple clients disconnect simultaneously', async () => { + const CLIENTS = 10; + const JOBS_PER_CLIENT = 20; + + // Push jobs for each "client" + const clientJobs: Map = new Map(); + + for (let c = 0; c < CLIENTS; c++) { + const clientId = `client-${c}`; + clientJobs.set(clientId, []); + + for (let j = 0; j < JOBS_PER_CLIENT; j++) { + const job = await qm.push('race-queue', { + data: { client: c, job: j }, + }); + clientJobs.get(clientId)!.push(job.id); + } + } + + // Pull jobs and register them with clients + for (let c = 0; c < CLIENTS; c++) { + const clientId = `client-${c}`; + for (let j = 0; j < JOBS_PER_CLIENT; j++) { + const job = await qm.pull('race-queue', 0); + if (job) { + qm.registerClientJob(clientId, job.id); + } + } + } + + let stats = qm.getStats(); + expect(stats.active).toBe(CLIENTS * JOBS_PER_CLIENT); + expect(stats.waiting).toBe(0); + + // Simulate all clients disconnecting simultaneously + const releasePromises = Array.from({ length: CLIENTS }, (_, c) => + qm.releaseClientJobs(`client-${c}`) + ); + + const results = await Promise.all(releasePromises); + + // Each client should have released their jobs + const totalReleased = results.reduce((a, b) => a + b, 0); + expect(totalReleased).toBe(CLIENTS * JOBS_PER_CLIENT); + + // All jobs should be back in waiting state + stats = qm.getStats(); + expect(stats.active).toBe(0); + expect(stats.waiting).toBe(CLIENTS * JOBS_PER_CLIENT); + + // Verify jobs can be pulled again + let pulledCount = 0; + for (let i = 0; i < CLIENTS * JOBS_PER_CLIENT; i++) { + const job = await qm.pull('race-queue', 0); + if (job) pulledCount++; + } + expect(pulledCount).toBe(CLIENTS * JOBS_PER_CLIENT); + }); + + test('should handle interleaved pull and release operations', async () => { + const ITERATIONS = 50; + + // Push initial jobs + for (let i = 0; i < ITERATIONS; i++) { + await qm.push('interleave-queue', { data: { i } }); + } + + const operations: Promise[] = []; + const processedJobs: Set = new Set(); + + // Interleave pull, ack, and release operations + for (let i = 0; i < ITERATIONS; i++) { + const clientId = `client-${i % 5}`; + + // Pull operation with client registration + operations.push( + qm.pull('interleave-queue', 50).then(async (job) => { + if (job) { + processedJobs.add(String(job.id)); + qm.registerClientJob(clientId, job.id); + // Randomly ack or let release handle it + if (Math.random() > 0.5) { + await qm.ack(job.id, { result: 'done' }); + qm.unregisterClientJob(clientId, job.id); + return 'acked'; + } + return 'pulled'; + } + return 'empty'; + }) + ); + + // Occasionally release client jobs + if (i % 10 === 9) { + operations.push(qm.releaseClientJobs(clientId).then((count) => `released-${count}`)); + } + } + + await Promise.all(operations); + + // Should not have lost any jobs - all should be completed, waiting, or active + const stats = qm.getStats(); + expect(stats.waiting + stats.active + stats.completed).toBe(ITERATIONS); + }); + + test('should not duplicate jobs when releasing under contention', async () => { + const JOBS = 100; + const CLIENT = 'contention-client'; + + // Push and pull all jobs + for (let i = 0; i < JOBS; i++) { + await qm.push('contention-queue', { data: { i } }); + } + + for (let i = 0; i < JOBS; i++) { + const job = await qm.pull('contention-queue', 0); + if (job) { + qm.registerClientJob(CLIENT, job.id); + } + } + + // Try to release multiple times concurrently + const releaseAttempts = Array.from({ length: 10 }, () => qm.releaseClientJobs(CLIENT)); + + const results = await Promise.all(releaseAttempts); + + // Only the first release should have released jobs + const totalReleased = results.reduce((a, b) => a + b, 0); + expect(totalReleased).toBe(JOBS); + + // Exactly JOBS should be in waiting state + const stats = qm.getStats(); + expect(stats.waiting).toBe(JOBS); + expect(stats.active).toBe(0); + }); + }); +}); diff --git a/test/race-conditions-contention.test.ts b/test/race-conditions-contention.test.ts new file mode 100644 index 00000000..09dc7513 --- /dev/null +++ b/test/race-conditions-contention.test.ts @@ -0,0 +1,110 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import type { JobId } from '../src/domain/types/job'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('5. High Concurrency Lock Contention', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should handle 100 concurrent pulls from same queue', async () => { + const JOBS = 100; + const QUEUE = 'contention-pull-queue'; + + // Push all jobs + for (let i = 0; i < JOBS; i++) { + await qm.push(QUEUE, { data: { i } }); + } + + // 100 concurrent pulls + const pulls = Array.from({ length: JOBS }, () => qm.pull(QUEUE, 100)); + + const results = await Promise.all(pulls); + const pulledJobs = results.filter((r) => r !== null); + const pulledIds = new Set(pulledJobs.map((j) => String(j!.id))); + + // Each job should be pulled exactly once (no duplicates) + expect(pulledIds.size).toBe(JOBS); + expect(pulledJobs.length).toBe(JOBS); + }); + + test('should handle concurrent acks for different jobs', async () => { + const JOBS = 50; + const QUEUE = 'contention-ack-queue'; + + // Push and pull jobs with locks + const jobs: Array<{ id: JobId; token?: string }> = []; + for (let i = 0; i < JOBS; i++) { + await qm.push(QUEUE, { data: { i } }); + const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); + if (job) { + jobs.push({ id: job.id, token: token ?? undefined }); + } + } + + // Concurrent acks + const acks = jobs.map((job) => + qm.ack(job.id, { result: 'done' }, job.token).then( + () => ({ success: true, id: job.id }), + (e) => ({ success: false, id: job.id, error: e.message }) + ) + ); + + const results = await Promise.all(acks); + const successful = results.filter((r) => r.success); + + expect(successful.length).toBe(JOBS); + + const stats = qm.getStats(); + expect(stats.completed).toBe(JOBS); + expect(stats.active).toBe(0); + }); + + test('should handle rapid push/pull cycles without losing jobs', async () => { + const CYCLES = 100; + const QUEUE = 'rapid-cycle-queue'; + const processedIds: Set = new Set(); + const errors: string[] = []; + + // Rapid push/pull/ack cycles + const cycles = Array.from({ length: CYCLES }, async (_, i) => { + try { + await qm.push(QUEUE, { data: { cycle: i } }); + const pulled = await qm.pull(QUEUE, 100); + if (pulled) { + processedIds.add(String(pulled.id)); + await qm.ack(pulled.id, { result: i }); + } + return true; + } catch (e) { + errors.push(String(e)); + return false; + } + }); + + const results = await Promise.all(cycles); + const successCount = results.filter((r) => r).length; + + // Most cycles should succeed + expect(successCount).toBeGreaterThan(CYCLES * 0.8); + expect(errors.length).toBeLessThan(CYCLES * 0.2); + + // Clean up any remaining jobs + while (true) { + const job = await qm.pull(QUEUE, 10); + if (!job) break; + await qm.ack(job.id, {}); + } + + const stats = qm.getStats(); + expect(stats.waiting).toBe(0); + expect(stats.active).toBe(0); + }); + }); +}); diff --git a/test/race-conditions-cross-queue.test.ts b/test/race-conditions-cross-queue.test.ts new file mode 100644 index 00000000..84fdffc3 --- /dev/null +++ b/test/race-conditions-cross-queue.test.ts @@ -0,0 +1,83 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('6. Cross-Queue Concurrent Operations', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should handle operations across 10 queues simultaneously', async () => { + const QUEUES = Array.from({ length: 10 }, (_, i) => `multi-queue-${i}`); + const JOBS_PER_QUEUE = 20; + + // Push to all queues concurrently + const pushPromises: Promise[] = []; + for (const queue of QUEUES) { + for (let j = 0; j < JOBS_PER_QUEUE; j++) { + pushPromises.push(qm.push(queue, { data: { queue, job: j } })); + } + } + await Promise.all(pushPromises); + + let stats = qm.getStats(); + expect(stats.waiting).toBe(QUEUES.length * JOBS_PER_QUEUE); + + // Pull and ack from all queues concurrently + const processPromises: Promise[] = []; + for (const queue of QUEUES) { + for (let j = 0; j < JOBS_PER_QUEUE; j++) { + processPromises.push( + qm.pull(queue, 100).then(async (job) => { + if (job) { + await qm.ack(job.id, { result: 'done' }); + return true; + } + return false; + }) + ); + } + } + + const results = await Promise.all(processPromises); + const processed = results.filter((r) => r).length; + + expect(processed).toBe(QUEUES.length * JOBS_PER_QUEUE); + + stats = qm.getStats(); + expect(stats.completed).toBe(QUEUES.length * JOBS_PER_QUEUE); + expect(stats.waiting).toBe(0); + expect(stats.active).toBe(0); + }); + + test('should handle drain during active processing', async () => { + const QUEUE = 'drain-test-queue'; + const JOBS = 50; + + // Push jobs + for (let i = 0; i < JOBS; i++) { + await qm.push(QUEUE, { data: { i } }); + } + + // Start pulling (but don't ack) + const pullPromises = Array.from({ length: 10 }, () => qm.pull(QUEUE, 0)); + await Promise.all(pullPromises); + + let stats = qm.getStats(); + expect(stats.active).toBe(10); + expect(stats.waiting).toBe(40); + + // Drain remaining waiting jobs + qm.drain(QUEUE); + + stats = qm.getStats(); + expect(stats.waiting).toBe(0); + expect(stats.active).toBe(10); // Active jobs not affected by drain + }); + }); +}); diff --git a/test/race-conditions-locks.test.ts b/test/race-conditions-locks.test.ts new file mode 100644 index 00000000..4ac419c3 --- /dev/null +++ b/test/race-conditions-locks.test.ts @@ -0,0 +1,112 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import type { JobId } from '../src/domain/types/job'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('2. Lock Expiration Under Load (checkExpiredLocks)', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should handle concurrent lock renewals and operations', async () => { + const JOBS = 50; + + // Push and pull jobs using pullWithLock + const jobs: Array<{ id: JobId; token: string }> = []; + for (let i = 0; i < JOBS; i++) { + await qm.push('lock-queue', { data: { i } }); + const { job, token } = await qm.pullWithLock('lock-queue', `owner-${i}`, 0); + if (job && token) { + jobs.push({ id: job.id, token }); + } + } + + expect(jobs.length).toBe(JOBS); + + // Concurrent operations: some renew, some ack + const operations: Promise[] = []; + + for (let i = 0; i < jobs.length; i++) { + const job = jobs[i]; + if (i % 2 === 0) { + // Renew lock using correct method name + operations.push( + Promise.resolve(qm.renewJobLock(job.id, job.token, 30000)).then((success) => ({ + type: 'renew', + success, + id: job.id, + })) + ); + } else { + // Ack with token + operations.push( + qm.ack(job.id, { result: 'done' }, job.token).then( + () => ({ type: 'ack', success: true, id: job.id }), + () => ({ type: 'ack', success: false, id: job.id }) + ) + ); + } + } + + const results = await Promise.all(operations); + + const renewed = results.filter((r) => r.type === 'renew' && r.success); + const acked = results.filter((r) => r.type === 'ack' && r.success); + + // All operations should succeed + expect(renewed.length + acked.length).toBe(JOBS); + }); + + test('should safely handle lock verification during concurrent operations', async () => { + const JOB_COUNT = 30; + + // Create jobs with locks + const jobs: Array<{ id: JobId; token: string }> = []; + for (let i = 0; i < JOB_COUNT; i++) { + await qm.push('verify-queue', { data: { i } }); + const { job, token } = await qm.pullWithLock('verify-queue', `owner-${i}`, 0); + if (job && token) { + jobs.push({ id: job.id, token }); + } + } + + expect(jobs.length).toBe(JOB_COUNT); + + // Concurrent verify, renew, and complete operations + const operations: Promise[] = []; + + for (const job of jobs) { + // Verify operation + operations.push( + Promise.resolve(qm.verifyLock(job.id, job.token)).then((valid) => ({ + type: 'verify', + valid, + })) + ); + + // Complete operation (with correct token) + operations.push( + qm.ack(job.id, {}, job.token).then( + () => ({ type: 'ack', success: true }), + () => ({ type: 'ack', success: false }) + ) + ); + } + + const results = await Promise.all(operations); + + // Some verify may fail because job was already acked + // But no errors should occur (race conditions would cause exceptions) + const ackResults = results.filter((r) => r.type === 'ack'); + const successfulAcks = ackResults.filter((r) => r.success); + + // All jobs should eventually be acked (only first ack per job succeeds) + expect(successfulAcks.length).toBe(JOB_COUNT); + }); + }); +}); diff --git a/test/race-conditions-mixed.test.ts b/test/race-conditions-mixed.test.ts new file mode 100644 index 00000000..a57bf24a --- /dev/null +++ b/test/race-conditions-mixed.test.ts @@ -0,0 +1,110 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('4. Mixed Concurrent Operations Stress Test', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture(); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should handle push/pull/ack/fail/release all concurrently', async () => { + const OPERATIONS = 200; + const QUEUES = ['stress-a', 'stress-b', 'stress-c']; + const errors: Error[] = []; + + const operations: Promise[] = []; + + for (let i = 0; i < OPERATIONS; i++) { + const queue = QUEUES[i % QUEUES.length]; + const clientId = `client-${i % 10}`; + const op = i % 5; + + switch (op) { + case 0: + // Push + operations.push( + qm.push(queue, { data: { i } }).catch((e) => { + errors.push(e); + return null; + }) + ); + break; + + case 1: + // Pull and register + operations.push( + qm + .pull(queue, 10) + .then((job) => { + if (job) { + qm.registerClientJob(clientId, job.id); + } + return job; + }) + .catch((e) => { + errors.push(e); + return null; + }) + ); + break; + + case 2: + // Ack random job + operations.push( + qm.pull(queue, 0).then(async (job) => { + if (job) { + try { + await qm.ack(job.id, { result: i }); + } catch { + // Job might have been released/processed already + } + } + }) + ); + break; + + case 3: + // Fail random job + operations.push( + qm.pull(queue, 0).then(async (job) => { + if (job) { + try { + await qm.fail(job.id, `Error ${i}`); + } catch { + // Job might have been released/processed already + } + } + }) + ); + break; + + case 4: + // Release client jobs + operations.push( + qm.releaseClientJobs(clientId).catch((e) => { + errors.push(e); + return 0; + }) + ); + break; + } + } + + await Promise.all(operations); + + // Should have no race condition errors + expect(errors.length).toBe(0); + + // Stats should be consistent + const stats = qm.getStats(); + expect(stats.waiting).toBeGreaterThanOrEqual(0); + expect(stats.active).toBeGreaterThanOrEqual(0); + expect(stats.completed).toBeGreaterThanOrEqual(0); + }); + }); +}); diff --git a/test/race-conditions-stalls.test.ts b/test/race-conditions-stalls.test.ts new file mode 100644 index 00000000..2d43ac69 --- /dev/null +++ b/test/race-conditions-stalls.test.ts @@ -0,0 +1,127 @@ +import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; +import type { QueueManager } from '../src/application/queueManager'; +import type { JobId } from '../src/domain/types/job'; +import { shardIndex } from '../src/shared/hash'; +import { raceConditionsFixture } from './support/race-conditions-fixture'; + +describe('Race Condition Tests', () => { + describe('3. Stall Detection Under Concurrent Load', () => { + let qm: QueueManager; + + const fixture = raceConditionsFixture({ stallCheckMs: 100 }); + beforeEach(async () => { + qm = await fixture.create(); + }); + afterEach(() => fixture.close()); + + test('should handle concurrent lock renewals without losing jobs', async () => { + const JOBS = 20; + const QUEUE = 'heartbeat-queue'; + + // Configure stall detection via shard + const shard = qm.getShards()[shardIndex(QUEUE)]; + shard.setStallConfig(QUEUE, { + enabled: true, + stallInterval: 200, + maxStalls: 3, + gracePeriod: 50, + }); + + // Push and pull jobs with locks + const jobs: Array<{ id: JobId; token: string }> = []; + for (let i = 0; i < JOBS; i++) { + await qm.push(QUEUE, { data: { i } }); + const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); + if (job && token) { + jobs.push({ id: job.id, token }); + } + } + + // Concurrent lock renewals (simulates heartbeats) + for (let round = 0; round < 5; round++) { + const renewPromises = jobs.map((job) => + Promise.resolve(qm.renewJobLock(job.id, job.token, 30000)) + ); + await Promise.all(renewPromises); + await Bun.sleep(50); + } + + // All jobs should still be active (lock renewals prevent stall) + const stats = qm.getStats(); + expect(stats.active).toBe(JOBS); + + // Complete all jobs + for (const job of jobs) { + await qm.ack(job.id, { result: 'done' }, job.token); + } + + const finalStats = qm.getStats(); + expect(finalStats.completed).toBe(JOBS); + expect(finalStats.active).toBe(0); + }); + + test('should not lose jobs during stall detection and recovery', async () => { + const JOBS = 15; + const QUEUE = 'stall-recovery-queue'; + + // Configure stall detection with longer intervals for reliable testing + const shard = qm.getShards()[shardIndex(QUEUE)]; + shard.setStallConfig(QUEUE, { + enabled: true, + stallInterval: 50, + maxStalls: 5, // Higher to avoid DLQ during test + gracePeriod: 10, + }); + + // Push jobs with high max attempts + for (let i = 0; i < JOBS; i++) { + await qm.push(QUEUE, { + data: { i }, + maxAttempts: 10, + }); + } + + // Pull all jobs with locks + const pulledJobs: Array<{ id: JobId; token: string }> = []; + for (let i = 0; i < JOBS; i++) { + const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); + if (job && token) pulledJobs.push({ id: job.id, token }); + } + expect(pulledJobs.length).toBe(JOBS); + + // Complete some jobs + const toComplete = Math.floor(JOBS / 3); + for (let i = 0; i < toComplete; i++) { + await qm.ack(pulledJobs[i].id, { result: 'done' }, pulledJobs[i].token); + } + + // Check that completed jobs are accounted for + let stats = qm.getStats(); + expect(stats.completed).toBe(toComplete); + + // Complete remaining jobs + for (let i = toComplete; i < JOBS; i++) { + try { + await qm.ack(pulledJobs[i].id, { result: 'done' }, pulledJobs[i].token); + } catch { + // Job might have been stalled and requeued, try to pull and complete + } + } + + // Pull and complete any remaining jobs + let iterations = 0; + while (iterations < 50) { + const { job, token } = await qm.pullWithLock(QUEUE, 'cleanup', 50); + if (!job) break; + await qm.ack(job.id, { result: 'recovered' }, token!); + iterations++; + } + + const finalStats = qm.getStats(); + // All jobs should be completed or in DLQ + expect(finalStats.waiting).toBe(0); + expect(finalStats.active).toBe(0); + expect(finalStats.completed + finalStats.dlq).toBe(JOBS); + }); + }); +}); diff --git a/test/race-conditions.test.ts b/test/race-conditions.test.ts index 7d1a9516..564efab0 100644 --- a/test/race-conditions.test.ts +++ b/test/race-conditions.test.ts @@ -1,926 +1,113 @@ -/** - * Race Condition Tests - * Tests for concurrent operations that could cause race conditions - * Specifically tests the fixes for: - * - releaseClientJobs() - * - checkExpiredLocks() - * - handleStalledJob() - */ - import { describe, test, expect, beforeEach, afterEach } from 'bun:test'; -import { QueueManager } from '../src/application/queueManager'; +import type { QueueManager } from '../src/application/queueManager'; import type { JobId } from '../src/domain/types/job'; -import { shardIndex } from '../src/shared/hash'; -import { unlink } from 'fs/promises'; - -const TEST_DB = './test-race.db'; - -async function cleanup() { - for (const f of [TEST_DB, `${TEST_DB}-wal`, `${TEST_DB}-shm`]) { - if (await Bun.file(f).exists()) await unlink(f); - } -} +import { raceConditionsFixture } from './support/race-conditions-fixture'; +import { crossRetryDeadlineAfterWaitingRead } from './support/race-conditions-clock'; +import { + assertRaceConservation, + chaosActions, + runChaos, + raceTraceJson, + type ChaosAction, +} from './support/race-conditions-chaos'; describe('Race Condition Tests', () => { - describe('1. Concurrent Client Disconnections (releaseClientJobs)', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should safely release jobs when multiple clients disconnect simultaneously', async () => { - const CLIENTS = 10; - const JOBS_PER_CLIENT = 20; - - // Push jobs for each "client" - const clientJobs: Map = new Map(); - - for (let c = 0; c < CLIENTS; c++) { - const clientId = `client-${c}`; - clientJobs.set(clientId, []); - - for (let j = 0; j < JOBS_PER_CLIENT; j++) { - const job = await qm.push('race-queue', { - data: { client: c, job: j }, - }); - clientJobs.get(clientId)!.push(job.id); - } - } - - // Pull jobs and register them with clients - for (let c = 0; c < CLIENTS; c++) { - const clientId = `client-${c}`; - for (let j = 0; j < JOBS_PER_CLIENT; j++) { - const job = await qm.pull('race-queue', 0); - if (job) { - qm.registerClientJob(clientId, job.id); - } - } - } - - let stats = qm.getStats(); - expect(stats.active).toBe(CLIENTS * JOBS_PER_CLIENT); - expect(stats.waiting).toBe(0); - - // Simulate all clients disconnecting simultaneously - const releasePromises = Array.from({ length: CLIENTS }, (_, c) => - qm.releaseClientJobs(`client-${c}`) - ); - - const results = await Promise.all(releasePromises); - - // Each client should have released their jobs - const totalReleased = results.reduce((a, b) => a + b, 0); - expect(totalReleased).toBe(CLIENTS * JOBS_PER_CLIENT); - - // All jobs should be back in waiting state - stats = qm.getStats(); - expect(stats.active).toBe(0); - expect(stats.waiting).toBe(CLIENTS * JOBS_PER_CLIENT); - - // Verify jobs can be pulled again - let pulledCount = 0; - for (let i = 0; i < CLIENTS * JOBS_PER_CLIENT; i++) { - const job = await qm.pull('race-queue', 0); - if (job) pulledCount++; - } - expect(pulledCount).toBe(CLIENTS * JOBS_PER_CLIENT); - }); - - test('should handle interleaved pull and release operations', async () => { - const ITERATIONS = 50; - - // Push initial jobs - for (let i = 0; i < ITERATIONS; i++) { - await qm.push('interleave-queue', { data: { i } }); - } - - const operations: Promise[] = []; - const processedJobs: Set = new Set(); - - // Interleave pull, ack, and release operations - for (let i = 0; i < ITERATIONS; i++) { - const clientId = `client-${i % 5}`; - - // Pull operation with client registration - operations.push( - qm.pull('interleave-queue', 50).then(async (job) => { - if (job) { - processedJobs.add(String(job.id)); - qm.registerClientJob(clientId, job.id); - // Randomly ack or let release handle it - if (Math.random() > 0.5) { - await qm.ack(job.id, { result: 'done' }); - qm.unregisterClientJob(clientId, job.id); - return 'acked'; - } - return 'pulled'; - } - return 'empty'; - }) - ); - - // Occasionally release client jobs - if (i % 10 === 9) { - operations.push(qm.releaseClientJobs(clientId).then((count) => `released-${count}`)); - } - } - - await Promise.all(operations); - - // Should not have lost any jobs - all should be completed, waiting, or active - const stats = qm.getStats(); - expect(stats.waiting + stats.active + stats.completed).toBe(ITERATIONS); - }); - - test('should not duplicate jobs when releasing under contention', async () => { - const JOBS = 100; - const CLIENT = 'contention-client'; - - // Push and pull all jobs - for (let i = 0; i < JOBS; i++) { - await qm.push('contention-queue', { data: { i } }); - } - - for (let i = 0; i < JOBS; i++) { - const job = await qm.pull('contention-queue', 0); - if (job) { - qm.registerClientJob(CLIENT, job.id); - } - } - - // Try to release multiple times concurrently - const releaseAttempts = Array.from({ length: 10 }, () => qm.releaseClientJobs(CLIENT)); - - const results = await Promise.all(releaseAttempts); - - // Only the first release should have released jobs - const totalReleased = results.reduce((a, b) => a + b, 0); - expect(totalReleased).toBe(JOBS); - - // Exactly JOBS should be in waiting state - const stats = qm.getStats(); - expect(stats.waiting).toBe(JOBS); - expect(stats.active).toBe(0); - }); - }); - - describe('2. Lock Expiration Under Load (checkExpiredLocks)', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should handle concurrent lock renewals and operations', async () => { - const JOBS = 50; - - // Push and pull jobs using pullWithLock - const jobs: Array<{ id: JobId; token: string }> = []; - for (let i = 0; i < JOBS; i++) { - await qm.push('lock-queue', { data: { i } }); - const { job, token } = await qm.pullWithLock('lock-queue', `owner-${i}`, 0); - if (job && token) { - jobs.push({ id: job.id, token }); - } - } - - expect(jobs.length).toBe(JOBS); - - // Concurrent operations: some renew, some ack - const operations: Promise[] = []; - - for (let i = 0; i < jobs.length; i++) { - const job = jobs[i]; - if (i % 2 === 0) { - // Renew lock using correct method name - operations.push( - Promise.resolve(qm.renewJobLock(job.id, job.token, 30000)).then((success) => ({ - type: 'renew', - success, - id: job.id, - })) - ); - } else { - // Ack with token - operations.push( - qm.ack(job.id, { result: 'done' }, job.token).then( - () => ({ type: 'ack', success: true, id: job.id }), - () => ({ type: 'ack', success: false, id: job.id }) - ) - ); - } - } - - const results = await Promise.all(operations); - - const renewed = results.filter((r) => r.type === 'renew' && r.success); - const acked = results.filter((r) => r.type === 'ack' && r.success); - - // All operations should succeed - expect(renewed.length + acked.length).toBe(JOBS); - }); - - test('should safely handle lock verification during concurrent operations', async () => { - const JOB_COUNT = 30; - - // Create jobs with locks - const jobs: Array<{ id: JobId; token: string }> = []; - for (let i = 0; i < JOB_COUNT; i++) { - await qm.push('verify-queue', { data: { i } }); - const { job, token } = await qm.pullWithLock('verify-queue', `owner-${i}`, 0); - if (job && token) { - jobs.push({ id: job.id, token }); - } - } - - expect(jobs.length).toBe(JOB_COUNT); - - // Concurrent verify, renew, and complete operations - const operations: Promise[] = []; - - for (const job of jobs) { - // Verify operation - operations.push( - Promise.resolve(qm.verifyLock(job.id, job.token)).then((valid) => ({ - type: 'verify', - valid, - })) - ); - - // Complete operation (with correct token) - operations.push( - qm.ack(job.id, {}, job.token).then( - () => ({ type: 'ack', success: true }), - () => ({ type: 'ack', success: false }) - ) - ); - } - - const results = await Promise.all(operations); - - // Some verify may fail because job was already acked - // But no errors should occur (race conditions would cause exceptions) - const ackResults = results.filter((r) => r.type === 'ack'); - const successfulAcks = ackResults.filter((r) => r.success); - - // All jobs should eventually be acked (only first ack per job succeeds) - expect(successfulAcks.length).toBe(JOB_COUNT); - }); - }); - - describe('3. Stall Detection Under Concurrent Load', () => { + describe('4. Mixed Concurrent Operations Stress Test', () => { let qm: QueueManager; - + const fixture = raceConditionsFixture(); beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ - dataPath: TEST_DB, - stallCheckMs: 100, // Fast stall check for testing - }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should handle concurrent lock renewals without losing jobs', async () => { - const JOBS = 20; - const QUEUE = 'heartbeat-queue'; - - // Configure stall detection via shard - const shard = qm.getShards()[shardIndex(QUEUE)]; - shard.setStallConfig(QUEUE, { - enabled: true, - stallInterval: 200, - maxStalls: 3, - gracePeriod: 50, - }); - - // Push and pull jobs with locks - const jobs: Array<{ id: JobId; token: string }> = []; - for (let i = 0; i < JOBS; i++) { - await qm.push(QUEUE, { data: { i } }); - const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); - if (job && token) { - jobs.push({ id: job.id, token }); - } - } - - // Concurrent lock renewals (simulates heartbeats) - for (let round = 0; round < 5; round++) { - const renewPromises = jobs.map((job) => - Promise.resolve(qm.renewJobLock(job.id, job.token, 30000)) - ); - await Promise.all(renewPromises); - await Bun.sleep(50); - } - - // All jobs should still be active (lock renewals prevent stall) - const stats = qm.getStats(); - expect(stats.active).toBe(JOBS); - - // Complete all jobs - for (const job of jobs) { - await qm.ack(job.id, { result: 'done' }, job.token); - } - - const finalStats = qm.getStats(); - expect(finalStats.completed).toBe(JOBS); - expect(finalStats.active).toBe(0); + qm = await fixture.create(); }); + afterEach(() => fixture.close()); - test('should not lose jobs during stall detection and recovery', async () => { - const JOBS = 15; - const QUEUE = 'stall-recovery-queue'; - - // Configure stall detection with longer intervals for reliable testing - const shard = qm.getShards()[shardIndex(QUEUE)]; - shard.setStallConfig(QUEUE, { - enabled: true, - stallInterval: 50, - maxStalls: 5, // Higher to avoid DLQ during test - gracePeriod: 10, - }); - - // Push jobs with high max attempts - for (let i = 0; i < JOBS; i++) { - await qm.push(QUEUE, { + async function populate(backoff?: number) { + const ids: JobId[] = []; + for (let i = 0; i < 100; i++) { + const job = await qm.push('invariant-queue', { data: { i }, maxAttempts: 10, + ...(backoff === undefined ? {} : { backoff }), }); + ids.push(job.id); } - - // Pull all jobs with locks - const pulledJobs: Array<{ id: JobId; token: string }> = []; - for (let i = 0; i < JOBS; i++) { - const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); - if (job && token) pulledJobs.push({ id: job.id, token }); - } - expect(pulledJobs.length).toBe(JOBS); - - // Complete some jobs - const toComplete = Math.floor(JOBS / 3); - for (let i = 0; i < toComplete; i++) { - await qm.ack(pulledJobs[i].id, { result: 'done' }, pulledJobs[i].token); - } - - // Check that completed jobs are accounted for - let stats = qm.getStats(); - expect(stats.completed).toBe(toComplete); - - // Complete remaining jobs - for (let i = toComplete; i < JOBS; i++) { - try { - await qm.ack(pulledJobs[i].id, { result: 'done' }, pulledJobs[i].token); - } catch { - // Job might have been stalled and requeued, try to pull and complete - } - } - - // Pull and complete any remaining jobs - let iterations = 0; - while (iterations < 50) { - const { job, token } = await qm.pullWithLock(QUEUE, 'cleanup', 50); - if (!job) break; - await qm.ack(job.id, { result: 'recovered' }, token!); - iterations++; - } - - const finalStats = qm.getStats(); - // All jobs should be completed or in DLQ - expect(finalStats.waiting).toBe(0); - expect(finalStats.active).toBe(0); - expect(finalStats.completed + finalStats.dlq).toBe(JOBS); - }); - }); - - describe('4. Mixed Concurrent Operations Stress Test', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should handle push/pull/ack/fail/release all concurrently', async () => { - const OPERATIONS = 200; - const QUEUES = ['stress-a', 'stress-b', 'stress-c']; - const errors: Error[] = []; - - const operations: Promise[] = []; - - for (let i = 0; i < OPERATIONS; i++) { - const queue = QUEUES[i % QUEUES.length]; - const clientId = `client-${i % 10}`; - const op = i % 5; - - switch (op) { - case 0: - // Push - operations.push( - qm.push(queue, { data: { i } }).catch((e) => { - errors.push(e); - return null; - }) - ); - break; - - case 1: - // Pull and register - operations.push( - qm - .pull(queue, 10) - .then((job) => { - if (job) { - qm.registerClientJob(clientId, job.id); - } - return job; - }) - .catch((e) => { - errors.push(e); - return null; - }) - ); - break; - - case 2: - // Ack random job - operations.push( - qm.pull(queue, 0).then(async (job) => { - if (job) { - try { - await qm.ack(job.id, { result: i }); - } catch { - // Job might have been released/processed already - } - } - }) - ); - break; - - case 3: - // Fail random job - operations.push( - qm.pull(queue, 0).then(async (job) => { - if (job) { - try { - await qm.fail(job.id, `Error ${i}`); - } catch { - // Job might have been released/processed already - } - } - }) - ); - break; - - case 4: - // Release client jobs - operations.push( - qm.releaseClientJobs(clientId).catch((e) => { - errors.push(e); - return 0; - }) - ); - break; - } - } - - await Promise.all(operations); - - // Should have no race condition errors - expect(errors.length).toBe(0); - - // Stats should be consistent - const stats = qm.getStats(); - expect(stats.waiting).toBeGreaterThanOrEqual(0); - expect(stats.active).toBeGreaterThanOrEqual(0); - expect(stats.completed).toBeGreaterThanOrEqual(0); - }); + expect(qm.getStats().waiting).toBe(100); + return ids; + } test('should maintain job count invariant under chaos', async () => { - const TOTAL_JOBS = 100; - const QUEUE = 'invariant-queue'; - - // Push all jobs - for (let i = 0; i < TOTAL_JOBS; i++) { - await qm.push(QUEUE, { data: { i }, maxAttempts: 10 }); - } - - const initialStats = qm.getStats(); - expect(initialStats.waiting).toBe(TOTAL_JOBS); - - // Chaos: random operations - ack or fail only (no client tracking to simplify) - const chaos: Promise[] = []; - for (let i = 0; i < 200; i++) { - const op = Math.floor(Math.random() * 3); - - switch (op) { - case 0: - // Pull without processing (will timeout later) - chaos.push(qm.pull(QUEUE, 5).catch(() => null)); - break; - case 1: - // Pull and ack - chaos.push( - qm.pull(QUEUE, 0).then(async (job) => { - if (job) { - try { - await qm.ack(job.id, {}); - } catch { - /* ignore */ - } - } - }) - ); - break; - case 2: - // Pull and fail - chaos.push( - qm.pull(QUEUE, 0).then(async (job) => { - if (job) { - try { - await qm.fail(job.id, 'chaos'); - } catch { - /* ignore */ - } - } - }) - ); - break; - } - } - - await Promise.all(chaos); - - // Wait for any pending retries - await Bun.sleep(100); - - // Invariant: total jobs should be accounted for - const midStats = qm.getStats(); - const accountedFor = midStats.waiting + midStats.active + midStats.completed + midStats.dlq; - - // All jobs should be somewhere in the system (allowing for some in-flight) - expect(accountedFor).toBeLessThanOrEqual(TOTAL_JOBS); - expect(accountedFor).toBeGreaterThan(0); - - // Clean up: complete all remaining jobs with longer timeout - let processed = 0; - let iterations = 0; - while (iterations < TOTAL_JOBS * 3) { - const job = await qm.pull(QUEUE, 200); - if (!job) break; - try { - await qm.ack(job.id, {}); - processed++; - } catch { - /* job might have been processed already */ - } - iterations++; - } - - // Final state: no waiting jobs - const endStats = qm.getStats(); - expect(endStats.waiting).toBe(0); - // All jobs should be accounted for somewhere (completed, DLQ, or still active) - // This test verifies no jobs are lost, but allows for some to still be in-flight - expect(endStats.completed + endStats.dlq + endStats.active).toBeGreaterThanOrEqual( - TOTAL_JOBS * 0.5 - ); - }); - }); - - describe('5. High Concurrency Lock Contention', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should handle 100 concurrent pulls from same queue', async () => { - const JOBS = 100; - const QUEUE = 'contention-pull-queue'; - - // Push all jobs - for (let i = 0; i < JOBS; i++) { - await qm.push(QUEUE, { data: { i } }); - } - - // 100 concurrent pulls - const pulls = Array.from({ length: JOBS }, () => qm.pull(QUEUE, 100)); - - const results = await Promise.all(pulls); - const pulledJobs = results.filter((r) => r !== null); - const pulledIds = new Set(pulledJobs.map((j) => String(j!.id))); - - // Each job should be pulled exactly once (no duplicates) - expect(pulledIds.size).toBe(JOBS); - expect(pulledJobs.length).toBe(JOBS); - }); - - test('should handle concurrent acks for different jobs', async () => { - const JOBS = 50; - const QUEUE = 'contention-ack-queue'; - - // Push and pull jobs with locks - const jobs: Array<{ id: JobId; token?: string }> = []; - for (let i = 0; i < JOBS; i++) { - await qm.push(QUEUE, { data: { i } }); - const { job, token } = await qm.pullWithLock(QUEUE, `owner-${i}`, 0); - if (job) { - jobs.push({ id: job.id, token: token ?? undefined }); - } - } - - // Concurrent acks - const acks = jobs.map((job) => - qm.ack(job.id, { result: 'done' }, job.token).then( - () => ({ success: true, id: job.id }), - (e) => ({ success: false, id: job.id, error: e.message }) - ) + const seed = 0xc1a05; + // Preserve default retry backoff. The seed fixes operation choices only; + // runtime jitter and asynchronous completion order are captured in the trace. + const actions = chaosActions(seed); + const ids = await populate(); + const trace = await runChaos(qm, 'invariant-queue', actions); + try { + expect(trace).toHaveLength(200); + for (const action of ['pull', 'ack', 'fail']) { + expect(trace.some((entry) => entry.action === action && entry.id !== null)).toBe(true); + } + const snapshot = assertRaceConservation(qm, 'invariant-queue', ids); + if (process.env.BUNQUEUE_RACE_TRACE === '1') { + console.log(raceTraceJson({ seed, actions, trace, snapshot })); + } + } catch (cause) { + throw new Error(JSON.stringify({ seed, actions, trace }), { cause }); + } + }); + + test('should account for all 100 jobs when 51 retries are delayed and 49 completed', async () => { + // A controlled long backoff keeps the exact partition independent of runner speed. + // The seeded case above exercises the original default-backoff contract. + const ids = await populate(60_000); + const actions: ChaosAction[] = [ + ...Array(51).fill('fail'), + ...Array(49).fill('ack'), + ...Array(100).fill('pull'), + ]; + const trace = await runChaos(qm, 'invariant-queue', actions); + expect(trace).toHaveLength(200); + expect(trace.filter((entry) => entry.action === 'fail' && entry.id !== null)).toHaveLength( + 51 ); - - const results = await Promise.all(acks); - const successful = results.filter((r) => r.success); - - expect(successful.length).toBe(JOBS); - - const stats = qm.getStats(); - expect(stats.completed).toBe(JOBS); - expect(stats.active).toBe(0); - }); - - test('should handle rapid push/pull cycles without losing jobs', async () => { - const CYCLES = 100; - const QUEUE = 'rapid-cycle-queue'; - const processedIds: Set = new Set(); - const errors: string[] = []; - - // Rapid push/pull/ack cycles - const cycles = Array.from({ length: CYCLES }, async (_, i) => { - try { - await qm.push(QUEUE, { data: { cycle: i } }); - const pulled = await qm.pull(QUEUE, 100); - if (pulled) { - processedIds.add(String(pulled.id)); - await qm.ack(pulled.id, { result: i }); - } - return true; - } catch (e) { - errors.push(String(e)); - return false; - } - }); - - const results = await Promise.all(cycles); - const successCount = results.filter((r) => r).length; - - // Most cycles should succeed - expect(successCount).toBeGreaterThan(CYCLES * 0.8); - expect(errors.length).toBeLessThan(CYCLES * 0.2); - - // Clean up any remaining jobs - while (true) { - const job = await qm.pull(QUEUE, 10); - if (!job) break; - await qm.ack(job.id, {}); - } - - const stats = qm.getStats(); - expect(stats.waiting).toBe(0); - expect(stats.active).toBe(0); - }); - }); - - describe('6. Cross-Queue Concurrent Operations', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('should handle operations across 10 queues simultaneously', async () => { - const QUEUES = Array.from({ length: 10 }, (_, i) => `multi-queue-${i}`); - const JOBS_PER_QUEUE = 20; - - // Push to all queues concurrently - const pushPromises: Promise[] = []; - for (const queue of QUEUES) { - for (let j = 0; j < JOBS_PER_QUEUE; j++) { - pushPromises.push(qm.push(queue, { data: { queue, job: j } })); - } - } - await Promise.all(pushPromises); - - let stats = qm.getStats(); - expect(stats.waiting).toBe(QUEUES.length * JOBS_PER_QUEUE); - - // Pull and ack from all queues concurrently - const processPromises: Promise[] = []; - for (const queue of QUEUES) { - for (let j = 0; j < JOBS_PER_QUEUE; j++) { - processPromises.push( - qm.pull(queue, 100).then(async (job) => { - if (job) { - await qm.ack(job.id, { result: 'done' }); - return true; - } - return false; - }) - ); - } - } - - const results = await Promise.all(processPromises); - const processed = results.filter((r) => r).length; - - expect(processed).toBe(QUEUES.length * JOBS_PER_QUEUE); - - stats = qm.getStats(); - expect(stats.completed).toBe(QUEUES.length * JOBS_PER_QUEUE); + expect(trace.filter((entry) => entry.action === 'ack' && entry.id !== null)).toHaveLength(49); + const { stats } = assertRaceConservation(qm, 'invariant-queue', ids); + expect(stats.delayed).toBe(51); + expect(stats.completed).toBe(49); expect(stats.waiting).toBe(0); expect(stats.active).toBe(0); + expect(stats.dlq).toBe(0); + }); + + test('should reject a missing original job even when the legacy half-count passes', async () => { + const ids = await populate(60_000); + // Deliberately remove a retained job through the real persistence path to + // prove the oracle cannot hide loss behind an aggregate half-count. + expect(await qm.cancel(ids[0])).toBe(true); + await runChaos(qm, 'invariant-queue', Array(200).fill('ack')); + expect(qm.getStats().completed).toBe(99); + expect(() => assertRaceConservation(qm, 'invariant-queue', ids)).toThrow( + 'Job conservation failed' + ); }); - test('should handle drain during active processing', async () => { - const QUEUE = 'drain-test-queue'; - const JOBS = 50; - - // Push jobs - for (let i = 0; i < JOBS; i++) { - await qm.push(QUEUE, { data: { i } }); - } - - // Start pulling (but don't ack) - const pullPromises = Array.from({ length: 10 }, () => qm.pull(QUEUE, 0)); - await Promise.all(pullPromises); - - let stats = qm.getStats(); - expect(stats.active).toBe(10); - expect(stats.waiting).toBe(40); - - // Drain remaining waiting jobs - qm.drain(QUEUE); - - stats = qm.getStats(); - expect(stats.waiting).toBe(0); - expect(stats.active).toBe(10); // Active jobs not affected by drain - }); - }); -}); - -describe('Race Condition Benchmarks', () => { - let qm: QueueManager; - - beforeEach(async () => { - await cleanup(); - qm = new QueueManager({ dataPath: TEST_DB }); - }); - - afterEach(async () => { - qm.shutdown(); - await cleanup(); - }); - - test('BENCHMARK: 1000 concurrent push operations', async () => { - const COUNT = 1000; - - const start = performance.now(); - const pushes = Array.from({ length: COUNT }, (_, i) => qm.push('bench-push', { data: { i } })); - await Promise.all(pushes); - const duration = performance.now() - start; - - console.log(`\n📊 1000 concurrent pushes: ${duration.toFixed(2)}ms`); - console.log(` Throughput: ${(COUNT / (duration / 1000)).toFixed(0)} ops/sec`); - - const stats = qm.getStats(); - expect(stats.waiting).toBe(COUNT); - }); - - test('BENCHMARK: 500 concurrent pull operations', async () => { - const COUNT = 500; - - // First push - for (let i = 0; i < COUNT; i++) { - await qm.push('bench-pull', { data: { i } }); - } - - const start = performance.now(); - const pulls = Array.from({ length: COUNT }, () => qm.pull('bench-pull', 100)); - const results = await Promise.all(pulls); - const duration = performance.now() - start; - - const pulled = results.filter((r) => r !== null).length; - - console.log(`\n📊 500 concurrent pulls: ${duration.toFixed(2)}ms`); - console.log(` Pulled: ${pulled} jobs`); - console.log(` Throughput: ${(pulled / (duration / 1000)).toFixed(0)} ops/sec`); - - expect(pulled).toBe(COUNT); - }); - - test('BENCHMARK: Mixed concurrent operations (push/pull/ack)', async () => { - const ITERATIONS = 300; - - const start = performance.now(); - const operations: Promise[] = []; - - for (let i = 0; i < ITERATIONS; i++) { - // Push - operations.push(qm.push('bench-mixed', { data: { i } })); - - // Pull and ack - operations.push( - qm.pull('bench-mixed', 50).then(async (job) => { - if (job) { - await qm.ack(job.id, {}); - return true; - } - return false; - }) + test('should retain a retry when its deadline crosses between waiting and delayed reads', async () => { + const job = await qm.push('invariant-queue', { + data: { retry: true }, + maxAttempts: 2, + backoff: 60_000, + }); + expect((await qm.pull('invariant-queue', 0))?.id).toBe(job.id); + expect(await qm.fail(job.id, 'retry across snapshot')).toBeUndefined(); + const retry = await qm.getJob(job.id); + expect(retry).not.toBeNull(); + if (!retry) throw new Error('Missing retained retry'); + const { value, crossed } = crossRetryDeadlineAfterWaitingRead(qm, retry.runAt, () => + assertRaceConservation(qm, 'invariant-queue', [job.id]) ); - } - - await Promise.all(operations); - const duration = performance.now() - start; - - console.log(`\n📊 ${ITERATIONS * 2} mixed operations: ${duration.toFixed(2)}ms`); - console.log(` Throughput: ${((ITERATIONS * 2) / (duration / 1000)).toFixed(0)} ops/sec`); - - // Clean up remaining - while (true) { - const job = await qm.pull('bench-mixed', 10); - if (!job) break; - await qm.ack(job.id, {}); - } - - const stats = qm.getStats(); - expect(stats.waiting).toBe(0); - expect(stats.active).toBe(0); - }); - - test('BENCHMARK: Client disconnect/release under load', async () => { - const CLIENTS = 20; - const JOBS_PER_CLIENT = 50; - - // Setup: push and pull jobs for each client - for (let c = 0; c < CLIENTS; c++) { - for (let j = 0; j < JOBS_PER_CLIENT; j++) { - await qm.push('bench-release', { data: { c, j } }); - } - } - - for (let c = 0; c < CLIENTS; c++) { - for (let j = 0; j < JOBS_PER_CLIENT; j++) { - const job = await qm.pull('bench-release', 0); - if (job) { - qm.registerClientJob(`client-${c}`, job.id); - } - } - } - - const start = performance.now(); - const releases = Array.from({ length: CLIENTS }, (_, c) => qm.releaseClientJobs(`client-${c}`)); - const results = await Promise.all(releases); - const duration = performance.now() - start; - - const totalReleased = results.reduce((a, b) => a + b, 0); - - console.log(`\n📊 ${CLIENTS} concurrent client releases: ${duration.toFixed(2)}ms`); - console.log(` Total jobs released: ${totalReleased}`); - console.log(` Throughput: ${(totalReleased / (duration / 1000)).toFixed(0)} jobs/sec`); - - expect(totalReleased).toBe(CLIENTS * JOBS_PER_CLIENT); + expect(crossed).toBe(true); + expect(value.stats.waiting).toBe(1); + expect(value.stats.delayed).toBe(0); + expect(value.membership.find(({ state }) => state === 'waiting')?.ids).toEqual([job.id]); + }); }); }); diff --git a/test/repro-postgres-public-progress-retry.test.ts b/test/repro-postgres-public-progress-retry.test.ts new file mode 100644 index 00000000..b6727736 --- /dev/null +++ b/test/repro-postgres-public-progress-retry.test.ts @@ -0,0 +1,202 @@ +import { expect, test } from 'bun:test'; +import { SQL } from 'bun'; +import type { Job } from '../src/client'; +import { + PostgresPublicApiHarness, + waitFor, + POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS, +} from './support/postgres-public-api-harness'; +import { expectedPostgresVersionPrefix } from './support/postgres-version'; +import { PublicDeliveryTrace } from './support/postgres-public-delivery-trace'; + +const url = Bun.env.BUNQUEUE_TEST_POSTGRES_URL; + +async function bounded(work: Promise, description: string, ms: number): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + work, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out ${description}`)), ms); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +test.skipIf(!url)( + 'a rolled-back Progress request retries with a new fenced delivery and one durable completion', + async () => { + const harness = await PostgresPublicApiHarness.start(url!, expectedPostgresVersionPrefix); + const sql = new SQL(url!, { max: 2, connectionTimeout: 5 }); + const blocker = new SQL(url!, { max: 1, connectionTimeout: 5 }); + const first = Promise.withResolvers>(); + const second = Promise.withResolvers>(); + const issueProgress = Promise.withResolvers(); + const rejected = Promise.withResolvers(); + const releaseFailure = Promise.withResolvers(); + const releaseSuccess = Promise.withResolvers(); + const locked = Promise.withResolvers(); + const unlock = Promise.withResolvers(); + const deliveries: { id: string; token?: string; attempts: number }[] = []; + const failures: { id?: string; token?: string; error: string }[] = []; + const trace = new PublicDeliveryTrace(); + let lock: Promise | undefined; + let primaryError: unknown; + try { + const name = harness.unique('progress-lock-retry'); + const queue = harness.queue<{ value: number }>(name, 0); + const observer = harness.queue<{ value: number }>(name, 3); + const worker = harness.worker<{ value: number }, number>( + name, + 1, + async (job) => { + const invocation = trace.start(job); + deliveries.push({ id: job.id, token: job.token, attempts: job.attemptsMade }); + try { + if (deliveries.length === 1) { + first.resolve(job); + await issueProgress.promise; + try { + await job.updateProgress({ phase: 'running' }); + } catch (error) { + rejected.resolve(error); + await releaseFailure.promise; + throw error; + } + throw new Error('Progress unexpectedly escaped the owned SQL row lock'); + } + second.resolve(job); + await releaseSuccess.promise; + await job.updateProgress({ phase: 'running' }); + invocation.stage = 'log'; + await job.log(`processed-${job.data.value}`); + return trace.returned(invocation, job.data.value * 2); + } catch (error) { + trace.threw(invocation, error); + throw error; + } + }, + { concurrency: 1 } + ); + worker.on('failed', (job, error) => { + trace.failed(job, error); + failures.push({ id: job?.id, token: job?.token, error: String(error) }); + }); + worker.on('error', (error) => trace.unexpected('error', error)); + worker.on('stalled', (id) => trace.unexpected('stalled', id)); + await worker.waitUntilReady(); + const job = await queue.add('double', { value: 7 }, { durable: true }); + await waitFor('first processor entry', () => deliveries.length === 1); + const initial = await first.promise; + lock = blocker.begin(async (tx) => { + await tx`SET LOCAL statement_timeout = '5s'`; + await tx`SELECT id FROM bunqueue_jobs WHERE namespace = ${harness.namespace} + AND id = ${job.id} FOR UPDATE`; + const [row] = await tx<{ pid: number }[]>`SELECT pg_backend_pid() AS pid`; + locked.resolve(row.pid); + await unlock.promise; + }); + // Observe rejection immediately, including failures before the lock-ready barrier. + void lock.catch(() => {}); + let blockerPid: number | undefined; + void locked.promise.then((pid) => { + blockerPid = pid; + }); + await waitFor('owned row lock', () => blockerPid !== undefined); + issueProgress.resolve(); + await waitFor('Progress blocked by the owned transaction', async () => { + const [row] = await sql<{ waiting: boolean }[]>`SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity WHERE ${blockerPid!} = ANY(pg_blocking_pids(pid)) + ) AS waiting`; + return row.waiting; + }); + let progressError: unknown; + void rejected.promise.then((error) => { + progressError = error; + }); + await waitFor('Progress lock-timeout response', () => progressError !== undefined, 10_000); + expect(String(progressError)).toContain('Internal server error'); + const [before] = await sql<{ attempts: number; lease_token: string }[]>` + SELECT attempts, lease_token FROM bunqueue_jobs + WHERE namespace = ${harness.namespace} AND id = ${job.id}`; + expect(before.attempts).toBe(0); + expect(before.lease_token).toBe(initial.token!); + unlock.resolve(); + await lock; + releaseFailure.resolve(); + await waitFor('retry processor entry', () => deliveries.length === 2); + const retry = await second.promise; + expect(failures).toEqual([ + { id: job.id, token: initial.token, error: String(progressError) }, + ]); + expect(initial.token).toBeTruthy(); + expect(retry.token).toBeTruthy(); + expect(retry.token).not.toBe(initial.token); + expect(deliveries.map(({ attempts }) => attempts)).toEqual([0, 1]); + const [active] = await sql<{ attempts: number; lease_token: string }[]>` + SELECT attempts, lease_token FROM bunqueue_jobs + WHERE namespace = ${harness.namespace} AND id = ${job.id}`; + expect(active).toEqual({ attempts: 1, lease_token: retry.token! }); + releaseSuccess.resolve(); + await waitFor( + 'one completed retry', + async () => (await observer.getJobCountsAsync()).completed === 1 + ); + expect(await observer.getJobCountsAsync()).toMatchObject({ + waiting: 0, + active: 0, + completed: 1, + failed: 0, + delayed: 0, + prioritized: 0, + paused: 0, + 'waiting-children': 0, + }); + expect(await observer.getJobState(job.id)).toBe('completed'); + const [events] = await sql<{ failed: string; completed: string }[]>` + SELECT count(*) FILTER (WHERE event_type = 'failed')::text AS failed, + count(*) FILTER (WHERE event_type = 'completed')::text AS completed + FROM bunqueue_events WHERE namespace = ${harness.namespace} AND job_id = ${job.id}`; + expect(events).toEqual({ failed: '1', completed: '1' }); + const retained = await observer.getJob(job.id); + expect(retained?.returnvalue).toBe(14); + expect(retained?.attemptsMade).toBe(1); + expect(await observer.getJobLogs(job.id)).toMatchObject({ + count: 1, + logs: ['[info] processed-7'], + }); + await trace.assert(url!, harness.namespace, name, new Map([[job.id, 14]])); + } catch (error) { + primaryError = error; + throw error; + } finally { + issueProgress.resolve(); + unlock.resolve(); + releaseFailure.resolve(); + releaseSuccess.resolve(); + const lockCleanup = await Promise.allSettled([ + bounded(Promise.resolve(lock), 'releasing owned row lock', 10_000), + ]); + const resources = await Promise.allSettled([ + bounded( + harness.close(), + 'closing public API fixture', + POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS + ), + sql.close({ timeout: 5 }), + blocker.close({ timeout: 5 }), + ]); + const errors = [...lockCleanup, ...resources].flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (errors.length) + throw new AggregateError( + primaryError ? [primaryError, ...errors] : errors, + 'Progress retry cleanup failed' + ); + } + }, + POSTGRES_PUBLIC_API_TEARDOWN_TIMEOUT_MS + 30_000 +); diff --git a/test/support/postgres-public-delivery-oracle.ts b/test/support/postgres-public-delivery-oracle.ts new file mode 100644 index 00000000..e41b07cb --- /dev/null +++ b/test/support/postgres-public-delivery-oracle.ts @@ -0,0 +1,216 @@ +export interface DeliveryInvocation { + id: string; + token?: string; + attempt: number; + start: number; + end?: number; + stage: 'progress' | 'log' | 'return'; + outcome?: 'returned' | 'threw'; + error?: string; + result?: number; +} + +export interface DeliveryFailure { + id?: string; + token?: string; + at: number; + error: string; +} + +interface TimelineEntry { + state: string; + timestamp: number; + attempt?: number; + error?: string; +} + +export interface DurableDeliveryEvent { + id: number; + jobId: string; + type: string; + at: number; + attempt?: number; + error?: string; + result?: unknown; + timeline: TimelineEntry[]; +} + +export interface DeliveryHistory { + invocations: DeliveryInvocation[]; + failures: DeliveryFailure[]; + unexpected: unknown[]; + events: DurableDeliveryEvent[]; + jobs: { + id: string; + state: string; + attempt: number; + result: unknown; + timeline: TimelineEntry[]; + }[]; + completions: { id: string; result: unknown }[]; +} + +/** This fixture permits only an applied, pre-log Progress failure followed by retry. */ +export function assertPublicDeliveryHistory( + history: DeliveryHistory, + expected: ReadonlyMap +): void { + const check: (condition: unknown, message: string) => asserts condition = ( + condition, + message + ) => { + if (!condition) throw new Error(`${message}\n${JSON.stringify(history)}`); + }; + check(history.unexpected.length === 0, 'Unexpected Worker error or stalled recovery'); + for (const entries of [history.invocations, history.jobs, history.completions]) { + check( + entries.every(({ id }) => expected.has(id)), + 'Unexpected job ID' + ); + } + check( + history.events.every(({ jobId }) => expected.has(jobId)), + 'Unexpected event job ID' + ); + check(history.jobs.length === expected.size, 'Terminal job count mismatch'); + check(history.completions.length === expected.size, 'Completion tombstone count mismatch'); + const tokens = new Set(); + for (const invocation of history.invocations) { + check( + typeof invocation.token === 'string' && + invocation.token.trim() && + !tokens.has(invocation.token), + 'Missing or repeated delivery token' + ); + tokens.add(invocation.token); + } + let totalFailures = 0; + for (const [id, result] of expected) { + const calls = history.invocations.filter((entry) => entry.id === id); + const events = history.events.filter((event) => event.jobId === id); + const transitions = events.filter(({ type }) => type !== 'pushed' && type !== 'progress'); + const pulls = events.filter(({ type }) => type === 'pulled'); + const failures = events.filter(({ type }) => type === 'failed'); + const retries = events.filter(({ type }) => type === 'retried'); + const completed = events.filter(({ type }) => type === 'completed'); + check(calls.length > 0 && pulls.length === calls.length, `${id}: invocation/pull mismatch`); + check( + failures.length === calls.length - 1 && retries.length === failures.length, + `${id}: unexplained repeated delivery or missing durable failure` + ); + check(completed.length === 1, `${id}: duplicate or missing durable completion`); + check( + transitions.map(({ type }) => type).join(',') === + [ + ...calls.flatMap((_, index) => (index ? ['retried', 'failed', 'pulled'] : ['pulled'])), + 'completed', + ].join(','), + `${id}: unexpected durable transition (including recovery/release)` + ); + for (let index = 0; index < calls.length; index++) { + const call = calls[index]; + check(call.end !== undefined && call.end > call.start, `${id}: unfinished invocation`); + check( + index === 0 || calls[index - 1].end! < call.start, + `${id}: overlapping processor invocations` + ); + } + for (let index = 0; index < calls.length; index++) { + const call = calls[index]; + const pull = pulls[index]; + check(call.attempt === index && pull.attempt === index, `${id}: noncontiguous attempt`); + check(call.end !== undefined, `${id}: unfinished invocation`); + if (index === calls.length - 1) { + check( + call.outcome === 'returned' && call.stage === 'return' && call.result === result, + `${id}: incorrect processor result` + ); + continue; + } + check( + call.outcome === 'threw' && + call.stage === 'progress' && + call.error === 'Error: Internal server error', + `${id}: unexpected processor failure or stage` + ); + const applied = history.failures.filter( + (failure) => failure.id === id && failure.token === call.token + ); + check( + applied.length === 1 && + applied[0].error === call.error && + applied[0].at > call.end && + applied[0].at < calls[index + 1].start, + `${id}: missing or unordered applied Worker failure` + ); + const failure = failures[index]; + const retry = retries[index]; + const nextPull = pulls[index + 1]; + check( + failure.attempt === index + 1 && + retry.attempt === index + 1 && + failure.error === 'Internal server error', + `${id}: durable failure attempt/error mismatch` + ); + check( + pull.id < retry.id && + retry.id < failure.id && + failure.id < nextPull.id && + pull.at <= failure.at && + failure.at <= nextPull.at, + `${id}: durable failure must precede next pull` + ); + const failedTimeline = nextPull.timeline.filter(({ state }) => state === 'failed'); + check( + failedTimeline.length === index + 1 && + failedTimeline[index].attempt === index + 1 && + failedTimeline[index].timestamp === failure.at && + failedTimeline[index].error === failure.error, + `${id}: missing durable failure timeline` + ); + totalFailures++; + } + const terminal = history.jobs.filter((job) => job.id === id); + const tombstones = history.completions.filter((job) => job.id === id); + check( + terminal.length === 1 && + terminal[0].state === 'completed' && + terminal[0].attempt === calls.length - 1 && + terminal[0].result === result, + `${id}: incorrect terminal job` + ); + check( + tombstones.length === 1 && + tombstones[0].result === result && + completed[0].result === result && + completed[0].attempt === calls.length - 1 && + pulls.at(-1)!.id < completed[0].id, + `${id}: incorrect durable completion result or order` + ); + check( + terminal[0].timeline + .filter(({ state }) => state !== 'waiting' && state !== 'delayed') + .map(({ state }) => state) + .join(',') === + [ + ...calls.flatMap((_, index) => (index ? ['failed', 'active'] : ['active'])), + 'completed', + ].join(','), + `${id}: unexpected terminal timeline` + ); + const activeTimeline = terminal[0].timeline.filter(({ state }) => state === 'active'); + const failedTimeline = terminal[0].timeline.filter(({ state }) => state === 'failed'); + check( + activeTimeline.every((entry, index) => entry.timestamp === pulls[index].at) && + failedTimeline.every( + (entry, index) => + entry.timestamp === failures[index].at && + entry.attempt === index + 1 && + entry.error === failures[index].error + ) && + terminal[0].timeline.at(-1)?.timestamp === completed[0].at, + `${id}: terminal timeline does not match durable transitions` + ); + } + check(history.failures.length === totalFailures, 'Unmatched Worker failure'); +} diff --git a/test/support/postgres-public-delivery-trace.ts b/test/support/postgres-public-delivery-trace.ts new file mode 100644 index 00000000..969f1bc1 --- /dev/null +++ b/test/support/postgres-public-delivery-trace.ts @@ -0,0 +1,124 @@ +import { SQL } from 'bun'; +import { + decodePostgresEvent, + decodePostgresJob, + decodePostgresValue, +} from '../../src/infrastructure/persistence/postgres/codec'; +import type { PostgresJobRow } from '../../src/infrastructure/persistence/postgres/types'; +import { + assertPublicDeliveryHistory, + type DeliveryHistory, + type DeliveryInvocation, +} from './postgres-public-delivery-oracle'; + +/** A separate record for every invocation preserves predecessor evidence on retry. */ +export class PublicDeliveryTrace { + private readonly history: Pick = { + invocations: [], + failures: [], + unexpected: [], + }; + + start(job: { id: string; token?: string; attemptsMade: number }): DeliveryInvocation { + const invocation: DeliveryInvocation = { + id: job.id, + token: job.token, + attempt: job.attemptsMade, + start: performance.now(), + stage: 'progress', + }; + this.history.invocations.push(invocation); + return invocation; + } + + returned(invocation: DeliveryInvocation, result: number): number { + Object.assign(invocation, { + end: performance.now(), + stage: 'return', + outcome: 'returned', + result, + }); + return result; + } + + threw(invocation: DeliveryInvocation, error: unknown): void { + Object.assign(invocation, { end: performance.now(), outcome: 'threw', error: String(error) }); + } + + failed(job: { id: string; token?: string } | undefined, error: unknown): void { + this.history.failures.push({ + id: job?.id, + token: job?.token, + error: String(error), + at: performance.now(), + }); + } + + unexpected(kind: string, detail: unknown): void { + this.history.unexpected.push({ at: performance.now(), kind, detail: String(detail) }); + } + + async assert( + url: string, + namespace: string, + queue: string, + expected: ReadonlyMap + ): Promise { + const sql = new SQL(url, { max: 1, connectionTimeout: 5 }); + try { + // Read terminal state and its committed events from the same snapshot. + const durable = await sql.begin(async (tx) => { + await tx`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY`; + await tx`SET LOCAL statement_timeout = '5s'`; + const rows = await tx` + SELECT * FROM bunqueue_jobs WHERE namespace = ${namespace} AND queue = ${queue} ORDER BY id`; + const eventRows = await tx< + { + id: number; + queue: string; + event_type: string; + job_id: string; + occurred_at: number; + payload: Uint8Array | null; + }[] + >` + SELECT * FROM bunqueue_events WHERE namespace = ${namespace} AND queue = ${queue} ORDER BY id`; + const completions = await tx<{ job_id: string; result: Uint8Array | null }[]>` + SELECT job_id, result FROM bunqueue_completions + WHERE namespace = ${namespace} AND queue = ${queue} ORDER BY job_id`; + return { + jobs: rows.map((row) => { + const stored = decodePostgresJob(row); + return { + id: row.id, + state: row.state, + attempt: row.attempts, + result: stored.result, + timeline: stored.job.timeline, + }; + }), + events: eventRows.map((row) => { + const event = decodePostgresEvent(row); + return { + id: event.id, + jobId: String(event.jobId), + type: event.type, + at: event.occurredAt, + attempt: event.job?.attempts, + error: event.error, + result: event.result, + timeline: event.job?.timeline ?? [], + }; + }), + completions: completions.map((row) => ({ + id: row.job_id, + result: decodePostgresValue(row.result, null, `deliveryCompletion:${row.job_id}`), + })), + }; + }); + assertPublicDeliveryHistory({ ...this.history, ...durable }, expected); + } finally { + await sql.close({ timeout: 5 }); + } + } +} diff --git a/test/support/race-conditions-chaos.ts b/test/support/race-conditions-chaos.ts new file mode 100644 index 00000000..03855ce6 --- /dev/null +++ b/test/support/race-conditions-chaos.ts @@ -0,0 +1,118 @@ +import { expect } from 'bun:test'; +import type { QueueManager } from '../../src/application/queueManager'; +import type { JobId } from '../../src/domain/types/job'; + +export function raceTraceJson(value: unknown): string { + return JSON.stringify(value, (_, item) => (typeof item === 'bigint' ? String(item) : item)); +} + +const states = [ + ['waiting', 'waiting'], + ['prioritized', 'prioritized'], + ['delayed', 'delayed'], + ['active', 'active'], + ['completed', 'completed'], + ['failed', 'dlq'], + ['waiting-children', 'waiting-children'], +] as const; + +/** Compare independent public listings and counters, retaining every original ID exactly once. */ +export function assertRaceConservation(manager: QueueManager, queue: string, ids: JobId[]) { + // A retry can become due between synchronous reads. Repeat only if this snapshot + // crossed a runAt boundary; never retry a count/ID assertion failure. + for (let attempt = 0; attempt < 100; attempt++) { + const started = Date.now(); + // A crossing retry can be absent from both the earlier waiting read and + // the later delayed read, so deadlines must come from an unfiltered read. + const deadlines = manager.getJobs(queue, { end: ids.length + 1 }).map((job) => job.runAt); + const buckets = states.map(([state, counter]) => ({ + state, + counter, + jobs: manager.getJobs(queue, { state, end: ids.length + 1 }), + })); + const stats = manager.getStats(); + const ended = Date.now(); + if (deadlines.some((runAt) => runAt > started && runAt <= ended)) { + continue; + } + const membership = buckets.map(({ state, jobs }) => ({ + state, + ids: jobs.map((job) => job.id), + })); + try { + const retained = buckets.flatMap(({ jobs }) => jobs.map((job) => job.id)); + expect(new Set(ids).size).toBe(ids.length); + expect(retained).toHaveLength(ids.length); + expect(new Set(retained).size).toBe(ids.length); + expect(new Set(retained)).toEqual(new Set(ids)); + for (const { counter, jobs } of buckets) expect(stats[counter]).toBe(jobs.length); + expect(states.reduce((sum, [, counter]) => sum + stats[counter], 0)).toBe(ids.length); + return { stats, membership }; + } catch (cause) { + throw new Error(`Job conservation failed: ${raceTraceJson({ stats, membership, ids })}`, { + cause, + }); + } + } + throw new Error('Could not capture a stable job-state snapshot'); +} + +export type ChaosAction = 'pull' | 'ack' | 'fail'; + +/** Test-owned xorshift32: never changes runtime backoff jitter or global Math.random. */ +export function chaosActions(seed: number): ChaosAction[] { + let value = seed | 0; + return Array.from({ length: 200 }, () => { + value ^= value << 13; + value ^= value >>> 17; + value ^= value << 5; + return (['pull', 'ack', 'fail'] as const)[(value >>> 0) % 3]; + }); +} + +export async function runChaos(manager: QueueManager, queue: string, actions: ChaosAction[]) { + const started = performance.now(); + const trace = actions.map((action, index) => ({ + index, + action, + startMs: 0, + endMs: 0, + id: null as JobId | null, + outcome: 'pending', + attempts: null as number | null, + runAt: null as number | null, + })); + const results = await Promise.allSettled( + trace.map(async (entry) => { + entry.startMs = performance.now() - started; + try { + const job = await manager.pull(queue, entry.action === 'pull' ? 5 : 0); + if (!job) { + entry.outcome = 'empty'; + return; + } + entry.id = job.id; + if (entry.action === 'ack') expect(await manager.ack(job.id, {})).toBeUndefined(); + if (entry.action === 'fail') expect(await manager.fail(job.id, 'chaos')).toBeUndefined(); + entry.outcome = await manager.getJobState(job.id); + if (entry.action === 'ack') expect(entry.outcome).toBe('completed'); + const retained = await manager.getJob(job.id); + entry.attempts = retained?.attempts ?? null; + entry.runAt = retained?.runAt ?? null; + } catch (error) { + entry.outcome = String(error); + throw error; + } finally { + entry.endMs = performance.now() - started; + } + }) + ); + const failures = results.filter((result) => result.status === 'rejected'); + if (failures.length) { + throw new AggregateError( + failures.map((result) => result.reason), + JSON.stringify(trace) + ); + } + return trace; +} diff --git a/test/support/race-conditions-clock.ts b/test/support/race-conditions-clock.ts new file mode 100644 index 00000000..456d9610 --- /dev/null +++ b/test/support/race-conditions-clock.ts @@ -0,0 +1,27 @@ +import { setSystemTime, spyOn } from 'bun:test'; +import type { QueueManager } from '../../src/application/queueManager'; + +/** Advance only the clock: each listing still reads the real manager and SQLite. */ +export function crossRetryDeadlineAfterWaitingRead( + manager: QueueManager, + runAt: number, + read: () => T +): { value: T; crossed: boolean } { + const getJobs = manager.getJobs.bind(manager); + let crossed = false; + const listing = spyOn(manager, 'getJobs').mockImplementation((queue, options = {}) => { + const jobs = getJobs(queue, options); + if (!crossed && options.state === 'waiting') { + crossed = true; + setSystemTime(new Date(Math.ceil(runAt) + 1)); + } + return jobs; + }); + try { + setSystemTime(new Date(Math.floor(runAt) - 1)); + return { value: read(), crossed }; + } finally { + listing.mockRestore(); + setSystemTime(); + } +} diff --git a/test/support/race-conditions-fixture.ts b/test/support/race-conditions-fixture.ts new file mode 100644 index 00000000..bc4c7c72 --- /dev/null +++ b/test/support/race-conditions-fixture.ts @@ -0,0 +1,26 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { QueueManager } from '../../src/application/queueManager'; + +/** Each test owns its SQLite files, even when different race suites run in parallel. */ +export function raceConditionsFixture(options: ConstructorParameters[0] = {}) { + let directory: string | undefined; + let manager: QueueManager | undefined; + return { + async create() { + directory = await mkdtemp(join(tmpdir(), 'bunqueue-race-')); + manager = new QueueManager({ ...options, dataPath: join(directory, 'queue.db') }); + return manager; + }, + async close() { + try { + manager?.shutdown(); + } finally { + manager = undefined; + if (directory) await rm(directory, { recursive: true, force: true }); + directory = undefined; + } + }, + }; +} From ac1cf0ca7ba05657f2e3bd1c45ec8102884f2d14 Mon Sep 17 00:00:00 2001 From: egeominotti <2592071+egeominotti@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:23:25 +0100 Subject: [PATCH 2/2] docs: record retry and conservation validation evidence --- docs/src/content/docs/changelog.md | 3 ++ feature_list.json | 20 +++++------ progress.md | 57 +++++++++++++++++++++++------- 3 files changed, 58 insertions(+), 22 deletions(-) diff --git a/docs/src/content/docs/changelog.md b/docs/src/content/docs/changelog.md index 170f931f..d8ca372d 100644 --- a/docs/src/content/docs/changelog.md +++ b/docs/src/content/docs/changelog.md @@ -214,6 +214,9 @@ head: ### Testing +- Record the completed retry-delivery and job-conservation test corrections + with their isolated gate results, case-preservation checks, skeptic approvals + and green source CI evidence. - Test assertions now observe real lease ownership and renewal, rate-limit admission/refill, concurrency-policy removal, exact DLQ pagination, stored job logs, waiting-event IDs and forced closure with a running processor. Fixture diff --git a/feature_list.json b/feature_list.json index 37dff1a8..33b4c848 100644 --- a/feature_list.json +++ b/feature_list.json @@ -37,9 +37,9 @@ "name": "Resolve PostgreSQL delivery and chaos accounting CI anomalies", "description": "Investigate and resolve postgres-public-api-processing-flake and race-chaos-accounting-flake with isolated deterministic reproductions, complete job-state evidence and regression coverage. Preserve processing exclusivity and full conservation assertions; distinguish fixture defects from engine defects before changing behavior.", "dependencies": [], - "status": "in-progress", + "status": "done", "branch": "fix-ci-delivery-accounting", - "evidence": "PARTIAL 2026-11-14, base42225c31, branchfix-ci-delivery-accounting; code implemented, exact-head CI pending. Real SQLite RED with original default backoff/waits/drain: all100IDs49completed+51delayed, old>=50 fails49; strict conservation GREEN. Read-clock boundary RED expected1got0, fixed by independent retry deadlines; missing-job mutation rejected. All18original race paths/options retained,17other bodies token-identical; final native family21/0/564, focused Linux17/0/559. Real PG17 row-lock RED proves applied Progress-failure retry token0/attempt0 -> newtoken/attempt1, result14/log1, durablefailed1/completed1, then oldonce1fails2. Strict checker21cases (19negative/2positive). FINAL owned-Linux PG16.16/17.12/18.7 each571pass/0fail/1opt-insoakskip,572tests126files. Artifacts artifacts/ci-delivery-accounting/. Historical CI43559077874 cause remains unclassified; no runtime/lease/SDK changes. Full bun run test:sandbox PASS on treee2c50d5737b2dff3d2efc2c4ad3e96318cb54a4a: unit11931/0/404,TCP553/0(69suites),embedded349/0(43),0errors/anomalies/OOM,423validsamples,6.79min; artifacts/test-sandbox/2026-11-14T10-45-13.122Z/. Fresh native Ubuntu24.04.5arm64 Machine gate PASS same tree: Bun1.4.4+38afee6c8,4CPU/16GiB,unit12062/0/404,TCP553/0,embedded349/0,tinyfs9/0,type890+2163/format2373,15m40s,matching snapshot/logs copied/Machine deleted; artifacts/machines/20261114T104513Z/manifest.md. ./init.sh green11s. Model/SDK standalone gates not triggered (tests/docs/state only; model runs inside unit gate). Complete21-file skeptic review found no blocking code issue; final evidence-delta approval/commit/push/CI remain pending.", + "evidence": "2026-11-14 completed; base42225c31, branchfix-ci-delivery-accounting, implementation bca26a94. Real SQLite RED with original default backoff/waits/drain: all100IDs49completed+51delayed, old>=50 fails49; strict conservation GREEN. Read-clock boundary RED expected1got0, fixed by independent retry deadlines; missing-job mutation rejected. All18original race paths/options retained,17other bodies token-identical; final native family21/0/564, focused Linux17/0/559. Real PG17 row-lock RED proves applied Progress-failure retry token0/attempt0 -> newtoken/attempt1, result14/log1, durablefailed1/completed1, then oldonce1fails2. Strict checker21cases (19negative/2positive). FINAL owned-Linux PG16.16/17.12/18.7 each571pass/0fail/1opt-insoakskip,572tests126files. Artifacts artifacts/ci-delivery-accounting/. Historical CI43559077874 cause remains unclassified; no runtime/lease/SDK changes. Full bun run test:sandbox PASS on treee2c50d5737b2dff3d2efc2c4ad3e96318cb54a4a: unit11931/0/404,TCP553/0(69suites),embedded349/0(43),0errors/anomalies/OOM,423validsamples,6.79min; artifacts/test-sandbox/2026-11-14T10-45-13.122Z/. Fresh native Ubuntu24.04.5arm64 Machine gate PASS same tree: Bun1.4.4+38afee6c8,4CPU/16GiB,unit12062/0/404,TCP553/0,embedded349/0,tinyfs9/0,type890+2163/format2373,15m40s,matching snapshot/logs copied/Machine deleted; artifacts/machines/20261114T104513Z/manifest.md. ./init.sh green11s. Model/SDK standalone gates not triggered (tests/docs/state only; model runs inside unit gate). CI43588585824 attempt1 on implementation bca26a9408cea04e443808f8af12351e67ae149a GREEN (ci-watch.sh exit0): https://github.com/egeominotti/bunqueue/actions/runs/43588585824 ;26successful jobs/9expected skips,unit12066/0/400,TCP553/0,embedded349/0,PG16/17/18.6 each571/0/1;all SDK/type/format/docs jobs passed. Artifacts artifacts/ci-delivery-accounting/source-ci-*.log/json. Pre-commit and separate pre-push skeptic PASS on exact implementation tree6a934451d7d247f8fe137170994875d455b65c82. Post-gate source delta6a934451 changes only changelog/state, gate-delta.sh exit3;408/0 readers,259 release posts and64 docs assets checked;latest init green9s. This final state-record commit and authorized PR192 integration follow without executable changes.", "issue": null }, { @@ -233,11 +233,11 @@ }, { "id": "postgres-public-api-processing-flake", - "name": "Reproduce intermittent PostgreSQL 17 public-worker processing counts", - "description": "Reproduce first: test/postgres-public-api-queue-worker.test.ts, shares Queue, Worker, QueueEvents, progress, results, and logs across four brokers, failed at line77 in PostgreSQL 17.x Integration Tests of PR191 CI run43559077874 attempt1 on dd147ccf (548pass/1fail/1skip). All24 completed/event IDs and executions.size24 passed, but every processor count===1 was false (10.928s); the same case passed494.55ms in source run43557827964 and passed the single failed-job rerun in attempt2. Runtime and test files were unchanged by dd147ccf. Preserve the assertion. Repeat on a fresh isolated PostgreSQL17 four-broker fixture under controlled load, recording affected IDs, processor counts, failed/stalled reasons, attempts, lock tokens, lease deadlines and timestamps. The processor increments before awaiting progress/log commands and default3attempts permit ordinary retries, so repeated invocation alone does not prove concurrent delivery. Distinguish a legitimate failure/recovery retry, a fixture assumption and an engine divergence before changing tests or runtime; never use the user's existing database or weaken the invariant.", + "name": "Validate PostgreSQL public-worker delivery per applied attempt", + "description": "Correct the demonstrated absolute processor-once oracle defect under a real Progress SQL rollback and the default job retry policy. Require nonempty distinct tokens, contiguous attempts, nonoverlapping processor calls, matching applied Worker FAIL and durable failure/retry/pull timelines, exact terminal results and one durable completed event/tombstone per original ID. Nineteen malformed-history cases and two valid histories protect the checker; a real four-broker row-lock regression proves rollback at attempt0 followed by a new-token attempt1 and one result/log. Unexplained repeats, recovery and other error stages still fail. Historical CI43559077874 cause remains unclassified because the original logs lacked per-attempt evidence; no runtime or lease settings changed.", "dependencies": [], - "status": "not-started", - "evidence": "2026-11-14, observation only, not fixed. CI43559077874 attempt1 job132338509583 failed; exactly one gh run rerun --failed, attempt2 job132340799679 passed; ci-watch.sh --run43559077874 exit3FLAKY,26validation jobs successful/9release skips. https://github.com/egeominotti/bunqueue/actions/runs/43559077874 . Raw failure/rerun summaries and watch logs artifacts/tests-quality-closure-ci-*.log/json; prior source PG17 log artifacts/tests-quality-source-ci-pg17.log. Read-only contract analysis .superpowers/sdd/2026-11-14-test-quality-speedups/pg17-delivery-ci-analysis.md; cause unproved, no assertion/runtime change.", + "status": "done", + "evidence": "Initial observation before the follow-up: 2026-11-14, observation only, not fixed. CI43559077874 attempt1 job132338509583 failed; exactly one gh run rerun --failed, attempt2 job132340799679 passed; ci-watch.sh --run43559077874 exit3FLAKY,26validation jobs successful/9release skips. https://github.com/egeominotti/bunqueue/actions/runs/43559077874 . Raw failure/rerun summaries and watch logs artifacts/tests-quality-closure-ci-*.log/json; prior source PG17 log artifacts/tests-quality-source-ci-pg17.log. Read-only contract analysis .superpowers/sdd/2026-11-14-test-quality-speedups/pg17-delivery-ci-analysis.md; cause unproved, no assertion/runtime change. Resolved oracle defect 2026-11-14 by ci-delivery-accounting-fixes in bca26a94. Real PG17 RED: old once1 rejects two evidenced legal attempts after a five-second SQL row-lock timeout; result14/log1, durablefailed1/completed1. Strict history GREEN on complete local PG16.16/17.12/18.7 suites, each571/0/1, and source CI PG16/17/18.6 each571/0/1. CI43588585824 attempt1 on implementation bca26a9408cea04e443808f8af12351e67ae149a GREEN (ci-watch.sh exit0): https://github.com/egeominotti/bunqueue/actions/runs/43588585824 ;26successful jobs/9expected skips,unit12066/0/400,TCP553/0,embedded349/0,PG16/17/18.6 each571/0/1;all SDK/type/format/docs jobs passed. Artifacts artifacts/ci-delivery-accounting/source-ci-*.log/json. Pre-commit and separate pre-push skeptic PASS on exact implementation tree6a934451d7d247f8fe137170994875d455b65c82. Shared full sandbox/fresh Ubuntu evidence and precise historical-cause limitation are recorded in ci-delivery-accounting-fixes. No engine fix is claimed. Commands: bun run test:postgres; bun run test:sandbox; bun run typecheck; bun run format:check; git diff --check. Exact commands, gated tree hashes, isolated totals and artifact paths are detailed in ci-delivery-accounting-fixes.", "issue": null }, { @@ -279,11 +279,11 @@ }, { "id": "race-chaos-accounting-flake", - "name": "Reproduce the legacy random chaos accounting failure", - "description": "Reproduce first: test/race-conditions.test.ts:606, Race Condition Tests > Mixed Concurrent Operations Stress Test > should maintain job count invariant under chaos, failed in Unit Tests job132344370497 of PR191 CI43560970951 attempt1 on1037ddd3: expected completed+dlq+active>=50, received49 (407.27ms). The unchanged code passed its single unit-job rerun in attempt2. It pushes100jobs with maxAttempts10, uses200unseeded Math.random pull/ack/fail operations, sleeps100ms and stops draining at the first empty200ms pull. Waiting excludes delayed and fail schedules retries1–3seconds later; a valid49counted+51delayed snapshot conserves100jobs but fails this predicate. CI did not print the state vector or IDs, so that explanation is not a proven cause and loss is not established. Preserve the failing assertion while diagnosing. Use a fresh isolated SQLite path, a recorded seed/operation trace and all100IDs; reproduce the valid-state counterexample, then distinguish omitted delayed work from an engine divergence. Any test fix must assert complete state and per-ID conservation instead of lowering the threshold, swallowing errors or using another arbitrary sleep.", + "name": "Count every retained job and state during concurrent chaos", + "description": "Replace the unseeded partial half-count oracle with 200 test-owned seeded operations and exact original-ID, mutually exclusive state-list and counter conservation across waiting, prioritized, delayed, active, completed, failed/DLQ and waiting-children. Propagate operation failures; do not treat an empty pull as proof that delayed work vanished. Real SQLite regressions cover 49completed+51delayed, rejection of one missing original job despite99completions, and a retry deadline crossing between waiting/delayed reads. Independent unfiltered deadlines guard only unstable snapshots, never assertion failures. Preserve all18 original case paths/options and17 other bodies while splitting files and giving each test its own temporary SQLite directory.", "dependencies": [], - "status": "not-started", - "evidence": "2026-11-14, observation only, not fixed. CI43560970951 attempt1 unit12041pass/1fail/399skip,488679assertions1127files479.59s; exactly one gh run rerun --failed, attempt2GREEN, ci-watch.sh exit3FLAKY. https://github.com/egeominotti/bunqueue/actions/runs/43560970951 . All other product/PG16-18/SDK/type/docs/format jobs passed; this commit changed metadata only and the test was untouched. Raw artifacts/tests-quality-final-pr-ci-unit-{failure.log,attempt1-summary.json,attempt2-summary.json,rerun-watch.log}; read-only .superpowers/sdd/2026-11-14-test-quality-speedups/chaos-accounting-ci-analysis.md confirms omitted-state possibility, not the actual CI cause.", + "status": "done", + "evidence": "Initial observation before the follow-up: 2026-11-14, observation only, not fixed. CI43560970951 attempt1 unit12041pass/1fail/399skip,488679assertions1127files479.59s; exactly one gh run rerun --failed, attempt2GREEN, ci-watch.sh exit3FLAKY. https://github.com/egeominotti/bunqueue/actions/runs/43560970951 . All other product/PG16-18/SDK/type/docs/format jobs passed; this commit changed metadata only and the test was untouched. Raw artifacts/tests-quality-final-pr-ci-unit-{failure.log,attempt1-summary.json,attempt2-summary.json,rerun-watch.log}; read-only .superpowers/sdd/2026-11-14-test-quality-speedups/chaos-accounting-ci-analysis.md confirms omitted-state possibility, not the actual CI cause. Resolved oracle defect 2026-11-14 by ci-delivery-accounting-fixes in bca26a94. Isolated Linux RED with original default backoff,100ms wait and first-empty200ms drain: all100IDs retained as49completed/51delayed, old>=50 fails49. New exact conservation GREEN; real missing-job mutation is rejected. Snapshot-boundary RED expected1got0, independent-deadline fix GREEN. Final native family21/0/564assertions; focused Linux17/0/559 excludes four benchmark-labelled cases. AST audit retains18original names/hierarchy/options and17unchanged bodies. CI43588585824 attempt1 on implementation bca26a9408cea04e443808f8af12351e67ae149a GREEN (ci-watch.sh exit0): https://github.com/egeominotti/bunqueue/actions/runs/43588585824 ;26successful jobs/9expected skips,unit12066/0/400,TCP553/0,embedded349/0,PG16/17/18.6 each571/0/1;all SDK/type/format/docs jobs passed. Artifacts artifacts/ci-delivery-accounting/source-ci-*.log/json. Pre-commit and separate pre-push skeptic PASS on exact implementation tree6a934451d7d247f8fe137170994875d455b65c82. Shared full sandbox/fresh Ubuntu evidence in ci-delivery-accounting-fixes; no runtime change. Commands: bun run test:sandbox; bun run typecheck; bun run format:check; git diff --check. Exact commands, gated tree hashes, isolated totals and artifact paths are detailed in ci-delivery-accounting-fixes.", "issue": null }, { diff --git a/progress.md b/progress.md index 3d49a220..fe3300ef 100644 --- a/progress.md +++ b/progress.md @@ -17,6 +17,14 @@ overview: [docs/agents/README.md](docs/agents/README.md). ### What's Done +- `ci-delivery-accounting-fixes`, `postgres-public-api-processing-flake` and + `race-chaos-accounting-flake`: implementation `bca26a94`, strict retry/durable + completion histories and exact all-state/ID conservation, all18 original race + cases retained. Sandbox/fresh Ubuntu gates passed; source PR #192 CI + [43588585824](https://github.com/egeominotti/bunqueue/actions/runs/43588585824) + attempt1 GREEN: unit12066/0/400, all PostgreSQL16/17/18.6 legs571/0/1 each, + TCP553/0,embedded349/0 and all SDK/type/format/docs jobs passed. Pre-commit and + separate pre-push skeptic PASS. Final state-record CI and integration follow. - `test-quality-speedups`: implementation `aa038a83`, all four task reviews and whole-branch/pre-push skeptic PASS; full sandbox and fresh Ubuntu green on tree `f3c1bdf4`. Source PR #191 CI @@ -62,11 +70,8 @@ overview: [docs/agents/README.md](docs/agents/README.md). ### What's In Progress -- `ci-delivery-accounting-fixes`, branch `fix-ci-delivery-accounting`, starts - from merged main `42225c31`. Both oracle defects have isolated RED/GREEN - evidence; strict delivery history and complete conservation checks are - implemented. PostgreSQL16–18, full sandbox and fresh Ubuntu gates passed; - final skeptic evidence-delta approval, commit/push and exact-head CI are next. +The delivery/accounting implementation is done and source CI is green. The +remaining PR #192 integration is described below. The three audit implementations are done, backed by green source CI. Their original staged worktrees remain preserved. Their integration is complete; @@ -74,11 +79,10 @@ other sessions' active work remains tracked in `feature_list.json`. ### What's Next -1. Finish `ci-delivery-accounting-fixes`: review the complete PostgreSQL16–18 - results, sandbox and fresh Ubuntu gates, obtain the mandatory skeptic PASS - before each authorized commit/push, then watch exact-head CI and integrate. - PR #191 already merged as `42225c31`; its main CI43563916357 is green and - the original main checkout was aligned. No version/publication action. +1. Validate/push the final evidence record for PR #192, watch its exact-head CI, + perform the authorized normal squash merge, watch main CI and fast-forward + the original main checkout. The implementation's CI43588585824 is already + GREEN on `bca26a94`. No version/publication action. 2. `postgres-completion-recheck-state-index`: the prune recheck's consumer-state probe on PostgreSQL 16 and 17 with stale statistics. 3. Owner decisions: `postgres-completion-read-model-plan` (trade-off measured, @@ -89,8 +93,8 @@ other sessions' active work remains tracked in `feature_list.json`. - The legacy partial-count predicate from CI43560970951 has been reproduced with all100IDs retained (49completed/51delayed), then replaced with exact - conservation checks under `ci-delivery-accounting-fixes`; its source CI is - pending. The original CI snapshot/seed were not retained. + conservation checks under `ci-delivery-accounting-fixes`; its source + CI43588585824 is GREEN on `bca26a94`. The original CI snapshot/seed were not retained. - The PostgreSQL invocation-count oracle from CI43559077874 rejects a proven legal Progress-failure retry. Strict token/attempt/durable-history checks now distinguish that path from unexplained repeats, but the specific historical @@ -146,6 +150,35 @@ other sessions' active work remains tracked in `feature_list.json`. Newest first. Prepend one entry per session; for multi-session work fill the [`session-handoff.md`](session-handoff.md) sections into the entry. +### 2026-11-14 — retry and conservation corrections validated in source CI + +- Implementation `bca26a9408cea04e443808f8af12351e67ae149a`, tree6a934451, is + committed and pushed as `egeominotti` on `fix-ci-delivery-accounting`; + [PR #192](https://github.com/egeominotti/bunqueue/pull/192) targets main. + Separate mandatory pre-commit and pre-push skeptic verdicts are PASS. +- Exact-head CI43588585824 attempt1 GREEN, watcher exit0:26successful jobs, + 9expected release skips. Unit12066/0/400,486376assertions1136files472.87s; + TCP69/69suites553/553, embedded43/43suites349/349, PostgreSQL16/17/18.6 each + 571/0/1opt-insoakskip,572tests126files. Both affected cases and the real + SQL-lock regression passed in every PostgreSQL leg. All SDK, public API, + WebSocket/SSE, type, format and docs jobs passed. Separate automated code + review43588585715 succeeded without findings. +- All three work items are done on this implementation evidence. Gated + executable treee2c50d57, sandbox/fresh Ubuntu receipts, RED/GREEN sequences, + case-preservation audit and limitations are recorded in feature_list.json. + The final source post-gate delta6a934451 changes only changelog/state; + gate-delta.sh exit3,408/0readers,259releaseposts/64docsassets and init9s green. +- Current closure changes only the changelog and work-state records; no new + executable changes. Remaining integration: focused readers/delta proof, + skeptic before each authorized closure commit/push, exact-head PR CI, normal + squash merge, main CI and original-main alignment. Existing authorization + persists; no release/version/publication action. All owned Machines deleted. +- Evidence: artifacts/ci-delivery-accounting/source-ci-{final.json,watch.log, + unit.log,tcp.log,embedded.log,pg16.log,pg17.log,pg18.log}; isolated manifests + and current ledger are linked in the implementation entry below. Historical + CI43559077874 cause remains unclassified; the corrected checker rejects any + repeated delivery lacking the specifically proven failure/retry history. + ### 2026-11-14 — delivery and accounting oracle defects reproduced - User requested investigation and fixes for both recorded CI anomalies, within