From c491c6989628b1e7f9bca88f1c266a19b5be0af9 Mon Sep 17 00:00:00 2001 From: Rickylabs Date: Wed, 12 Aug 2026 09:52:47 +0200 Subject: [PATCH] fix(streams): distinguish producer refusal reasons --- .../slices/1405/context-pack.md | 14 ++ .../slices/1405/drift.md | 20 ++ .../slices/1405/supervisor.md | 14 ++ .../slices/1405/worklog.md | 67 ++++++ .../src/application/create-durable-stream.ts | 16 +- .../durable-stream-producer-supervisor.ts | 23 +- .../src/domain/producer-contract-v1.ts | 2 + ...-stream-producer-contract_behavior_test.ts | 198 +++++++++++++++++- 8 files changed, 335 insertions(+), 19 deletions(-) create mode 100644 .llm/runs/release-0.0.6-features--orchestration/slices/1405/context-pack.md create mode 100644 .llm/runs/release-0.0.6-features--orchestration/slices/1405/drift.md create mode 100644 .llm/runs/release-0.0.6-features--orchestration/slices/1405/supervisor.md create mode 100644 .llm/runs/release-0.0.6-features--orchestration/slices/1405/worklog.md diff --git a/.llm/runs/release-0.0.6-features--orchestration/slices/1405/context-pack.md b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/context-pack.md new file mode 100644 index 0000000000..2883843010 --- /dev/null +++ b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/context-pack.md @@ -0,0 +1,14 @@ +# Context pack — #1405 + +- Branch: `fix/1405-durable-producer-rejection-taxonomy` at baseline `01aa12b67`. +- Archetype: 3 — Runtime / Behavior; no scope overlay. +- Locked reasons: graceful close drain is `producer-stopping`; non-retryable transport failure is + `transport-refused`; retryable failures at `maxAttempts` remain `retry-exhausted`. +- Behavior boundary: reason strings only; no acceptance, retry-count, settlement class, delivery, + cancellation, telemetry-classification, #1398, or scaffold changes. +- PLAN-EVAL: N/A per the owner brief and orchestration supervisor record. +- Implementation gates: scoped wrappers, quality gate, target quality scan, doc lint, JSR audit, + focused tests, telemetry guard, and package-configured full suite are green. +- Gate caveat: the brief's bare `deno test packages/plugin-streams-core` exits 1 because it omits + the package suite's required env permission; the configured `--allow-all` task passes 33/33. +- Orchestrator retains slice review, IMPL-EVAL, ready-state, merge, and release authority. diff --git a/.llm/runs/release-0.0.6-features--orchestration/slices/1405/drift.md b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/drift.md new file mode 100644 index 0000000000..f75acee9c0 --- /dev/null +++ b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/drift.md @@ -0,0 +1,20 @@ +# Drift — #1405 + +## 2026-08-12 — bare test command lacks package permissions + +- Severity: minor gate-invocation mismatch. +- Expected: `deno test packages/plugin-streams-core` passes as written in the brief. +- Observed: it exits 1 with 19 `NotCapable` failures because tests read + `DURABLE_STREAMS_URL` and the command grants no env permission; 14 permission-free tests pass. +- Action: preserved the red evidence and ran the checked-in package task, which grants the package's + declared test permissions and passes 33/33. No test or permission surface was edited. + +## 2026-08-12 — root quality gate omits the target package + +- Severity: minor gate-coverage limitation. +- Expected: mandatory `quality:gate` provides changed-package quality evidence. +- Observed: it exits 0, but its configured scan/doctrine roots do not include + `packages/plugin-streams-core`. +- Action: ran an explicit target quality scan (`findings=[]`, `allowCount=0`) and target doctrine + audit. The doctrine audit has no failures and one 500-line-cap warning: the supervisor is 515 + lines after this locked change (baseline 497). A structural split is outside the no-refactor slice. diff --git a/.llm/runs/release-0.0.6-features--orchestration/slices/1405/supervisor.md b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/supervisor.md new file mode 100644 index 0000000000..8651922640 --- /dev/null +++ b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/supervisor.md @@ -0,0 +1,14 @@ +# Slice identity — #1405 + +| Field | Value | +| --- | --- | +| Lane | `light_implementation` | +| Generator | Codex · OpenAI · GPT-5.6 Sol · low | +| Worktree | `/home/codex/repos/ns006-1405` | +| Branch | `fix/1405-durable-producer-rejection-taxonomy` | +| Baseline | `origin/main@01aa12b67` | +| Issue | `#1405` | +| PLAN-EVAL | N/A — fully specified mechanical slice | +| IMPL-EVAL | Separate orchestrator-owned evaluator session | + +The milestone orchestrator retains slice review, merge, publish, and release authority. diff --git a/.llm/runs/release-0.0.6-features--orchestration/slices/1405/worklog.md b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/worklog.md new file mode 100644 index 0000000000..5c641fc6c4 --- /dev/null +++ b/.llm/runs/release-0.0.6-features--orchestration/slices/1405/worklog.md @@ -0,0 +1,67 @@ +# Worklog — #1405 durable producer rejection taxonomy + +## Design + +### Public surface + +- Add exactly `transport-refused` to `StreamWriteUnknownReasonV1`. +- Reuse the existing `producer-stopping` rejection reason during graceful close drain. +- Keep the existing root re-exports; add no entrypoint or export. + +### Domain vocabulary and ports + +- Closing intent is private supervisor state, visible from `close()` entry until shutdown settles. +- A non-retryable transport failure is a positive refusal; retryable failure at the attempt bound is + exhaustion. +- Existing transport, clock, random, queue, and lifecycle ports remain unchanged. + +### Constants + +- No new constant collection is required; the published reason unions remain the finite vocabulary. + +### Commit slice + +- S1 changes only reason selection and adds deterministic close/refusal/exhaustion tests. Proving + gates are the focused tests plus the complete gate set named in the slice brief. + +### Deferred scope + +- No acceptance, delivery, cancellation, retry-count, telemetry-classification, scaffold, #1398, + merge, or release changes. + +### Contributor path + +- Start with `producer-contract-v1.ts` for reason vocabulary, then follow supervisor settlement and + rejection selection into the contract behavior tests. + +## Phase status + +- `PLAN-EVAL: N/A` — the owner brief and orchestrator research fully specify the two selectors, + locked vocabulary, negative tests, boundaries, and gates. +- Implementation and generator gates complete on the assigned `light_implementation` lane. +- Separate orchestrator-owned slice review and IMPL-EVAL remain pending. + +## Evidence + +| Gate | Exit | Result | +| --- | ---: | --- | +| Focused contract behavior | 0 | 8 passed, 0 failed; includes all four new negative tests. | +| Existing telemetry classification guard | 0 | 2 passed, 0 failed. | +| Scoped check wrapper | 0 | 43 files, 1 batch, 0 failures/occurrences. | +| Scoped lint wrapper | 0 | 43 files, 1 batch, 0 occurrences. | +| Scoped format wrapper | 0 | 43 files, 1 batch, 0 findings. | +| Post-format reason grep | 0 | `transport-refused` and `producer-stopping` remain in contract, selector, and tests. | +| `deno task quality:gate` | 0 | `quality:scan` and `arch:check` completed; existing repository warnings only. The configured quality roots omit this package. | +| Explicit target quality scan | 0 | `packages/plugin-streams-core/src`; `findings=[]`, `allowCount=0`. | +| Explicit target doctrine audit | 0 | `FAIL=0 WARN=1 INFO=1`; supervisor is 515 lines versus the 500-line advisory cap, and architecture docs are informationally absent. | +| Full export-map doc lint | 0 | 4 entrypoints; `totalErrors=0`, `totalMissingJSDoc=0`. | +| JSR audit | 0 | dry-run OK; one non-failing slow-types banner warning. | +| Brief's exact `deno test packages/plugin-streams-core` | 1 | 14 passed, 19 failed solely with `NotCapable` because the command omits `--allow-env`. | +| Package-declared `deno task --cwd packages/plugin-streams-core test` | 0 | 33 passed, 0 failed. | + +## Reconcile + +- #1405 remains the only resolving issue; the draft PR must carry `Closes #1405` and target + `main`. No #1398 surface was touched. +- No new dependency, export-map key, lint suppression, unsafe cast, `any`, or architecture debt was + introduced. diff --git a/packages/plugin-streams-core/src/application/create-durable-stream.ts b/packages/plugin-streams-core/src/application/create-durable-stream.ts index eff1f487f5..e67867e7f9 100644 --- a/packages/plugin-streams-core/src/application/create-durable-stream.ts +++ b/packages/plugin-streams-core/src/application/create-durable-stream.ts @@ -129,7 +129,7 @@ export class DurableStreamProducer context?: StreamWriteContextV1, ): StreamWriteReceiptV1 { if (this.closed) { - return this.#reject(stateRejection(this.state.state)); + return this.#reject(this.#supervisor.writeRejectionReason() ?? 'producer-failed'); } const definition = this.#schema[entityType]; if (!definition) { @@ -157,7 +157,7 @@ export class DurableStreamProducer context?: StreamWriteContextV1, ): StreamWriteReceiptV1 { if (this.closed) { - return this.#reject(stateRejection(this.state.state)); + return this.#reject(this.#supervisor.writeRejectionReason() ?? 'producer-failed'); } const definition = this.#schema[entityType]; if (!definition) { @@ -248,18 +248,6 @@ export class DurableStreamProducer } } -function stateRejection( - state: StreamProducerStateSnapshotV1['state'], -): StreamWriteRejectionReasonV1 { - if (state === 'stopping') { - return 'producer-stopping'; - } - if (state === 'stopped') { - return 'producer-stopped'; - } - return 'producer-failed'; -} - function resolveRequiredStreamUrl(streamPath: string): string { try { return buildStreamUrl(streamPath); diff --git a/packages/plugin-streams-core/src/application/durable-stream-producer-supervisor.ts b/packages/plugin-streams-core/src/application/durable-stream-producer-supervisor.ts index 693a285952..c7d7c1f0a2 100644 --- a/packages/plugin-streams-core/src/application/durable-stream-producer-supervisor.ts +++ b/packages/plugin-streams-core/src/application/durable-stream-producer-supervisor.ts @@ -62,6 +62,7 @@ export class DurableStreamProducerSupervisor { bufferedBytes: 0, }; #accepted = true; + #closing = false; #epoch = 0; #nextSequence = 0; @@ -110,6 +111,11 @@ export class DurableStreamProducerSupervisor { return !this.#accepted; } + /** Explain why a new write cannot be accepted. */ + writeRejectionReason(): StreamWriteRejectionReasonV1 | undefined { + return this.#writeRejectionReason(); + } + /** Accept an already serialized event into the bounded FIFO. */ enqueue(body: string, lifecycle?: StreamProducerWriteLifecycle): StreamWriteReceiptV1 { const rejection = this.#writeRejectionReason(); @@ -213,6 +219,7 @@ export class DurableStreamProducerSupervisor { return Promise.resolve(); } this.#accepted = false; + this.#closing = true; this.#closePromise = this.#closeGracefully(); return this.#closePromise; } @@ -298,7 +305,11 @@ export class DurableStreamProducerSupervisor { if (this.#abort.signal.aborted) { return false; } - if (!isRetryable(connected.failure) || attempt === this.#reconnectPolicy.maxAttempts) { + if (!isRetryable(connected.failure)) { + this.#failActive(entry, connected.failure, attempt); + return false; + } + if (attempt === this.#reconnectPolicy.maxAttempts) { this.#failActive(entry, connected.failure, attempt); return false; } @@ -416,7 +427,11 @@ export class DurableStreamProducerSupervisor { ): void { this.#queue.settle(active, { status: 'delivery-unknown', - reason: failure.kind === 'aborted' ? 'transport-aborted' : 'retry-exhausted', + reason: failure.kind === 'aborted' + ? 'transport-aborted' + : !isRetryable(failure) + ? 'transport-refused' + : 'retry-exhausted', error: failure.message, }); this.#fail(`${failure.message} (attempt ${attempt})`); @@ -473,8 +488,10 @@ export class DurableStreamProducerSupervisor { return 'producer-stopping'; case 'stopped': return 'producer-stopped'; - default: + case 'failed': return 'producer-failed'; + default: + return this.#closing ? 'producer-stopping' : 'producer-failed'; } } diff --git a/packages/plugin-streams-core/src/domain/producer-contract-v1.ts b/packages/plugin-streams-core/src/domain/producer-contract-v1.ts index 283d44fb36..ea6f6e3224 100644 --- a/packages/plugin-streams-core/src/domain/producer-contract-v1.ts +++ b/packages/plugin-streams-core/src/domain/producer-contract-v1.ts @@ -83,6 +83,8 @@ export type StreamWriteCancellationReasonV1 = 'producer-stopped' | 'producer-fai /** Reasons an attempted write cannot be reported as delivered or rejected. */ export type StreamWriteUnknownReasonV1 = | 'retry-exhausted' + /** The transport positively refused the write with a non-retryable failure. */ + | 'transport-refused' | 'transport-aborted' | 'producer-stopped'; diff --git a/packages/plugin-streams-core/tests/application/durable-stream-producer-contract_behavior_test.ts b/packages/plugin-streams-core/tests/application/durable-stream-producer-contract_behavior_test.ts index 9b309142cc..60e3522145 100644 --- a/packages/plugin-streams-core/tests/application/durable-stream-producer-contract_behavior_test.ts +++ b/packages/plugin-streams-core/tests/application/durable-stream-producer-contract_behavior_test.ts @@ -1,4 +1,4 @@ -import { assertEquals } from '@std/assert'; +import { assertEquals, assertNotEquals } from '@std/assert'; import { DurableStreamProducer } from '../../src/application/create-durable-stream.ts'; import type { StreamProducerClockPort } from '../../src/ports/stream-producer-clock-port.ts'; import type { StreamProducerRandomPort } from '../../src/ports/stream-producer-random-port.ts'; @@ -17,16 +17,36 @@ const retryable: StreamProducerTransportResultV1 = { ok: false, failure: { kind: 'retryable', message: 'offline' }, }; +const refused: StreamProducerTransportResultV1 = { + ok: false, + failure: { kind: 'non-retryable', message: 'refused' }, +}; const acknowledged: StreamProducerTransportResultV1 = { ok: true, value: { duplicate: false }, }; +const appendRetryable: StreamProducerTransportResultV1 = { + ok: false, + failure: { kind: 'retryable', message: 'offline' }, +}; +const appendRefused: StreamProducerTransportResultV1 = { + ok: false, + failure: { kind: 'non-retryable', message: 'refused' }, +}; class ScriptedTransport implements StreamProducerTransportPort { readonly #connectResults: StreamProducerTransportResultV1[]; + readonly #appendResults: StreamProducerTransportResultV1[]; + appendCalls = 0; - constructor(connectResults: StreamProducerTransportResultV1[] = [connected]) { + constructor( + connectResults: StreamProducerTransportResultV1[] = [connected], + appendResults: StreamProducerTransportResultV1[] = [ + acknowledged, + ], + ) { this.#connectResults = [...connectResults]; + this.#appendResults = [...appendResults]; } connect(_input: StreamProducerConnectInputV1): Promise> { @@ -35,21 +55,53 @@ class ScriptedTransport implements StreamProducerTransportPort { append( _input: StreamProducerAppendInputV1, + ): Promise> { + this.appendCalls++; + return Promise.resolve(this.#appendResults.shift() ?? acknowledged); + } + + close( + _input: StreamProducerCloseInputV1, ): Promise> { return Promise.resolve(acknowledged); } +} + +class DrainControlledTransport implements StreamProducerTransportPort { + appendStarted = false; + #releaseAppend: (() => void) | undefined; + + connect(_input: StreamProducerConnectInputV1): Promise> { + return Promise.resolve(connected); + } + + append( + _input: StreamProducerAppendInputV1, + ): Promise> { + this.appendStarted = true; + return new Promise((resolve) => { + this.#releaseAppend = () => resolve(acknowledged); + }); + } close( _input: StreamProducerCloseInputV1, ): Promise> { return Promise.resolve(acknowledged); } + + releaseAppend(): void { + this.#releaseAppend?.(); + this.#releaseAppend = undefined; + } } class ManualClock implements StreamProducerClockPort { #release: (() => void) | undefined; + sleepCalls = 0; sleep(_delayMs: number, options: Readonly<{ signal?: AbortSignal }> = {}): Promise { + this.sleepCalls++; return new Promise((resolve, reject) => { this.#release = resolve; options.signal?.addEventListener( @@ -97,6 +149,16 @@ async function waitForState( } } +async function waitFor(predicate: () => boolean, description: string): Promise { + const deadline = Date.now() + 1_000; + while (!predicate()) { + if (Date.now() >= deadline) { + throw new Error(`timed out waiting for ${description}`); + } + await Promise.resolve(); + } +} + type ReturnTypeDefinition = { readonly execution: { readonly schema: ReturnType['execution']['schema']; @@ -213,3 +275,135 @@ Deno.test('readiness resolves only after a reconnect reaches ready', async () => restoreEnv(); } }); + +Deno.test('close drain rejects a concurrent write as producer-stopping', async () => { + const restoreEnv = setStreamsUrl(); + const transport = new DrainControlledTransport(); + const producer = new DurableStreamProducer({ + streamPath: '/contract/close-drain-rejection', + schema: createStreamTopicFixture(), + producerId: 'contract-close-drain-rejection', + transport, + }); + + try { + producer.upsert('execution', { id: 'draining' }); + await waitFor(() => transport.appendStarted, 'append to hold the close drain open'); + const closing = producer.close(); + + const rejected = producer.upsert('execution', { id: 'during-close' }); + const outcome = await rejected.completion; + assertEquals(outcome, { status: 'rejected', reason: 'producer-stopping' }); + if (outcome.status !== 'rejected') { + throw new Error(`expected rejected outcome, received ${outcome.status}`); + } + assertNotEquals(outcome.reason, 'producer-failed'); + + transport.releaseAppend(); + await closing; + } finally { + transport.releaseAppend(); + await producer.stop(); + restoreEnv(); + } +}); + +Deno.test('first append refusal settles transport-refused without retrying', async () => { + const restoreEnv = setStreamsUrl(); + const transport = new ScriptedTransport([connected], [appendRefused]); + const producer = new DurableStreamProducer({ + streamPath: '/contract/append-refused', + schema: createStreamTopicFixture(), + producerId: 'contract-append-refused', + reconnectPolicy: { maxAttempts: 3 }, + transport, + }); + + try { + const outcome = await producer.upsert('execution', { id: 'refused' }).completion; + assertEquals(outcome.status, 'delivery-unknown'); + if (outcome.status !== 'delivery-unknown') { + throw new Error(`expected delivery-unknown outcome, received ${outcome.status}`); + } + assertEquals(outcome.reason, 'transport-refused'); + assertNotEquals(outcome.reason, 'retry-exhausted'); + assertEquals(transport.appendCalls, 1); + } finally { + await producer.stop(); + restoreEnv(); + } +}); + +Deno.test('retryable append failures at maxAttempts settle retry-exhausted', async () => { + const restoreEnv = setStreamsUrl(); + const transport = new ScriptedTransport( + [connected, connected, connected], + [appendRetryable, appendRetryable, appendRetryable], + ); + const clock = new ManualClock(); + const producer = new DurableStreamProducer({ + streamPath: '/contract/append-retry-exhausted', + schema: createStreamTopicFixture(), + producerId: 'contract-append-retry-exhausted', + reconnectPolicy: { maxAttempts: 3 }, + transport, + clock, + random: new FixedRandom(), + }); + + try { + const receipt = producer.upsert('execution', { id: 'exhausted' }); + await waitFor(() => clock.sleepCalls === 1, 'first append retry backoff'); + clock.release(); + await waitFor(() => clock.sleepCalls === 2, 'second append retry backoff'); + clock.release(); + const outcome = await receipt.completion; + assertEquals(outcome.status, 'delivery-unknown'); + if (outcome.status !== 'delivery-unknown') { + throw new Error(`expected delivery-unknown outcome, received ${outcome.status}`); + } + assertEquals(outcome.reason, 'retry-exhausted'); + assertNotEquals(outcome.reason, 'transport-refused'); + assertEquals(transport.appendCalls, 3); + } finally { + clock.release(); + await producer.stop(); + restoreEnv(); + } +}); + +Deno.test('non-retryable reconnect failure settles transport-refused', async () => { + const restoreEnv = setStreamsUrl(); + const transport = new ScriptedTransport( + [connected, refused], + [appendRetryable], + ); + const clock = new ManualClock(); + const producer = new DurableStreamProducer({ + streamPath: '/contract/connect-refused', + schema: createStreamTopicFixture(), + producerId: 'contract-connect-refused', + reconnectPolicy: { maxAttempts: 3 }, + transport, + clock, + random: new FixedRandom(), + }); + + try { + const receipt = producer.upsert('execution', { id: 'connect-refused' }); + await waitFor(() => clock.sleepCalls === 1, 'append retry backoff'); + clock.release(); + const outcome = await receipt.completion; + assertEquals(outcome.status, 'delivery-unknown'); + if (outcome.status !== 'delivery-unknown') { + throw new Error(`expected delivery-unknown outcome, received ${outcome.status}`); + } + assertEquals(outcome.reason, 'transport-refused'); + assertNotEquals(outcome.reason, 'retry-exhausted'); + assertEquals(transport.appendCalls, 1); + } finally { + clock.release(); + await producer.stop(); + restoreEnv(); + } +});