From ddb8cda8e98d39009e906bb5bf990e9d66ab255a Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 13:48:46 +0100 Subject: [PATCH 01/15] fix(labeler): redrive the subscription-DO notify on console mutation replay A transient afterCommit failure after a label commits leaves the row publication_pending=1 and drops the live broadcast; the guard replay branch did not re-drive the notify, so a client retry never recovered it. The label-issue and emergency replay branches now re-fire the post-commit notify (keyed on the committed action), giving retries a cheap path back to a live broadcast. The reconciliation publication-pending sweep remains the durable backstop. Refs Sol finding 7. --- apps/labeler/src/console-mutation-api.ts | 8 +++++ .../labeler/test/console-mutation-api.test.ts | 30 +++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index 4e869e9f94..add22d6a31 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -291,6 +291,11 @@ async function runLabelMutation( const outcome = await guardMutation(request, spec, guardDeps); if (outcome.outcome === "replay") { await assertIssuancePersisted(deps.db, [outcome.actionId]); + // Redrive the subscription-DO notify: the original request's afterCommit may + // have failed, leaving the committed label `publication_pending = 1`. Replay + // is the cheap latency path back to a live broadcast; the reconciliation + // sweep is the durable backstop. + deps.defer(deps.afterCommit(outcome.actionId)); deferLabelNotify(deps, storedDescriptor(outcome.result)); return jsonData(outcome.result); } @@ -1040,6 +1045,9 @@ async function runEmergencyAction( if (outcome.outcome === "replay") { const stored = storedDescriptor(outcome.result); await assertIssuancePersisted(deps.db, [stored.actionId]); + // Redrive the subscription-DO notify in case the original afterCommit dropped + // it (see runLabelMutation's replay branch). + deps.defer(deps.afterCommit(stored.actionId)); if (action === "takedown") deferTakedownNotify(deps, stored); return jsonData(stored); } diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index d194a55d49..49ce76763b 100644 --- a/apps/labeler/test/console-mutation-api.test.ts +++ b/apps/labeler/test/console-mutation-api.test.ts @@ -595,6 +595,36 @@ describe("console mutation: replay and conflict", () => { ).toBe(1); }); + it("redrives the subscription-DO notify on replay (Finding 7)", async () => { + const uri = await seedReleaseSubject("replay-redrive"); + const key = nextKey(); + const request = () => + post("/admin/api/labels/issue", { + uri, + val: "security-yanked", + confirmation: "replay-redrive", + reason: "redrive on replay", + idempotencyKey: key, + }); + const notified: string[] = []; + const deps = mutationDeps({ + afterCommit: async (actionId) => { + notified.push(actionId); + }, + }); + + const first = await handleConsoleMutation(request(), deps); + expect(first.status).toBe(200); + const second = await handleConsoleMutation(request(), deps); + expect(second.status).toBe(200); + + // afterCommit (the live subscription-DO notify) fired on the proceed path AND + // again on the replay, keyed on the same committed action — so a client retry + // re-drives a broadcast the original request may have dropped. + expect(notified).toHaveLength(2); + expect(new Set(notified).size).toBe(1); + }); + it("returns 409 for the same key with a different fingerprint", async () => { const uri = await seedReleaseSubject("conflict"); const key = nextKey(); From 122a17e5e11b32aab130fc13b4f0b7bf894d495a Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 13:49:15 +0100 Subject: [PATCH 02/15] fix(labeler): re-sign retired-key labels on WebSocket subscription replay After a routine key rotation the DID document publishes only the new key, but retained rows keep their old-key signatures; queryLabels re-signs lazily on read while the subscribeLabels replay path sent rows unchanged, so a fresh aggregator rejected the first old-key frame and could not advance. The replay reader now reuses queryLabels' lazy re-sign helper (exported from query-labels), bringing stale-key rows onto the active key before framing and persisting the result so the work is not repeated per connection. Sequence and ordering are preserved; a signing pause throws rather than serving an unverifiable frame. Refs Sol finding 5. --- apps/labeler/src/query-labels.ts | 16 +- apps/labeler/src/subscribe-labels.ts | 22 ++- apps/labeler/test/subscribe-resign.test.ts | 172 +++++++++++++++++++++ 3 files changed, 205 insertions(+), 5 deletions(-) create mode 100644 apps/labeler/test/subscribe-resign.test.ts diff --git a/apps/labeler/src/query-labels.ts b/apps/labeler/src/query-labels.ts index ebc127a4d5..d490a5800e 100644 --- a/apps/labeler/src/query-labels.ts +++ b/apps/labeler/src/query-labels.ts @@ -14,7 +14,7 @@ const POSITIVE_INTEGER = /^[1-9]\d*$/; const DEFAULT_LIMIT = 50; const MAX_LIMIT = 250; -interface LabelRow { +export interface LabelRow { id: number; sequence: number; ver: number; @@ -100,7 +100,19 @@ export async function queryLabels( }); } -async function resignStaleLabels( +/** + * Lazily brings a page of retained label rows onto the active signing key: any + * row whose `signing_key_version` differs from the active key is re-signed with + * the current key and persisted (its prior signature archived in + * `label_signature_history`), mutating the passed rows in place. `sequence`, + * `cts`, and every label field except the signature are untouched, so ordering + * and identity are preserved. Shared by the public `queryLabels` reader and the + * WebSocket subscription replay so both serve verifiable frames after a routine + * key rotation. Throws when the current signing key is unavailable (paused mid + * rotation, or a configuration/state mismatch) — the caller must not serve the + * still-stale rows. + */ +export async function resignStaleLabels( db: D1Database, labels: LabelRow[], signingInput?: VersionedLabelSigner | (() => Promise), diff --git a/apps/labeler/src/subscribe-labels.ts b/apps/labeler/src/subscribe-labels.ts index 0b71af5c8c..5c821fb26d 100644 --- a/apps/labeler/src/subscribe-labels.ts +++ b/apps/labeler/src/subscribe-labels.ts @@ -1,7 +1,10 @@ import { encode, toBytes } from "@atcute/cbor"; import { DurableObject } from "cloudflare:workers"; +import { getLabelerIdentityConfig } from "./config.js"; +import { resignStaleLabels, type LabelRow } from "./query-labels.js"; import type { IssuedLabel } from "./service.js"; +import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js"; export const LABEL_SUBSCRIPTION_DO_NAME = "main"; @@ -208,15 +211,28 @@ export class LabelSubscriptionDO extends DurableObject { private async labelsAfter(cursor: number, through: number): Promise { const rows = await this.env.DB.prepare( - `SELECT sequence, ver, src, uri, cid, val, neg, cts, exp, sig + `SELECT id, sequence, ver, src, uri, cid, val, neg, cts, exp, sig, + signing_key_id, signing_key_version FROM issued_labels WHERE sequence > ? AND sequence <= ? ORDER BY sequence ASC LIMIT ?`, ) .bind(cursor, through, REPLAY_PAGE_SIZE) - .all(); - return (rows.results ?? []).map((row) => ({ + .all(); + // Bring any row signed with a retired key onto the active key before framing + // it, the same lazy re-sign the public query reader does — a fresh aggregator + // verifies replayed frames against the current published key, so an + // un-re-signed retained row would fail verification and stall the stream. The + // re-sign preserves sequence and ordering; a signing pause throws (mirroring + // queryLabels' 503) rather than serving a still-stale frame. + const resigned = await resignStaleLabels(this.env.DB, rows.results ?? [], async () => + createRuntimeSigner( + await getLabelerIdentityConfig(this.env), + getRuntimeSigningSecret(this.env), + ), + ); + return resigned.map((row) => ({ sequence: row.sequence, label: rowToLabel(row), })); diff --git a/apps/labeler/test/subscribe-resign.test.ts b/apps/labeler/test/subscribe-resign.test.ts new file mode 100644 index 0000000000..b4316a3230 --- /dev/null +++ b/apps/labeler/test/subscribe-resign.test.ts @@ -0,0 +1,172 @@ +import { decodeFirst, fromBytes } from "@atcute/cbor"; +import { + createLabelSigner, + verifyLabel, + type LabelDidDocument, + type SignedLabel, +} from "@emdash-cms/registry-moderation"; +import { applyD1Migrations, env, SELF } from "cloudflare:test"; +import { beforeAll, describe, expect, it } from "vitest"; + +import { initializeSigningState } from "../src/signing-rotation.js"; + +interface TestEnv { + DB: D1Database; + TEST_MIGRATIONS: Parameters[1]; +} + +const testEnv = env as unknown as TestEnv; +const LABELER_DID = "did:web:labels.emdashcms.com"; +const PUBLISHER_DID = "did:plc:publisher000000000000000000"; +const ACTIVE_MULTIKEY = "zDnaepsL7AXenJkVYdkh5KuKsSU7Ykh7kyXaLLU7auN9FWSiZ"; +const RETIRED_PRIVATE_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAI"; +const RETIRED_MULTIKEY = "zDnaer52RTwabaBeMkKYYwZmEFqPabLW78cRK62iovMUQhFif"; + +beforeAll(async () => { + await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); + // The worker env signs v1 with ACTIVE_MULTIKEY, so the active signing key here + // matches the deployment config the subscription DO re-signs with. + await initializeSigningState(testEnv.DB, { + issuerDid: LABELER_DID, + keyVersion: "v1", + publicKeyMultibase: ACTIVE_MULTIKEY, + }); +}); + +function retiredDocument(): LabelDidDocument { + return { + id: LABELER_DID, + verificationMethod: [ + { + id: "#atproto_label", + type: "Multikey", + controller: LABELER_DID, + publicKeyMultibase: RETIRED_MULTIKEY, + }, + ], + }; +} + +function activeDocument(): LabelDidDocument { + return { + id: LABELER_DID, + verificationMethod: [ + { + id: "#atproto_label", + type: "Multikey", + controller: LABELER_DID, + publicKeyMultibase: ACTIVE_MULTIKEY, + }, + ], + }; +} + +/** Persists a label signed with the retired key at a stale key version, exactly + * the shape a routine rotation leaves behind in retained history. */ +async function seedRetiredKeyLabel(uri: string): Promise { + const signer = await createLabelSigner({ + issuerDid: LABELER_DID, + privateKey: RETIRED_PRIVATE_KEY, + resolveDid: async () => retiredDocument(), + }); + const cts = new Date().toISOString(); + const unsigned = { ver: 1, uri, val: "security-yanked", cts } as const; + const returned = await signer.sign(unsigned); + const idempotencyKey = `resign-seed-${uri}`; + await testEnv.DB.prepare( + `INSERT INTO issuance_actions (actor, type, reason, idempotency_key, created_at) + VALUES (?, 'manual-label', 'retired-key seed', ?, ?)`, + ) + .bind(LABELER_DID, idempotencyKey, cts) + .run(); + await testEnv.DB.prepare( + `INSERT INTO issued_labels + (action_id, ver, src, uri, cid, val, neg, cts, exp, sig, signing_key_id, + signing_key_version, publication_pending) + SELECT id, 1, ?, ?, NULL, 'security-yanked', 0, ?, NULL, ?, ?, 'v0', 0 + FROM issuance_actions WHERE idempotency_key = ?`, + ) + .bind(LABELER_DID, uri, cts, returned.sig, `${LABELER_DID}#atproto_label`, idempotencyKey) + .run(); + const row = await testEnv.DB.prepare( + `SELECT sequence FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.idempotency_key = ?`, + ) + .bind(idempotencyKey) + .first<{ sequence: number }>(); + return row!.sequence; +} + +async function subscribe(cursor: number): Promise { + const response = await SELF.fetch( + `https://test/xrpc/com.atproto.label.subscribeLabels?cursor=${cursor}`, + { headers: { upgrade: "websocket" } }, + ); + expect(response.status).toBe(101); + if (!response.webSocket) throw new Error("subscription did not upgrade to a WebSocket"); + response.webSocket.accept(); + return response.webSocket; +} + +function decodeLabel(message: ArrayBuffer): { + seq: number; + label: Record; +} { + const [header, payload] = decodeFirst(new Uint8Array(message)) as [ + { op: number; t: string }, + Uint8Array, + ]; + expect(header).toEqual({ op: 1, t: "#labels" }); + const event = decodeFirst(payload)[0] as { seq: number; labels: Record[] }; + const label = event.labels[0]; + if (!label) throw new Error("labels event did not contain a label"); + return { seq: event.seq, label }; +} + +async function nextLabelWithSeq(ws: WebSocket, seq: number): Promise> { + return new Promise((resolve) => { + const listener = (event: MessageEvent) => { + const decoded = decodeLabel(event.data as ArrayBuffer); + if (decoded.seq !== seq) return; + ws.removeEventListener("message", listener); + resolve(decoded.label); + }; + ws.addEventListener("message", listener); + }); +} + +describe("subscription replay re-signs retired-key labels (Finding 5)", () => { + it("delivers a replayed frame that verifies under the active key and persists the re-sign", async () => { + const uri = `at://${PUBLISHER_DID}/com.emdashcms.experimental.package.release/resign-replay:1.0.0`; + const sequence = await seedRetiredKeyLabel(uri); + + const ws = await subscribe(sequence - 1); + const frame = await nextLabelWithSeq(ws, sequence); + ws.close(); + + const label: SignedLabel = { + ver: 1, + src: String(frame.src), + uri: String(frame.uri), + val: String(frame.val), + cts: String(frame.cts), + sig: fromBytes(frame.sig as { $bytes: string }), + }; + // The replayed frame no longer verifies under the retired key... + await expect( + verifyLabel({ label, resolveDid: async () => retiredDocument() }), + ).rejects.toThrow(); + // ...it verifies under the active (published) key a fresh aggregator would use. + await expect( + verifyLabel({ label, resolveDid: async () => activeDocument() }), + ).resolves.toMatchObject({ uri, val: "security-yanked" }); + + // The re-sign is persisted, so the work is not repeated per connection. + const persisted = await testEnv.DB.prepare( + `SELECT signing_key_version FROM issued_labels WHERE sequence = ?`, + ) + .bind(sequence) + .first<{ signing_key_version: string }>(); + expect(persisted?.signing_key_version).toBe("v1"); + }); +}); From 8e5a0bc5f1ec9b3a2bdd996427a730f8db0379ae Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 13:52:03 +0100 Subject: [PATCH 03/15] fix(labeler): publish automated assessment labels live, with a reconciliation backstop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Automated pending/outcome/block/negation labels committed without ever notifying the subscription DO, so a connected aggregator received nothing until it reconnected or an unrelated notification arrived. The orchestrator now issues finalization labels publication_pending=1 and broadcasts each to the same subscription-DO publisher the console path uses, off the commit. A new publication-pending sweep in the reconciliation cron re-drives any notify that was dropped (assessment or console path alike) — the durable guarantee that a stranded row reaches subscribers and stops blocking the next key rotation. Refs Sol findings 4 and 7. --- apps/labeler/src/assessment-orchestrator.ts | 50 +++++++- apps/labeler/src/assessment-workflow.ts | 4 + apps/labeler/src/index.ts | 20 ++- apps/labeler/src/reconciliation.ts | 67 ++++++++++ apps/labeler/src/service.ts | 2 +- apps/labeler/src/subscribe-labels.ts | 26 ++-- .../test/assessment-orchestrator.test.ts | 120 +++++++++++++++++- apps/labeler/test/reconciliation.test.ts | 91 ++++++++++++- 8 files changed, 363 insertions(+), 17 deletions(-) diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 72f5fff6fa..ea428631a7 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -31,9 +31,12 @@ import { resolvePolicyOutcome, type OutcomeLabel, type PolicyOutcome } from "./p import type { ModerationPolicy } from "./policy.js"; import { buildIssuanceStatements, + markPublicationAccepted, type AutomatedIssuanceAction, type AutomatedLabelProposal, + type IssuedLabel, } from "./service.js"; +import type { LabelPublisher } from "./subscribe-labels.js"; /** * A stage's finding is the canonical normalized contract (`findings.ts`). @@ -106,6 +109,13 @@ export interface AssessmentOrchestratorOptions { * AI stages accumulate (`assessment-stages.ts`); `undefined` (or an undefined * return) leaves the stored value unchanged. */ resolveCoverageJson?: () => string | undefined; + /** Broadcasts each finalized label to the subscription DO after the batch + * commits (the same publisher the console path uses via `createLabelPublisher`). + * When present, finalization labels are issued `publication_pending = 1` and + * this drives the live notify; the reconciliation `publication_pending` sweep is + * the durable backstop for a notify that fails here. Omitted in tests that don't + * exercise publication. */ + publisher?: LabelPublisher; } export class AssessmentOrchestrator { @@ -119,6 +129,7 @@ export class AssessmentOrchestrator { private readonly sleep: (ms: number) => Promise; private readonly retryDelayMs: number; private readonly resolveCoverageJson: (() => string | undefined) | undefined; + private readonly publisher: LabelPublisher | undefined; constructor(opts: AssessmentOrchestratorOptions) { this.db = opts.db; @@ -131,6 +142,7 @@ export class AssessmentOrchestrator { this.sleep = opts.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms))); this.retryDelayMs = opts.retryDelayMs ?? 0; this.resolveCoverageJson = opts.resolveCoverageJson; + this.publisher = opts.publisher; } async runAssessment(runId: string): Promise { @@ -270,7 +282,7 @@ export class AssessmentOrchestrator { }); const statements = [...finalization.statements]; - const postCommits: Array<() => Promise> = []; + const postCommits: Array<() => Promise> = []; const issue = async ( val: string, @@ -298,7 +310,11 @@ export class AssessmentOrchestrator { ...proposal, }, now, - false, + // Mark for publication only when a publisher is wired: the post-commit + // notify (below) drains it, the reconciliation sweep backstops a failed + // notify, and rotation waits for the drain. With no publisher (tests), + // the label commits already-published so nothing strands. + this.publisher !== undefined, // Gate every finalization label on the run reaching `toState`, so a // concurrent cancel/delete that no-ops the CAS also no-ops the labels. { requireAssessmentState: toState }, @@ -382,12 +398,40 @@ export class AssessmentOrchestrator { const raced = await getAssessment(this.db, assessment.id); throw new AssessmentFinalizationConflictError(assessment.id, toState, raced?.state ?? null); } - for (const postCommit of postCommits) await postCommit(); + const issued: IssuedLabel[] = []; + for (const postCommit of postCommits) issued.push(await postCommit()); + await this.publishLabels(issued); const finalised = await getAssessment(this.db, assessment.id); if (!finalised) throw new Error(`assessment ${assessment.id} disappeared after finalization`); return finalised; } + + /** + * Live broadcast of the finalized labels to the subscription DO, best-effort: + * the batch has already committed, so a notify failure must never fail the run. + * A dropped notify leaves the row `publication_pending = 1` for the + * reconciliation sweep to re-drive (which also unblocks a rotation waiting on + * the drain). Mirrors `service.ts` `issueLabel`: when the publisher manages + * publication state (the DO clears the flag on `/notify`) the caller does not. + */ + private async publishLabels(issued: readonly IssuedLabel[]): Promise { + const publisher = this.publisher; + if (!publisher) return; + for (const label of issued) { + try { + await publisher.publish(label); + if (!publisher.managesPublicationState) await markPublicationAccepted(this.db, label); + } catch (error) { + console.error("[assessment-orchestrator] label publication failed", { + assessmentId: + label.action.type === "automated-assessment" ? label.action.assessmentId : undefined, + sequence: label.sequence, + error: error instanceof Error ? error.message : String(error), + }); + } + } + } } /** diff --git a/apps/labeler/src/assessment-workflow.ts b/apps/labeler/src/assessment-workflow.ts index db0dbeb68a..f0d91fef1c 100644 --- a/apps/labeler/src/assessment-workflow.ts +++ b/apps/labeler/src/assessment-workflow.ts @@ -52,6 +52,7 @@ import { createNotifyDeps, notifyAssessmentOutcome } from "./notification-trigge import { MODERATION_POLICY, type ModerationPolicy } from "./policy.js"; import { createReleaseResolver, type ReleaseReader } from "./release-resolution.js"; import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js"; +import { createLabelPublisher } from "./subscribe-labels.js"; const RUN_STEP_CONFIG = { retries: { limit: 3, delay: "10 seconds" as const, backoff: "exponential" as const }, @@ -118,6 +119,9 @@ export async function executeAssessmentInstance( ai: env.AI, }), resolveCoverageJson: () => serializeCoverage(coverage), + // Same subscription-DO publisher the console path uses: finalized labels + // broadcast live post-commit, with the reconciliation sweep as the backstop. + publisher: createLabelPublisher(env), }); const finalized = await orchestrator.runAssessment(assessmentId); await notifyOutcome(env, finalized); diff --git a/apps/labeler/src/index.ts b/apps/labeler/src/index.ts index 36cffc7efd..8313e9273c 100644 --- a/apps/labeler/src/index.ts +++ b/apps/labeler/src/index.ts @@ -15,9 +15,9 @@ import { runNotificationSweep } from "./notification-sweep.js"; import { createNotifyDeps, type NotifyDeps } from "./notification-triggers.js"; import { runProlongedErrorEscalation } from "./prolonged-error.js"; import { queryLabels } from "./query-labels.js"; -import { reconcileAssessments } from "./reconciliation.js"; +import { reconcileAssessments, sweepPendingPublications } from "./reconciliation.js"; import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js"; -import { LABEL_SUBSCRIPTION_DO_NAME } from "./subscribe-labels.js"; +import { LABEL_SUBSCRIPTION_DO_NAME, notifyLabelSubscription } from "./subscribe-labels.js"; import { handleAssessmentXrpc } from "./xrpc-router.js"; import { xrpcError } from "./xrpc.js"; @@ -111,6 +111,22 @@ export default { }), ); + // Publication-pending sweep: re-drive the subscription-DO notify for labels + // whose live post-commit broadcast was dropped, so a stranded row can't block + // an aggregator or the next key rotation. Its own branch so a sweep failure + // never disturbs the passes around it. + ctx.waitUntil( + sweepPendingPublications({ + db: env.DB, + notify: (sequence) => notifyLabelSubscription(env, sequence), + now: new Date(), + }).catch((err: unknown) => { + console.error("[labeler] publication sweep failed", { + error: err instanceof Error ? err.message : String(err), + }); + }), + ); + // Publisher-notification retry sweep (plan W10.5): re-drive failed / // crash-stuck sends, abandon the exhausted, prune terminal rows. Isolated in // its own branch so a sweep failure never disturbs the passes above. diff --git a/apps/labeler/src/reconciliation.ts b/apps/labeler/src/reconciliation.ts index 7380ff5636..3663d3980a 100644 --- a/apps/labeler/src/reconciliation.ts +++ b/apps/labeler/src/reconciliation.ts @@ -19,6 +19,15 @@ import type { AssessmentState } from "./assessment-lifecycle.js"; const STUCK_STATES: readonly AssessmentState[] = ["verifying", "pending", "running"]; const DEFAULT_STALE_THRESHOLD_MS = 60 * 60 * 1000; +/** How long a committed label may sit `publication_pending` before the sweep + * re-drives its subscription-DO notify. Short relative to the stuck-run + * threshold: a stranded pending row blocks the next key rotation. */ +const DEFAULT_PUBLICATION_STALE_THRESHOLD_MS = 5 * 60 * 1000; + +/** Per-pass cap on re-driven publications. The 5-minute cron reconvenes to drain + * a larger backlog across passes rather than fan out an unbounded batch. */ +const PUBLICATION_SWEEP_LIMIT = 200; + export interface StuckAssessmentRun { id: string; uri: string; @@ -76,3 +85,61 @@ export async function reconcileAssessments( return { stuckRuns, subjectsWithoutRuns }; } + +export interface PendingPublicationSweepDeps { + db: D1Database; + /** Drives the subscription-DO notify for one committed sequence. The DO + * broadcasts it and clears `publication_pending`; a throw leaves the flag set + * for the next pass. */ + notify: (sequence: number) => Promise; + now: Date; + thresholdMs?: number; + limit?: number; +} + +export interface PendingPublicationSweepReport { + redriven: number; + failed: number; +} + +/** + * Durable backstop for the live post-commit notify (assessment finalization and + * the console mutation path both issue labels `publication_pending = 1` and + * broadcast off the response path). A transient notify failure otherwise strands + * the row pending forever — an aggregator never receives it and, worse, the next + * key rotation refuses to activate while a row signed with the outgoing key stays + * pending. This re-drives the DO notify for pending rows older than the + * threshold; the DO clears the flag on success. + */ +export async function sweepPendingPublications( + deps: PendingPublicationSweepDeps, +): Promise { + const thresholdMs = deps.thresholdMs ?? DEFAULT_PUBLICATION_STALE_THRESHOLD_MS; + const limit = deps.limit ?? PUBLICATION_SWEEP_LIMIT; + const staleBefore = new Date(deps.now.getTime() - thresholdMs).toISOString(); + + const rows = await deps.db + .prepare( + `SELECT sequence FROM issued_labels + WHERE publication_pending = 1 AND sequence IS NOT NULL AND cts <= ? + ORDER BY sequence ASC LIMIT ?`, + ) + .bind(staleBefore, limit) + .all<{ sequence: number }>(); + + let redriven = 0; + let failed = 0; + for (const row of rows.results ?? []) { + try { + await deps.notify(row.sequence); + redriven++; + } catch (error) { + failed++; + console.error("[labeler] reconciliation: publication redrive failed", { + sequence: row.sequence, + error: error instanceof Error ? error.message : String(error), + }); + } + } + return { redriven, failed }; +} diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index 1256fcd6f0..6dd5e39d3c 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -599,7 +599,7 @@ export async function issueAutomatedAssessmentLabel( return issueLabel(db, config, signer, action, proposal, now, publisher); } -async function markPublicationAccepted(db: D1Database, issued: IssuedLabel): Promise { +export async function markPublicationAccepted(db: D1Database, issued: IssuedLabel): Promise { await db .prepare( `UPDATE issued_labels SET publication_pending = 0 diff --git a/apps/labeler/src/subscribe-labels.ts b/apps/labeler/src/subscribe-labels.ts index 5c821fb26d..9e71aecf29 100644 --- a/apps/labeler/src/subscribe-labels.ts +++ b/apps/labeler/src/subscribe-labels.ts @@ -46,18 +46,26 @@ export interface LabelPublisher { publish(issued: IssuedLabel): Promise; } -export function createLabelPublisher(env: Env): LabelPublisher { +/** + * Notifies the subscription DO of a committed label sequence. The DO broadcasts + * it and clears the row's `publication_pending` flag. Shared by the live + * publisher and the reconciliation `publication_pending` sweep so both drive the + * exact same DO endpoint. + */ +export async function notifyLabelSubscription(env: Env, sequence: number): Promise { const subscription = env.LABEL_SUBSCRIPTION.getByName(LABEL_SUBSCRIPTION_DO_NAME); + const response = await subscription.fetch("https://labeler.internal/notify", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ sequence }), + }); + if (!response.ok) throw new Error(`label notification failed with ${response.status}`); +} + +export function createLabelPublisher(env: Env): LabelPublisher { return { managesPublicationState: true, - async publish(issued) { - const response = await subscription.fetch("https://labeler.internal/notify", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ sequence: issued.sequence }), - }); - if (!response.ok) throw new Error(`label notification failed with ${response.status}`); - }, + publish: (issued) => notifyLabelSubscription(env, issued.sequence), }; } diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index ae9d1b7245..406c29af5b 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -33,8 +33,9 @@ import { import { FindingValidationError, HISTORY_FINDING_CATEGORIES } from "../src/findings.js"; import { analyzeHistory } from "../src/history-context.js"; import { MODERATION_POLICY } from "../src/policy.js"; -import { issueManualLabel } from "../src/service.js"; +import { issueManualLabel, type IssuedLabel } from "../src/service.js"; import { initializeSigningState } from "../src/signing-rotation.js"; +import type { LabelPublisher } from "../src/subscribe-labels.js"; import { canonicalBundle, checksumOf, file } from "./bundle-fixture.js"; interface TestEnv { @@ -49,6 +50,24 @@ const PRIVATE_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAE"; const MULTIKEY = "zDnaepsL7AXenJkVYdkh5KuKsSU7Ykh7kyXaLLU7auN9FWSiZ"; const config = { labelerDid: LABELER_DID, signingKeyVersion: "v1" }; +/** Records every label handed to `publish`, mimicking the console/DO publisher. */ +function recordingPublisher(managesPublicationState: boolean): { + publisher: LabelPublisher; + published: IssuedLabel[]; +} { + const published: IssuedLabel[] = []; + return { + published, + publisher: { + managesPublicationState, + async publish(issued) { + published.push(issued); + }, + }, + }; +} + + beforeAll(async () => { await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); await initializeSigningState(testEnv.DB, { @@ -1119,3 +1138,102 @@ describe("AssessmentOrchestrator: real acquire stage (W7.2)", () => { expect(error?.neg).toBe(0); }); }); + +describe("AssessmentOrchestrator: live publication (Finding 4)", () => { + it("issues finalization labels publication_pending and broadcasts each to the publisher", async () => { + const run = await pendingRun({ name: "publish-live", cidValue: await cid("publish-live") }); + const { publisher, published } = recordingPublisher(true); + const orchestrator = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + publisher, + }); + + const result = await orchestrator.runAssessment(run.id); + expect(result.state).toBe("passed"); + + // Every label this run committed was broadcast: the assessment-pending + // negation and the assessment-passed positive. + const publishedVals = published.map((p) => p.label.val).toSorted(); + expect(publishedVals).toEqual(["assessment-passed", "assessment-pending"]); + + // A publisher that manages publication state (the real DO clears the flag on + // /notify) leaves the rows pending here — the fake never cleared them. + const pending = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels WHERE uri = ? AND cid = ? AND publication_pending = 1`, + ) + .bind(run.uri, run.cid) + .first<{ n: number }>(); + expect(pending?.n).toBe(2); + }); + + it("clears publication_pending via markPublicationAccepted for a non-managing publisher", async () => { + const run = await pendingRun({ name: "publish-accept", cidValue: await cid("publish-accept") }); + const { publisher, published } = recordingPublisher(false); + const orchestrator = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + publisher, + }); + + await orchestrator.runAssessment(run.id); + expect(published.length).toBe(2); + + const pending = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels WHERE uri = ? AND cid = ? AND publication_pending = 1`, + ) + .bind(run.uri, run.cid) + .first<{ n: number }>(); + expect(pending?.n).toBe(0); + }); + + it("finalizes and leaves rows publication_pending when the notify fails — the sweep backstops", async () => { + const run = await pendingRun({ name: "publish-fail", cidValue: await cid("publish-fail") }); + const failing: LabelPublisher = { + managesPublicationState: true, + publish: () => Promise.reject(new Error("subscription DO unreachable")), + }; + const orchestrator = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + publisher: failing, + }); + + // A dropped notify never fails the run: the batch already committed. + const result = await orchestrator.runAssessment(run.id); + expect(result.state).toBe("passed"); + + const pending = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels WHERE uri = ? AND cid = ? AND publication_pending = 1`, + ) + .bind(run.uri, run.cid) + .first<{ n: number }>(); + expect(pending?.n).toBe(2); + }); + + it("issues finalization labels already-published (publication_pending = 0) when no publisher is wired", async () => { + const run = await pendingRun({ name: "publish-none", cidValue: await cid("publish-none") }); + const orchestrator = await buildOrchestrator(); + + await orchestrator.runAssessment(run.id); + + const pending = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels WHERE uri = ? AND cid = ? AND publication_pending = 1`, + ) + .bind(run.uri, run.cid) + .first<{ n: number }>(); + expect(pending?.n).toBe(0); + }); +}); diff --git a/apps/labeler/test/reconciliation.test.ts b/apps/labeler/test/reconciliation.test.ts index 37bef9b4e1..12ef2e8d31 100644 --- a/apps/labeler/test/reconciliation.test.ts +++ b/apps/labeler/test/reconciliation.test.ts @@ -7,7 +7,7 @@ import { createSubject, transitionAssessmentState, } from "../src/assessment-store.js"; -import { reconcileAssessments } from "../src/reconciliation.js"; +import { reconcileAssessments, sweepPendingPublications } from "../src/reconciliation.js"; interface TestEnv { DB: D1Database; @@ -138,3 +138,92 @@ describe("reconcileAssessments", () => { expect(withTightThreshold.stuckRuns.some((run) => run.id === id)).toBe(true); }); }); + +let seedCounter = 0; + +/** Inserts a signed-label row directly with a chosen `cts` and pending flag, + * returning its trigger-assigned sequence. */ +async function seedPendingLabel(opts: { + cts: string; + publicationPending: boolean; +}): Promise { + seedCounter++; + const idempotencyKey = `sweep-seed-${seedCounter}`; + await testEnv.DB.prepare( + `INSERT INTO issuance_actions (actor, type, reason, idempotency_key, created_at) + VALUES (?, 'manual-label', 'sweep seed', ?, ?)`, + ) + .bind("did:example:seed", idempotencyKey, opts.cts) + .run(); + await testEnv.DB.prepare( + `INSERT INTO issued_labels + (action_id, ver, src, uri, cid, val, neg, cts, exp, sig, signing_key_id, + signing_key_version, publication_pending) + SELECT id, 1, 'did:example:seed', ?, NULL, 'security-yanked', 0, ?, NULL, ?, + 'did:example:seed#atproto_label', 'v1', ? + FROM issuance_actions WHERE idempotency_key = ?`, + ) + .bind( + `at://did:example:seed/com.emdashcms.experimental.package.release/sweep-${seedCounter}:1.0.0`, + opts.cts, + new Uint8Array([1, 2, 3]), + opts.publicationPending ? 1 : 0, + idempotencyKey, + ) + .run(); + const row = await testEnv.DB.prepare( + `SELECT sequence FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.idempotency_key = ?`, + ) + .bind(idempotencyKey) + .first<{ sequence: number }>(); + return row!.sequence; +} + +describe("sweepPendingPublications", () => { + const now = new Date("2026-07-18T12:00:00.000Z"); + const oldCts = new Date(now.getTime() - 30 * 60 * 1000).toISOString(); + const recentCts = new Date(now.getTime() - 60 * 1000).toISOString(); + + it("re-drives only pending rows older than the threshold", async () => { + const stale = await seedPendingLabel({ cts: oldCts, publicationPending: true }); + const recent = await seedPendingLabel({ cts: recentCts, publicationPending: true }); + const settled = await seedPendingLabel({ cts: oldCts, publicationPending: false }); + + const notified: number[] = []; + const report = await sweepPendingPublications({ + db: testEnv.DB, + notify: async (sequence) => { + notified.push(sequence); + }, + now, + }); + + expect(notified).toContain(stale); + expect(notified).not.toContain(recent); + expect(notified).not.toContain(settled); + expect(report.redriven).toBeGreaterThanOrEqual(1); + expect(report.failed).toBe(0); + }); + + it("counts a failed notify without throwing, leaving the row for a later pass", async () => { + const stuck = await seedPendingLabel({ cts: oldCts, publicationPending: true }); + + const report = await sweepPendingPublications({ + db: testEnv.DB, + notify: (sequence) => + sequence === stuck ? Promise.reject(new Error("DO unreachable")) : Promise.resolve(), + now, + }); + + expect(report.failed).toBeGreaterThanOrEqual(1); + // The row is untouched (still pending) — the sweep never clears the flag + // itself; only a successful DO notify does. + const stillPending = await testEnv.DB.prepare( + `SELECT publication_pending FROM issued_labels WHERE sequence = ?`, + ) + .bind(stuck) + .first<{ publication_pending: number }>(); + expect(stillPending?.publication_pending).toBe(1); + }); +}); From e3f7bc08951a6bc60e9c6d525a343450d0ea62b8 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 13:52:31 +0100 Subject: [PATCH 04/15] fix(labeler): guard the finalization CAS on signing state so a rotation pause can't strand a run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If signing paused between finalization prep and the batch commit, the guarded label INSERTs no-oped while the unguarded assessment CAS still committed terminal state — leaving the run terminal with its pending-negation, outcome, and block labels missing, and the Workflow retry seeing terminal and never repairing them. The CAS now shares the same signing-state predicate as the issuance statements, so the whole db.batch is all-or-nothing: a mid-batch pause no-ops the CAS too, the run stays running, and the retry re-runs finalization after signing resumes. Refs Sol finding 6. --- apps/labeler/src/assessment-orchestrator.ts | 11 +++ apps/labeler/src/assessment-store.ts | 49 +++++++--- .../test/assessment-orchestrator.test.ts | 92 ++++++++++++++++++- 3 files changed, 140 insertions(+), 12 deletions(-) diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index ea428631a7..bea0bab109 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -36,6 +36,7 @@ import { type AutomatedLabelProposal, type IssuedLabel, } from "./service.js"; +import { getSigningStatusIfInitialized } from "./signing-rotation.js"; import type { LabelPublisher } from "./subscribe-labels.js"; /** @@ -270,6 +271,12 @@ export class AssessmentOrchestrator { const positiveLabels: readonly OutcomeLabel[] = outcome?.labels ?? []; const coverageJson = this.resolveCoverageJson?.(); + // Read the signing status once so the CAS transition carries the same + // signing-state predicate as every label issuance below: a signing pause + // landing between prep and the batch then no-ops the CAS too, keeping the run + // `running` for the Workflow retry rather than committing terminal state with + // its labels suppressed. + const signingStatus = await getSigningStatusIfInitialized(this.db); const finalization = buildFinalizationStatements(this.db, { assessmentId: assessment.id, fromState: "running", @@ -279,6 +286,10 @@ export class AssessmentOrchestrator { cid: assessment.cid, now, ...(coverageJson !== undefined ? { coverageJson } : {}), + signingGuard: { + isPrebootstrap: signingStatus === null, + activeKeyVersion: this.config.signingKeyVersion, + }, }); const statements = [...finalization.statements]; diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index e821b32972..7c3c1e5dfd 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/apps/labeler/src/assessment-store.ts @@ -474,6 +474,16 @@ export interface FinalizationInput { publicSummary?: string; coverageJson?: string; supersedesAssessmentId?: string; + /** + * Gate the CAS transition on the same signing-state predicate the finalization + * label issuances use (`buildIssuanceStatements`). A signing pause landing + * between statement preparation and the batch commit otherwise no-ops every + * guarded label INSERT while the unguarded CAS still commits terminal state, + * stranding the run terminal with its labels missing. Sharing the predicate + * makes the batch all-or-nothing: a mid-batch pause no-ops the CAS too, the run + * stays `running`, and the Workflow retry re-runs finalization after resume. + */ + signingGuard?: { isPrebootstrap: boolean; activeKeyVersion: string }; } export interface FinalizationStatements { @@ -502,6 +512,32 @@ export function buildFinalizationStatements( `buildFinalizationStatements is for decision outcomes; use transitionAssessmentState for ${input.toState}`, ); const now = (input.now ?? new Date()).toISOString(); + // Mirror of `buildIssuanceStatements`' signing guard so the CAS shares the + // batch's all-or-nothing behaviour under a mid-batch signing pause. + const signingGuardSql = + input.signingGuard === undefined + ? "" + : `\n\t\t\t\t AND ( + (? = 1 AND NOT EXISTS (SELECT 1 FROM signing_state)) + OR (? = 0 AND EXISTS ( + SELECT 1 FROM signing_state + WHERE id = 1 AND phase = 'active' AND active_key_version = ? + )) + )`; + const casBinds: unknown[] = [ + input.toState, + now, + Date.parse(now), + input.publicSummary ?? null, + input.coverageJson ?? null, + input.supersedesAssessmentId ?? null, + input.assessmentId, + input.fromState, + ]; + if (input.signingGuard !== undefined) { + const prebootstrap = input.signingGuard.isPrebootstrap ? 1 : 0; + casBinds.push(prebootstrap, prebootstrap, input.signingGuard.activeKeyVersion); + } const statements: D1PreparedStatement[] = [ db .prepare( @@ -510,18 +546,9 @@ export function buildFinalizationStatements( public_summary = COALESCE(?, public_summary), coverage_json = COALESCE(?, coverage_json), supersedes_assessment_id = COALESCE(?, supersedes_assessment_id) - WHERE id = ? AND state = ?`, + WHERE id = ? AND state = ?${signingGuardSql}`, ) - .bind( - input.toState, - now, - Date.parse(now), - input.publicSummary ?? null, - input.coverageJson ?? null, - input.supersedesAssessmentId ?? null, - input.assessmentId, - input.fromState, - ), + .bind(...casBinds), ]; let pointerUpdateIndex: number | null = null; if (CURRENT_POINTER_STATES.has(input.toState)) { diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index 406c29af5b..2fd378a372 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -34,7 +34,11 @@ import { FindingValidationError, HISTORY_FINDING_CATEGORIES } from "../src/findi import { analyzeHistory } from "../src/history-context.js"; import { MODERATION_POLICY } from "../src/policy.js"; import { issueManualLabel, type IssuedLabel } from "../src/service.js"; -import { initializeSigningState } from "../src/signing-rotation.js"; +import { + abortRoutineKeyRotation, + beginRoutineKeyRotation, + initializeSigningState, +} from "../src/signing-rotation.js"; import type { LabelPublisher } from "../src/subscribe-labels.js"; import { canonicalBundle, checksumOf, file } from "./bundle-fixture.js"; @@ -48,6 +52,7 @@ const LABELER_DID = "did:web:labels.emdashcms.com"; const PUBLISHER_DID = "did:plc:publisher000000000000000000"; const PRIVATE_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAE"; const MULTIKEY = "zDnaepsL7AXenJkVYdkh5KuKsSU7Ykh7kyXaLLU7auN9FWSiZ"; +const ROTATED_MULTIKEY = "zDnaer52RTwabaBeMkKYYwZmEFqPabLW78cRK62iovMUQhFif"; const config = { labelerDid: LABELER_DID, signingKeyVersion: "v1" }; /** Records every label handed to `publish`, mimicking the console/DO publisher. */ @@ -67,6 +72,27 @@ function recordingPublisher(managesPublicationState: boolean): { }; } +/** A D1 wrapper that runs `onFirstBatch` immediately before the first + * `db.batch` call — the seam between finalization prep and commit. Everything + * else delegates to the real database. */ +function pauseBeforeFirstBatch(db: D1Database, onFirstBatch: () => Promise): D1Database { + let triggered = false; + return new Proxy(db, { + get(target, prop, receiver) { + if (prop === "batch") { + return async (statements: D1PreparedStatement[]) => { + if (!triggered) { + triggered = true; + await onFirstBatch(); + } + return target.batch(statements); + }; + } + const value = Reflect.get(target, prop, receiver) as unknown; + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} beforeAll(async () => { await applyD1Migrations(testEnv.DB, testEnv.TEST_MIGRATIONS); @@ -1237,3 +1263,67 @@ describe("AssessmentOrchestrator: live publication (Finding 4)", () => { expect(pending?.n).toBe(0); }); }); + +describe("AssessmentOrchestrator: signing pause between prep and commit (Finding 6)", () => { + it("leaves the run running and issues no labels when signing pauses before the batch commits", async () => { + const run = await pendingRun({ name: "pause-race", cidValue: await cid("pause-race") }); + const rotationId = "finding6-pause"; + // Pause signing in the seam between finalization prep (statements built while + // active) and the batch commit. + const db = pauseBeforeFirstBatch(testEnv.DB, async () => { + await beginRoutineKeyRotation(testEnv.DB, { + rotationId, + expectedActiveKeyVersion: "v1", + nextKeyVersion: "v2-finding6", + nextPublicKeyMultibase: ROTATED_MULTIKEY, + }); + }); + const orchestrator = new AssessmentOrchestrator({ + db, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + }); + + // The CAS shares the issuance signing guard, so the whole batch no-ops: the + // finalization conflict is raised rather than a suppressed-label signing error. + await expect(orchestrator.runAssessment(run.id)).rejects.toBeInstanceOf( + AssessmentFinalizationConflictError, + ); + + // The run is still running (not stranded terminal) and NO label leaked. + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("running"); + const labels = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ?`, + ) + .bind(run.id) + .first<{ n: number }>(); + expect(labels?.n).toBe(0); + + // Resume signing and re-run (as the Workflow retry does): finalization now + // completes and issues the labels the paused attempt withheld. + await abortRoutineKeyRotation(testEnv.DB, { + rotationId, + expectedPendingKeyVersion: "v2-finding6", + }); + const resumed = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: stubStages, + sleep: () => Promise.resolve(), + }); + const finalized = await resumed.runAssessment(run.id); + expect(finalized.state).toBe("passed"); + const pendingNeg = await testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND cid = ? AND val = 'assessment-pending'`, + ) + .bind(run.uri, run.cid) + .first<{ neg: number }>(); + expect(pendingNeg?.neg).toBe(1); + }); +}); From 8b548bbf0dfa64de81f3b5a24e96e41d5f6068c2 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:25:44 +0100 Subject: [PATCH 05/15] perf(labeler): index issued_labels for the publication-pending sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconciliation sweep filtered issued_labels by publication_pending ordered by sequence with no covering index, full-scanning a monotonically growing table every cron tick — a D1 query-timeout there would strand pending rows and block the very rotation drain the sweep exists to unblock. Adds migration 0011 with a partial covering index on (sequence, cts) WHERE publication_pending = 1, and a query-plan guard so the sweep can't silently regress to a full scan. Follow-up to emdashbot review on #2115. --- .../0011_publication_pending_index.sql | 11 +++++++++++ apps/labeler/test/reconciliation.test.ts | 17 +++++++++++++++++ 2 files changed, 28 insertions(+) create mode 100644 apps/labeler/migrations/0011_publication_pending_index.sql diff --git a/apps/labeler/migrations/0011_publication_pending_index.sql b/apps/labeler/migrations/0011_publication_pending_index.sql new file mode 100644 index 0000000000..03d97c880d --- /dev/null +++ b/apps/labeler/migrations/0011_publication_pending_index.sql @@ -0,0 +1,11 @@ +-- Partial covering index for the reconciliation publication-pending sweep +-- (`sweepPendingPublications`): it scans `publication_pending = 1` rows ordered +-- by `sequence` and filtered by `cts`. Without this the sweep full-scans the +-- monotonically growing `issued_labels` table every cron tick, risking a D1 +-- query-timeout that would strand pending rows and block the rotation drain the +-- sweep exists to unblock. The partial predicate keeps the index to the tiny +-- set of un-broadcast rows; `(sequence, cts)` covers the SELECT, the cts filter, +-- and the sequence ordering. +CREATE INDEX idx_issued_labels_publication_pending +ON issued_labels(sequence, cts) +WHERE publication_pending = 1; diff --git a/apps/labeler/test/reconciliation.test.ts b/apps/labeler/test/reconciliation.test.ts index 12ef2e8d31..ab6ba52599 100644 --- a/apps/labeler/test/reconciliation.test.ts +++ b/apps/labeler/test/reconciliation.test.ts @@ -226,4 +226,21 @@ describe("sweepPendingPublications", () => { .first<{ publication_pending: number }>(); expect(stillPending?.publication_pending).toBe(1); }); + + it("plans the sweep query through the partial index, not a full table scan", async () => { + // Guards against the sweep degrading to a full scan of the monotonically + // growing issued_labels table (a D1 query-timeout would strand pending rows + // and block the rotation drain). The query and the 0011 partial index must + // stay in sync. + const plan = await testEnv.DB.prepare( + `EXPLAIN QUERY PLAN SELECT sequence FROM issued_labels + WHERE publication_pending = 1 AND sequence IS NOT NULL AND cts <= ? + ORDER BY sequence ASC LIMIT ?`, + ) + .bind(new Date().toISOString(), 200) + .all<{ detail: string }>(); + const detail = (plan.results ?? []).map((row) => row.detail).join(" | "); + expect(detail).toContain("idx_issued_labels_publication_pending"); + expect(detail).not.toContain("SCAN issued_labels"); + }); }); From 90719558a6a19538cef458a1262a298a3d778e2b Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 14:25:56 +0100 Subject: [PATCH 06/15] docs(labeler): drop the stale mid-batch signing-flip gap from finalize The F6 fix guards the finalization CAS on the same signing-state predicate as the label issuances, so the race-analysis comment's first bullet (claiming the CAS is unguarded and a flip could commit terminal state with labels suppressed) is no longer true. Replaces it with the now-closed description and keeps the two genuinely-remaining narrower gaps. Follow-up to emdashbot review on #2115. --- apps/labeler/src/assessment-orchestrator.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index bea0bab109..77255244b8 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -374,11 +374,12 @@ export class AssessmentOrchestrator { // of `running` in this gap therefore no-ops the CAS AND every label — nothing // leaks — and the lost race raises AssessmentFinalizationConflictError. // + // A signing-state flip mid-batch is closed the same way: the CAS carries the + // same signing-state guard as every label insert (buildFinalizationStatements' + // signingGuard), so a flip no-ops the CAS too and the batch commits nothing — + // the run stays `running` for the Workflow retry. + // // Narrower gaps remain, tracked with the real-stage wiring: - // - a signing-state flip mid-batch: the label inserts are guarded on active - // signing state and no-op if it flips, but the CAS is not, so a flip could - // commit the terminal state with its labels suppressed (the CAS still - // changed a row, so the postCommit below surfaces it as a signing error); // - a CID supersession landing in this gap does not move the run out of // `running`, so the CAS succeeds and this run finalizes labels for its own // CID (the pointer upsert is guarded on created-at ordering); From 38eb4974bc2fd71e6c62808f1dad23aa0af83008 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 16:12:34 +0100 Subject: [PATCH 07/15] fix(labeler): publish discovery-consumer labels live, with the sweep backstop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #4 fix gave the orchestrator path live publication but missed the discovery-consumer path: its initial assessment-pending issuance and its deletion negations committed publication_pending=0 with no subscription-DO broadcast, so connected subscribers never received them and the new reconciliation sweep (pending=1 only) could not recover them. Both issuances now take the same subscription-DO publisher the orchestrator uses — publication_pending=1 on commit plus a best-effort post-commit notify (a dropped broadcast never fails the discovery message; the sweep re-drives it). The rotation-drain contract is unchanged: these pending rows drain exactly like the orchestrator's. Follow-up to Sol review on #2115. --- apps/labeler/src/discovery-consumer.ts | 34 ++++++ apps/labeler/test/discovery-consumer.test.ts | 103 +++++++++++++++++++ 2 files changed, 137 insertions(+) diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index ee42c02e59..8d5e717b30 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -72,6 +72,7 @@ import { } from "./record-verification.js"; import { issueAutomatedAssessmentLabel, LabelIssuanceUnavailableError } from "./service.js"; import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js"; +import { createLabelPublisher, type LabelPublisher } from "./subscribe-labels.js"; /** * Stub identifiers for the model/prompt/scanner-set components of the run @@ -93,6 +94,13 @@ export interface DiscoveryConsumerDeps { * runKey, so a redelivered event dedups onto the same instance * (assessment-dispatch.ts). */ assessmentWorkflow: AssessmentWorkflowBinding; + /** Live broadcast for the pending-label and deletion-negation issuances. When + * present they commit `publication_pending = 1` and notify the subscription DO + * post-commit (the same publisher the orchestrator path uses); the + * reconciliation sweep is the durable backstop for a dropped notify. Omitted in + * tests that don't exercise publication (labels then commit already-published, + * matching the orchestrator's no-publisher behaviour). */ + publisher?: LabelPublisher; fetch?: typeof fetch; now?: () => Date; /** @@ -399,6 +407,7 @@ async function verifyAndCreateRun( }, { uri, cid: job.cid, val: "assessment-pending" }, now, + deps.publisher, ); // Hand the run to its Workflow instance. The instance id is the run's runKey, @@ -500,6 +509,7 @@ async function negatePendingForDeletedRuns( }, { uri: run.uri, cid: run.cid, val: "assessment-pending", neg: true }, now, + deps.publisher, ); } } @@ -543,6 +553,7 @@ async function createProductionDiscoveryDeps(env: Env): Promise { expect(negated?.neg).toBe(1); }); }); + +describe("processDiscoveryMessage: live publication (Sol follow-up)", () => { + function recordingPublisher(): { publisher: LabelPublisher; published: number[] } { + const published: number[] = []; + return { + published, + publisher: { + managesPublicationState: true, + async publish(issued) { + published.push(issued.sequence); + }, + }, + }; + } + + async function pendingLabelRow( + assessmentId: string, + neg: number, + ): Promise<{ sequence: number; publication_pending: number }> { + const row = await testEnv.DB.prepare( + `SELECT l.sequence, l.publication_pending FROM issued_labels l + JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ? AND l.val = 'assessment-pending' AND l.neg = ?`, + ) + .bind(assessmentId, neg) + .first<{ sequence: number; publication_pending: number }>(); + if (!row) throw new Error("assessment-pending label not found"); + return row; + } + + it("issues the discovery pending label publication_pending=1 and broadcasts it live", async () => { + const job = await jobFor({ rkey: rkey() }); + const { publisher, published } = recordingPublisher(); + const deps = { ...(await buildDeps()), publisher, verify: verifiedFor(job) }; + + await processDiscoveryMessage(job, new FakeMessage(), deps); + + const assessment = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); + const row = await pendingLabelRow(assessment!.id, 0); + // Committed publication_pending=1 (pre-fix: 0, so the sweep could never + // recover it) and broadcast to the subscription DO. + expect(row.publication_pending).toBe(1); + expect(published).toContain(row.sequence); + }); + + it("survives a dropped broadcast and leaves the pending label for the sweep", async () => { + const job = await jobFor({ rkey: rkey() }); + const failing: LabelPublisher = { + managesPublicationState: true, + publish: () => Promise.reject(new Error("subscription DO unreachable")), + }; + const deps = { + ...(await buildDeps()), + publisher: bestEffortPublisher(failing), + verify: verifiedFor(job), + }; + const msg = new FakeMessage(); + + await processDiscoveryMessage(job, msg, deps); + + // Best-effort: a failed live broadcast never fails/retries the message. + expect(msg.acked).toBe(1); + expect(msg.retried).toBe(0); + + const assessment = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); + const row = await pendingLabelRow(assessment!.id, 0); + expect(row.publication_pending).toBe(1); + + // The reconciliation sweep re-drives the stranded row. + const swept: number[] = []; + await sweepPendingPublications({ + db: testEnv.DB, + notify: async (sequence) => { + swept.push(sequence); + }, + now: new Date(Date.now() + 60_000), + thresholdMs: 0, + }); + expect(swept).toContain(row.sequence); + }); + + it("issues the deletion negation publication_pending=1 and broadcasts it", async () => { + const job = await jobFor({ rkey: rkey() }); + const { publisher, published } = recordingPublisher(); + const deps = { ...(await buildDeps()), publisher }; + await processDiscoveryMessage(job, new FakeMessage(), { ...deps, verify: verifiedFor(job) }); + const assessment = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job)); + published.length = 0; + + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + await processDiscoveryMessage(deleteJob, new FakeMessage(), { + ...deps, + confirmDeleted: () => Promise.resolve(true), + }); + + const row = await pendingLabelRow(assessment!.id, 1); + expect(row.publication_pending).toBe(1); + expect(published).toContain(row.sequence); + }); +}); From fb3bb8b2b5a5e86d9ff25352a56a2514bdecad83 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 17:55:50 +0100 Subject: [PATCH 08/15] fix(labeler): negate a deleted run's pending label before cancelling it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delete path cancelled pending/running runs to terminal 'cancelled' before issuing their assessment-pending negations. If issuance was paused (signing rotation) the negation threw and the message retried, but on redelivery listNonTerminalAssessmentsForUri excluded the already-cancelled runs — so there was nothing left to negate, the message acked, and the active assessment-pending label survived a deleted release forever. Each run's negation is now issued before its terminal cancellation, so a paused/failed negation leaves the run non-terminal and re-discoverable on redelivery; a run is retired only after its negation commits, and the message cannot ack while any pending/running run is un-negated. No active assessment-pending survives an acked delete. Follow-up to Sol review on #2115. --- apps/labeler/src/discovery-consumer.ts | 91 ++++++++++++-------- apps/labeler/test/discovery-consumer.test.ts | 57 ++++++++++++ 2 files changed, 112 insertions(+), 36 deletions(-) diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 8d5e717b30..500191a3f7 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -235,8 +235,7 @@ export async function processDiscoveryMessage( controller.ack(); return; } - const cancelled = await applyDiscoveryDelete(deps.db, uri, now()); - await negatePendingForDeletedRuns(deps, cancelled, now()); + await applyDiscoveryDelete(deps, uri, now()); controller.ack(); } catch (err) { await classifyDiscoveryError(err, job, deps, controller, now()); @@ -455,13 +454,26 @@ async function transitionOrObserve( } } -/** Tombstones the subject, cancels non-terminal runs, and returns the runs - * that had already reached `pending` (so they carry an active - * `assessment-pending` label the caller must negate). */ -async function applyDiscoveryDelete(db: D1Database, uri: string, now: Date): Promise { - await deleteSubjectsByUri(db, { uri, now }); - const runs = await listNonTerminalAssessmentsForUri(db, uri); - const hadPending: Assessment[] = []; +/** + * Tombstones the subject and retires its non-terminal runs. For each run that + * reached `pending`/`running` — and therefore carries an active + * `assessment-pending` label — the negation is issued BEFORE the terminal + * cancellation, so a failed or paused negation (signing mid-rotation) leaves the + * run non-terminal and re-discoverable by `listNonTerminalAssessmentsForUri` on + * redelivery. Cancelling first would drop the run from that set, stranding the + * pending label live forever once the message acks. The invariant: a run is + * cancelled only after its pending negation has committed, and the message + * cannot ack while any pending/running run is still un-negated (a throw + * propagates to `classifyDiscoveryError` → retry), so no active + * `assessment-pending` survives an acked delete. + */ +async function applyDiscoveryDelete( + deps: DiscoveryConsumerDeps, + uri: string, + now: Date, +): Promise { + await deleteSubjectsByUri(deps.db, { uri, now }); + const runs = await listNonTerminalAssessmentsForUri(deps.db, uri); for (const run of runs) { if ( run.state !== "observed" && @@ -470,9 +482,18 @@ async function applyDiscoveryDelete(db: D1Database, uri: string, now: Date): Pro run.state !== "running" ) continue; - if (run.state === "pending" || run.state === "running") hadPending.push(run); + // observed/verifying runs never issued a pending label (it is issued once a + // run reaches `pending`); pending/running runs carry one, so negate before + // the cancellation retires the run out of the recovery set. + if (run.state === "pending" || run.state === "running") + await negateRunPendingLabel(deps, run, now); try { - await transitionAssessmentState(db, { id: run.id, from: run.state, to: "cancelled", now }); + await transitionAssessmentState(deps.db, { + id: run.id, + from: run.state, + to: "cancelled", + now, + }); } catch (err) { // A concurrent invocation already moved this run past `from` — // harmless, the delete's intent (no non-terminal run survives) is @@ -480,38 +501,36 @@ async function applyDiscoveryDelete(db: D1Database, uri: string, now: Date): Pro if (!(err instanceof AssessmentTransitionConflictError)) throw err; } } - return hadPending; } /** - * Negates the `assessment-pending` label each cancelled run issued, so a - * deleted release stops advertising an in-progress assessment. Uses the - * run's own assessment id and a deterministic idempotency key, so a - * redelivered delete converges. Idempotent and best-effort per run: a run - * whose pending is already negated (finalized) no-ops. + * Negates one run's `assessment-pending` label so a deleted release stops + * advertising an in-progress assessment. Deterministic idempotency key (the + * run's runKey), so a redelivered delete converges and a run whose pending is + * already negated (finalized) no-ops. Publishes best-effort with the sweep as + * backstop. Throws `LabelIssuanceUnavailableError` when signing is paused — the + * caller must let that propagate so the delete retries. */ -async function negatePendingForDeletedRuns( +async function negateRunPendingLabel( deps: DiscoveryConsumerDeps, - runs: Assessment[], + run: Assessment, now: Date, ): Promise { - for (const run of runs) { - await issueAutomatedAssessmentLabel( - deps.db, - deps.config, - deps.signer, - { - actor: deps.config.labelerDid, - type: "automated-assessment", - assessmentId: run.id, - reason: "subject deleted", - idempotencyKey: automatedIdempotencyKey(run.runKey, "assessment-pending", true), - }, - { uri: run.uri, cid: run.cid, val: "assessment-pending", neg: true }, - now, - deps.publisher, - ); - } + await issueAutomatedAssessmentLabel( + deps.db, + deps.config, + deps.signer, + { + actor: deps.config.labelerDid, + type: "automated-assessment", + assessmentId: run.id, + reason: "subject deleted", + idempotencyKey: automatedIdempotencyKey(run.runKey, "assessment-pending", true), + }, + { uri: run.uri, cid: run.cid, val: "assessment-pending", neg: true }, + now, + deps.publisher, + ); } function jobUri(job: DiscoveryJob): string { diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 68fbbfd93a..9aee3a57dd 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -601,6 +601,63 @@ describe("processDiscoveryMessage: delete", () => { expect(dl?.n).toBe(0); }); + it("negates the pending label on redelivery after a paused first delivery — no stale pending survives", async () => { + const job = await jobFor({ rkey: rkey() }); + const deps = await buildDeps(); + await processDiscoveryMessage(job, new FakeMessage(), { ...deps, verify: verifiedFor(job) }); + const runKey = await runKeyFor(job); + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("pending"); + + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + const latestPending = () => + testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uriFor(job)) + .first<{ neg: number }>(); + + // First delivery: signing paused mid-rotation, so the pending-negation throws + // and the message retries. + const first = new FakeMessage(); + await testEnv.DB.prepare( + `UPDATE signing_state SET phase = 'paused', pending_key_version = 'v2', + pending_public_multikey = ?, rotation_id = 'rot-redeliver' WHERE id = 1`, + ) + .bind(MULTIKEY) + .run(); + try { + await processDiscoveryMessage(deleteJob, first, { + ...deps, + confirmDeleted: () => Promise.resolve(true), + }); + } finally { + await testEnv.DB.prepare( + `UPDATE signing_state SET phase = 'active', pending_key_version = NULL, + pending_public_multikey = NULL, rotation_id = NULL WHERE id = 1`, + ).run(); + } + expect(first.retried).toBe(1); + expect(first.acked).toBe(0); + // The crux of the fix: negating BEFORE cancelling means the throw leaves the + // run non-terminal (still `pending`), so the redelivery can re-discover it. + // Pre-fix the run was cancelled here and the redelivery found nothing to negate. + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("pending"); + expect((await latestPending())?.neg).toBe(0); + + // Redelivery with signing resumed: the negation now commits and the run is + // retired, so no active assessment-pending survives the delete. + const second = new FakeMessage(); + await processDiscoveryMessage(deleteJob, second, { + ...deps, + confirmDeleted: () => Promise.resolve(true), + }); + expect(second.acked).toBe(1); + expect(second.retried).toBe(0); + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("cancelled"); + expect((await latestPending())?.neg).toBe(1); + }); + it("dead-letters a forged/premature delete whose record still resolves, suppressing nothing", async () => { const job = await jobFor({ rkey: rkey() }); const deps = await buildDeps(); From 21e061198456dad2e1b693cf672ccd6627615f44 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 20:56:47 +0100 Subject: [PATCH 09/15] fix(labeler): guard the finalization CAS on the subject not being tombstoned MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finalization's isSubjectCurrent re-check is a separate read from its commit CAS, a TOCTOU: a delete that tombstoned the subject in between still let the CAS commit 'running -> outcome' plus positive outcome/block labels, then the delete's own cancel CAS conflicted and was swallowed — leaving live block labels on a deleted release the evaluator would still honor. The finalization CAS now carries a not-deleted predicate on the run's subject (same all-or-nothing batch idiom as the F6 signing guard), so a tombstone landing before the batch no-ops the CAS and every label gated on the outcome state; finalization retries and stales the run out, never labelling a deleted subject. Follow-up to Sol round-3 review on #2115. --- apps/labeler/src/assessment-orchestrator.ts | 12 ++++ apps/labeler/src/assessment-store.ts | 20 ++++++- .../test/assessment-orchestrator.test.ts | 60 +++++++++++++++++++ 3 files changed, 91 insertions(+), 1 deletion(-) diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 77255244b8..f577d1f146 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -290,6 +290,10 @@ export class AssessmentOrchestrator { isPrebootstrap: signingStatus === null, activeKeyVersion: this.config.signingKeyVersion, }, + // Close the delete-vs-finalization TOCTOU: a delete tombstoning the + // subject after the currency re-check below no-ops this CAS at commit + // time, so no outcome/block label commits for a deleted subject. + guardSubjectNotDeleted: true, }); const statements = [...finalization.statements]; @@ -379,6 +383,14 @@ export class AssessmentOrchestrator { // signingGuard), so a flip no-ops the CAS too and the batch commits nothing — // the run stays `running` for the Workflow retry. // + // A delete tombstoning the subject in this gap is closed likewise: the CAS + // carries a not-deleted predicate on this run's subject + // (buildFinalizationStatements' guardSubjectNotDeleted). A pure tombstone does + // not move the run out of `running` (that only happens via the delete's + // separate cancel CAS, which can lose the race), so without this predicate the + // CAS would commit outcome/block labels for a subject the delete just removed; + // with it, the tombstone no-ops the CAS and the retry stales the run out. + // // Narrower gaps remain, tracked with the real-stage wiring: // - a CID supersession landing in this gap does not move the run out of // `running`, so the CAS succeeds and this run finalizes labels for its own diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index 7c3c1e5dfd..85848cf72a 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/apps/labeler/src/assessment-store.ts @@ -484,6 +484,17 @@ export interface FinalizationInput { * stays `running`, and the Workflow retry re-runs finalization after resume. */ signingGuard?: { isPrebootstrap: boolean; activeKeyVersion: string }; + /** + * Gate the CAS on the subject `(uri, cid)` still being non-tombstoned at commit + * time, closing the delete-vs-finalization TOCTOU: finalization's + * `isSubjectCurrent` re-check is a separate read from this commit, so a delete + * that tombstones the subject in between would otherwise let the CAS commit + * terminal state with live outcome/block labels for a deleted release. Folding + * the not-deleted predicate into the CAS makes a tombstone landing before the + * batch no-op the CAS (and, transitively, every label gated on `toState`), so + * finalization retries and stales out instead of labelling a deleted subject. + */ + guardSubjectNotDeleted?: boolean; } export interface FinalizationStatements { @@ -538,6 +549,13 @@ export function buildFinalizationStatements( const prebootstrap = input.signingGuard.isPrebootstrap ? 1 : 0; casBinds.push(prebootstrap, prebootstrap, input.signingGuard.activeKeyVersion); } + // Not-tombstoned predicate on this run's subject, evaluated at commit time in + // the same batch — a delete landing after finalization's currency re-check + // no-ops the CAS here rather than committing labels for a deleted subject. + const subjectGuardSql = input.guardSubjectNotDeleted + ? `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects WHERE uri = ? AND cid = ? AND deleted_at IS NULL)` + : ""; + if (input.guardSubjectNotDeleted) casBinds.push(input.uri, input.cid); const statements: D1PreparedStatement[] = [ db .prepare( @@ -546,7 +564,7 @@ export function buildFinalizationStatements( public_summary = COALESCE(?, public_summary), coverage_json = COALESCE(?, coverage_json), supersedes_assessment_id = COALESCE(?, supersedes_assessment_id) - WHERE id = ? AND state = ?${signingGuardSql}`, + WHERE id = ? AND state = ?${signingGuardSql}${subjectGuardSql}`, ) .bind(...casBinds), ]; diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index 2fd378a372..f4c2634590 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -1327,3 +1327,63 @@ describe("AssessmentOrchestrator: signing pause between prep and commit (Finding expect(pendingNeg?.neg).toBe(1); }); }); + +describe("AssessmentOrchestrator: delete racing finalization (Blocker 1)", () => { + it("commits no outcome/block labels when the subject is tombstoned between the currency re-check and the batch", async () => { + const run = await pendingRun({ name: "delete-toctou", cidValue: await cid("delete-toctou") }); + const blockingStages: OrchestratorStages = { + ...stubStages, + codeAi: () => Promise.resolve([finding({ category: "malware", severity: "critical" })]), + }; + // Tombstone the subject in the seam between finalize's currency re-check and + // the batch commit — the exact TOCTOU window the CAS subject-guard closes. + const db = pauseBeforeFirstBatch(testEnv.DB, async () => { + await deleteSubject(testEnv.DB, { uri: run.uri, cid: run.cid }); + }); + const orchestrator = new AssessmentOrchestrator({ + db, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: blockingStages, + sleep: () => Promise.resolve(), + }); + + // The guarded CAS no-ops against the tombstone, so the whole batch commits + // nothing and the finalization conflict is raised. + await expect(orchestrator.runAssessment(run.id)).rejects.toBeInstanceOf( + AssessmentFinalizationConflictError, + ); + + // The run is still running and NO label — not the block, not even the pending + // negation — was committed for the now-deleted subject. + expect((await getAssessment(testEnv.DB, run.id))?.state).toBe("running"); + const labels = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ?`, + ) + .bind(run.id) + .first<{ n: number }>(); + expect(labels?.n).toBe(0); + + // The Workflow retry re-runs finalization, sees the deleted subject, and + // stales the run out — still never labelling a deleted release. + const resumed = new AssessmentOrchestrator({ + db: testEnv.DB, + config, + signer: await signer(), + policy: MODERATION_POLICY, + stages: blockingStages, + sleep: () => Promise.resolve(), + }); + const finalized = await resumed.runAssessment(run.id); + expect(finalized.state).toBe("stale"); + const labelsAfter = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ?`, + ) + .bind(run.id) + .first<{ n: number }>(); + expect(labelsAfter?.n).toBe(0); + }); +}); From 0d68908a69f55343a4e182a87d71736d7b3cb177 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 20:57:01 +0100 Subject: [PATCH 10/15] fix(labeler): retry (never dead-letter+ack) a delete-path mutation failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 'a negation throw prevents ack' invariant only held for recognized issuance-unavailable errors; an unexpected signing/D1 failure during the delete's tombstone/negate/cancel fell through to dead-letter+ack (the create-path policy), leaving the run non-terminal with its assessment-pending label live on a deleted subject forever. The delete handler now splits the absence-verification phase (classified like create: transient retries, forged/permanent dead-letters) from the mutation phase, which ALWAYS retries on failure so the label can never be left live on an acked delete. A genuinely permanent fault exhausts to the DLQ via max_retries — acceptable versus acking a stranded label. Follow-up to Sol round-3 review on #2115. --- apps/labeler/src/discovery-consumer.ts | 50 +++++++++++------- apps/labeler/test/discovery-consumer.test.ts | 53 ++++++++++++++++++++ 2 files changed, 86 insertions(+), 17 deletions(-) diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 500191a3f7..24dc8ef5b1 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -213,32 +213,48 @@ export async function processDiscoveryMessage( const uri = jobUri(job); if (job.operation === "delete") { + // A delete suppresses assessment work (tombstone + cancel runs), so it gets + // the same distrust as a create: confirm the record is genuinely gone at the + // PDS before acting. A still-present record means a forged or premature + // delete — dead-letter it, suppress nothing. A verification failure here + // classifies like the create path (transient retries, permanent dead-letters). + let absent: boolean; try { - // A delete suppresses assessment work (tombstone + cancel runs), so it - // gets the same distrust as a create: confirm the record is genuinely - // gone at the PDS before acting. A still-present record means a forged - // or premature delete — dead-letter it, suppress nothing. const confirmAbsent = deps.confirmDeleted ?? confirmRecordAbsent; - const absent = await confirmAbsent({ + absent = await confirmAbsent({ uri, didDocumentResolver: deps.didDocumentResolver, ...(deps.fetch ? { fetch: deps.fetch } : {}), }); - if (!absent) { - await writeDeadLetter( - deps.db, - job, - "DELETE_RECORD_PRESENT", - "record still resolves", - now(), - ); - controller.ack(); - return; - } + } catch (err) { + await classifyDiscoveryError(err, job, deps, controller, now()); + return; + } + if (!absent) { + await writeDeadLetter(deps.db, job, "DELETE_RECORD_PRESENT", "record still resolves", now()); + controller.ack(); + return; + } + try { await applyDiscoveryDelete(deps, uri, now()); controller.ack(); } catch (err) { - await classifyDiscoveryError(err, job, deps, controller, now()); + // The mutation phase (tombstone → pending-negation → cancellation) can + // leave a run's `assessment-pending` label live on a now-deleted subject + // if it fails partway. Acking here — as the create path's unexpected-error + // policy would — strands that label forever, so ALWAYS retry. Redelivery + // re-attempts idempotently; a genuinely permanent failure exhausts to the + // DLQ, which is acceptable versus acking a live label on a deleted subject. + console.error( + "[labeler] discovery delete mutation failed; retrying to avoid a stranded label", + { + did: job.did, + collection: job.collection, + rkey: job.rkey, + error: err instanceof Error ? err.message : String(err), + }, + ); + controller.retry(); } return; } diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 9aee3a57dd..7c3decea51 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -658,6 +658,59 @@ describe("processDiscoveryMessage: delete", () => { expect((await latestPending())?.neg).toBe(1); }); + it("retries (never dead-letter+acks) an unexpected issuance error during delete negation, leaving no acked live label", async () => { + const job = await jobFor({ rkey: rkey() }); + const deps = await buildDeps(); + await processDiscoveryMessage(job, new FakeMessage(), { ...deps, verify: verifiedFor(job) }); + const runKey = await runKeyFor(job); + + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + const latestPending = () => + testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uriFor(job)) + .first<{ neg: number }>(); + + // An UNEXPECTED (not issuance-unavailable) failure during the negation — a + // signer that throws a plain error, standing in for an HSM/D1 fault. + const brokenSigner: LabelSigner = { + issuerDid: LABELER_DID, + sign: () => Promise.reject(new Error("HSM offline")), + }; + const first = new FakeMessage(); + await processDiscoveryMessage(deleteJob, first, { + ...deps, + signer: brokenSigner, + confirmDeleted: () => Promise.resolve(true), + }); + + // The delete path RETRIES rather than dead-letter+acking (the create path's + // unexpected-error policy) — acking would strand the live pending label on a + // deleted subject. + expect(first.retried).toBe(1); + expect(first.acked).toBe(0); + const dl = await testEnv.DB.prepare(`SELECT COUNT(*) AS n FROM dead_letters WHERE rkey = ?`) + .bind(job.rkey) + .first<{ n: number }>(); + expect(dl?.n).toBe(0); + // The pending label is still live and the run non-terminal — recoverable. + expect((await latestPending())?.neg).toBe(0); + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("pending"); + + // Redelivery with a working signer completes the delete and negates the label. + const second = new FakeMessage(); + await processDiscoveryMessage(deleteJob, second, { + ...deps, + confirmDeleted: () => Promise.resolve(true), + }); + expect(second.acked).toBe(1); + expect(second.retried).toBe(0); + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("cancelled"); + expect((await latestPending())?.neg).toBe(1); + }); + it("dead-letters a forged/premature delete whose record still resolves, suppressing nothing", async () => { const job = await jobFor({ rkey: rkey() }); const deps = await buildDeps(); From bcd20f794a44a4d533e8867d4ac0a373c9a785b9 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 21:06:35 +0100 Subject: [PATCH 11/15] docs(labeler): correct the delete-mutation retry reference post-merge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After merging the integration branch (which added #2113's transient-DNS retry backoff), the applyDiscoveryDelete docstring still said a failed negation propagates to classifyDiscoveryError. Since the Blocker-2 fix, the delete mutation phase has its own always-retry catch — update the comment to match. No behavior change. --- apps/labeler/src/discovery-consumer.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 25d6ca8303..c0b057e188 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -494,8 +494,8 @@ async function transitionOrObserve( * pending label live forever once the message acks. The invariant: a run is * cancelled only after its pending negation has committed, and the message * cannot ack while any pending/running run is still un-negated (a throw - * propagates to `classifyDiscoveryError` → retry), so no active - * `assessment-pending` survives an acked delete. + * propagates to the delete handler's mutation-phase catch, which always retries), + * so no active `assessment-pending` survives an acked delete. */ async function applyDiscoveryDelete( deps: DiscoveryConsumerDeps, From 5cd4cc88cc267207a8e567689a6f7a2dad987371 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 22:07:29 +0100 Subject: [PATCH 12/15] fix(labeler): guard the initial pending-label issuance against a concurrent delete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The create path committed a recreated run as pending and THEN issued its positive assessment-pending with no commit-time guard. A concurrent delete could, in the gap, tombstone the subject, negate + cancel the (still-positive-less) run, and ack; the create then committed its positive at a higher sequence, winning label reduction — a live pending label on a tombstoned, cancelled subject. (Queue invocations overlap; max_concurrency is unset.) The initial issuance now goes through buildIssuanceStatements gated atomically on BOTH the run still being pending AND the exact (uri,cid) subject still undeleted (the same guard idiom as Blocker 1's finalization CAS). On a guard miss the label no-ops: the issuance is obsolete, so the create neither publishes nor dispatches a Workflow for a label that never committed; a non-persist not explained by the guard (a signing flip) still retries. With this and Blocker 1, every automated- positive issuance site in the discovery consumer is now delete-guarded (the delete-path negation is negative and exempt). Follow-up to Sol round-4 review on #2115. --- apps/labeler/src/assessment-store.ts | 18 ++++ apps/labeler/src/discovery-consumer.ts | 101 ++++++++++++++++--- apps/labeler/src/service.ts | 18 +++- apps/labeler/test/discovery-consumer.test.ts | 77 +++++++++++++- 4 files changed, 196 insertions(+), 18 deletions(-) diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index 85848cf72a..0d5a19fa39 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/apps/labeler/src/assessment-store.ts @@ -138,6 +138,24 @@ export async function deleteSubjectsByUri( .run(); } +/** + * True when a non-tombstoned subject row exists for this exact `(uri, cid)` — the + * same condition the `requireSubjectNotDeleted` issuance guard checks. Distinct + * from `isSubjectCurrent`: this ignores CID-supersession, so it classifies a + * guarded-issuance miss precisely (a superseded-but-undeleted subject would not + * miss the guard, so it must not read as deleted here). + */ +export async function subjectIsUndeleted( + db: D1Database, + input: { uri: string; cid: string }, +): Promise { + const row = await db + .prepare(`SELECT 1 FROM subjects WHERE uri = ? AND cid = ? AND deleted_at IS NULL`) + .bind(input.uri, input.cid) + .first(); + return row !== null; +} + /** * A subject is current when its row isn't tombstoned and no later-observed, * non-deleted subject at the same URI carries a different CID. Used diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index c0b057e188..bd6d0ac509 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -51,6 +51,7 @@ import { deleteSubjectsByUri, getAssessment, listNonTerminalAssessmentsForUri, + subjectIsUndeleted, transitionAssessmentState, type Assessment, } from "./assessment-store.js"; @@ -71,7 +72,14 @@ import { RecordVerificationError, type RecordVerificationFailureReason, } from "./record-verification.js"; -import { issueAutomatedAssessmentLabel, LabelIssuanceUnavailableError } from "./service.js"; +import { + buildIssuanceStatements, + issueAutomatedAssessmentLabel, + LabelIssuanceUnavailableError, + readIssuedLabelByActionKey, + type AutomatedIssuanceAction, + type AutomatedLabelProposal, +} from "./service.js"; import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js"; import { createLabelPublisher, type LabelPublisher } from "./subscribe-labels.js"; @@ -423,21 +431,15 @@ async function verifyAndCreateRun( await advanceToPending(deps.db, assessment, now); - await issueAutomatedAssessmentLabel( - deps.db, - deps.config, - deps.signer, - { - actor: deps.config.labelerDid, - type: "automated-assessment", - assessmentId: assessment.id, - reason: "initial discovery", - idempotencyKey: automatedIdempotencyKey(runKey, "assessment-pending", false), - }, - { uri, cid: job.cid, val: "assessment-pending" }, - now, - deps.publisher, - ); + const outcome = await issueInitialPendingLabel(deps, assessment.id, runKey, uri, job.cid, now); + if (outcome === "obsolete") { + // A concurrent delete tombstoned the subject (or cancelled the run) before + // the positive assessment-pending could commit — the run is moot. The + // issuance no-op'd, so there is nothing to publish and nothing to assess; + // the delete owns tombstone + cancel + negation. Do not dispatch a Workflow + // for a label that never committed. + return; + } // Hand the run to its Workflow instance. The instance id is the run's runKey, // so a redelivered event (same runKey) converges on the same instance rather @@ -450,6 +452,73 @@ async function verifyAndCreateRun( }); } +/** + * Issues the initial positive `assessment-pending` label, atomically gated at + * commit on BOTH the run still being `pending` AND the subject `(uri, cid)` still + * undeleted (`buildIssuanceStatements`' `requireAssessmentState` + + * `requireSubjectNotDeleted`). A concurrent delete that tombstones the subject or + * cancels the run in the gap after `advanceToPending` makes the guarded insert + * match no row: the issuance is obsolete — no label commits, so this returns + * without publishing or signalling a dispatch, and the delete's negation owns the + * stream. A non-persist NOT explained by the guard (a signing flip mid-batch) + * throws `LabelIssuanceUnavailableError` so the message retries. Reuses the same + * guarded-issuance machinery as finalization and the console path (no second SQL + * path); `readIssuedLabelByActionKey` tolerates a legitimate no-op without the + * signing-diagnosis throw `issueLabel`'s post-commit applies. + */ +async function issueInitialPendingLabel( + deps: DiscoveryConsumerDeps, + assessmentId: string, + runKey: string, + uri: string, + cid: string, + now: Date, +): Promise<"issued" | "obsolete"> { + const idempotencyKey = automatedIdempotencyKey(runKey, "assessment-pending", false); + const action: AutomatedIssuanceAction = { + actor: deps.config.labelerDid, + type: "automated-assessment", + assessmentId, + reason: "initial discovery", + idempotencyKey, + }; + const proposal: AutomatedLabelProposal = { uri, cid, val: "assessment-pending" }; + + // A redelivery whose first attempt already committed the label converges here: + // re-drive the live notify (best-effort) and treat it as issued. + const existing = await readIssuedLabelByActionKey(deps.db, idempotencyKey); + if (existing) { + if (deps.publisher) await deps.publisher.publish(existing); + return "issued"; + } + + const { statements } = await buildIssuanceStatements( + deps.db, + deps.config, + deps.signer, + action, + proposal, + now, + deps.publisher !== undefined, + { requireAssessmentState: "pending", requireSubjectNotDeleted: { uri, cid } }, + ); + await deps.db.batch(statements); + + const issued = await readIssuedLabelByActionKey(deps.db, idempotencyKey); + if (issued) { + if (deps.publisher) await deps.publisher.publish(issued); + return "issued"; + } + + // The guarded insert matched no row. If the run is no longer `pending` or the + // subject was tombstoned, a concurrent delete won — a benign no-op. Anything + // else (the signing guard no-op'ing on a mid-batch pause/rotation) is retryable. + const run = await getAssessment(deps.db, assessmentId); + if (!run || run.state !== "pending" || !(await subjectIsUndeleted(deps.db, { uri, cid }))) + return "obsolete"; + throw new LabelIssuanceUnavailableError("initial pending label did not persist"); +} + /** * Advances a freshly-created (or redelivered) run from `observed` to * `pending`, tolerating a concurrent invocation that already did some or all diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index 6dd5e39d3c..c9f4149597 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -120,6 +120,16 @@ export interface BuildIssuanceOptions { * automated-assessment action (it carries the assessmentId to check). */ requireAssessmentState?: AssessmentState; + /** + * Gate the action insert (and thus its label) additionally on the subject + * `(uri, cid)` still being non-tombstoned at commit time. The initial discovery + * issuance pairs this with `requireAssessmentState: "pending"` so a concurrent + * delete that tombstones the subject or cancels the run in the gap before this + * commit makes the positive `assessment-pending` no-op — no live label is + * resurrected for a deleted release. Gating the action (not the label) leaves no + * orphan label, same as the state guard. + */ + requireSubjectNotDeleted?: { uri: string; cid: string }; } /** @@ -214,6 +224,11 @@ export async function buildIssuanceStatements( requireState === undefined ? "" : `\n\t\t\t\t AND EXISTS (SELECT 1 FROM assessments WHERE id = ? AND state = ?)`; + const requireSubject = options.requireSubjectNotDeleted; + const subjectGuardSql = + requireSubject === undefined + ? "" + : `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects WHERE uri = ? AND cid = ? AND deleted_at IS NULL)`; const actionBinds: unknown[] = [ action.actor, action.type, @@ -233,6 +248,7 @@ export async function buildIssuanceStatements( proposal.val, ]; if (requireState !== undefined) actionBinds.push(assessmentId, requireState); + if (requireSubject !== undefined) actionBinds.push(requireSubject.uri, requireSubject.cid); const statements: D1PreparedStatement[] = [ db @@ -257,7 +273,7 @@ export async function buildIssuanceStatements( ) AND l2.neg = 0 AND a2.type <> 'automated-assessment' ) - )${stateGuardSql} + )${stateGuardSql}${subjectGuardSql} ON CONFLICT(idempotency_key) DO NOTHING`, ) .bind(...actionBinds), diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 4a78f93ca9..adec284218 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -33,7 +33,7 @@ import { RecordVerificationError, type DidDocumentResolverLike, } from "../src/record-verification.js"; -import { issueAutomatedAssessmentLabel } from "../src/service.js"; +import { issueAutomatedAssessmentLabel, type IssuedLabel } from "../src/service.js"; import { initializeSigningState } from "../src/signing-rotation.js"; import type { LabelPublisher } from "../src/subscribe-labels.js"; @@ -561,6 +561,81 @@ describe("processDiscoveryMessage: verification failures", () => { }); }); +describe("processDiscoveryMessage: create racing delete (Blocker 1 create-path)", () => { + it("commits no live pending label and does not dispatch when a delete completes before the create's positive commits", async () => { + const job = await jobFor({ rkey: rkey() }); + const workflow = new FakeAssessmentWorkflow(); + const published: IssuedLabel[] = []; + const publisher: LabelPublisher = { + managesPublicationState: true, + async publish(issued) { + published.push(issued); + }, + }; + const baseDeps = { ...(await buildDeps()), assessmentWorkflow: workflow, publisher }; + const runKey = await runKeyFor(job); + + // Barrier: the create's positive issuance signs its label right before the + // commit batch. Hook that seam to let a full delete complete first — + // tombstone + negate + cancel + ack — exactly the concurrent-delete window. + const realSigner = await signer(); + let deleteDone = false; + const barrierSigner: LabelSigner = { + issuerDid: realSigner.issuerDid, + async sign(label) { + if (!deleteDone) { + deleteDone = true; + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + await processDiscoveryMessage(deleteJob, new FakeMessage(), { + ...baseDeps, + confirmDeleted: () => Promise.resolve(true), + }); + } + return realSigner.sign(label); + }, + }; + + const create = new FakeMessage(); + await processDiscoveryMessage(job, create, { + ...baseDeps, + signer: barrierSigner, + verify: verifiedFor(job), + }); + + // The create acked (the run is obsolete), never dispatched a Workflow, and + // committed no positive assessment-pending — the guarded issuance no-op'd. + expect(create.acked).toBe(1); + expect(create.retried).toBe(0); + expect(workflow.created.length).toBe(0); + + const assessment = await getAssessmentByRunKey(testEnv.DB, runKey); + expect(assessment?.state).toBe("cancelled"); + const positive = await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels l JOIN issuance_actions a ON a.id = l.action_id + WHERE a.assessment_id = ? AND l.val = 'assessment-pending' AND l.neg = 0`, + ) + .bind(assessment!.id) + .first<{ n: number }>(); + expect(positive?.n).toBe(0); + + // No positive assessment-pending was ever broadcast. + expect( + published.some( + (entry) => entry.label.val === "assessment-pending" && entry.label.neg !== true, + ), + ).toBe(false); + + // The active label state carries no live pending — only the delete's negation. + const winner = await testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uriFor(job)) + .first<{ neg: number }>(); + expect(winner?.neg).toBe(1); + }); +}); + describe("processDiscoveryMessage: delete", () => { it("tombstones the subject and cancels non-terminal runs", async () => { const job = await jobFor({ rkey: rkey() }); From 94fb562f08cad3c6e2a374f1cedbc039b68dac9d Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 18 Jul 2026 22:33:35 +0100 Subject: [PATCH 13/15] fix(labeler): guard the operator-rerun pending label against a concurrent delete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The operator rerun issued its positive assessment-pending with no commit-time guard and while its run was still 'observed' (advance to 'pending' is deferred), opening two integrity races with a concurrent discovery-delete (queue + HTTP overlap; max_concurrency unset): (a) ordering: the rerun commits its run+pending after a delete tombstoned and snapshotted the non-terminal runs, so the delete never negates it and the positive survives on the tombstoned subject; (b) observed gap: the delete-path negation only fired for pending/running runs, so a delete that saw the rerun's 'observed' run cancelled it WITHOUT negating its already-live positive. Fix (a): gate the rerun's assessment-pending issuance on requireSubjectNotDeleted + requireAssessmentState (threaded through prepareAutomatedLabelIssuance); on a miss the label no-ops, assertIssuancePersisted aborts before the deferred tail, so nothing is published, dispatched, or advanced. Fix (b): the delete-path negation now fires for any non-terminal run that committed a positive pending (keyed on the committed positive, not lifecycle state) — no dangling negations, no interleaving where a cancelled/deleted run keeps a live positive. With Blocker 1 and the round-4 discovery guard, every positive assessment-pending issuance across the discovery and console paths is now delete-safe. Follow-up to adversary review on #2115. --- apps/labeler/src/console-mutation-api.ts | 10 +++ apps/labeler/src/discovery-consumer.ts | 38 ++++---- apps/labeler/src/service.ts | 3 +- .../test/console-assessment-mutations.test.ts | 35 ++++++++ apps/labeler/test/discovery-consumer.test.ts | 86 ++++++++++++++++++- 5 files changed, 153 insertions(+), 19 deletions(-) diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index add22d6a31..3103e31e68 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -730,6 +730,16 @@ async function runRerun( }, { uri: assessment.uri, cid: assessment.cid, val: "assessment-pending" }, ctx.now, + // Gate the rerun's positive assessment-pending on the run (created `observed` + // in this batch) and the subject still being undeleted at commit time, so a + // concurrent discovery-delete that tombstoned the subject makes the positive + // no-op instead of resurrecting a live label on a deleted release. On a miss + // the label does not persist -> `assertIssuancePersisted` below aborts before + // `deferRerunTail`, so nothing is published, dispatched, or advanced. + { + requireAssessmentState: "observed", + requireSubjectNotDeleted: { uri: assessment.uri, cid: assessment.cid }, + }, ); const descriptor: RerunDescriptor = { diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index bd6d0ac509..be04c51e78 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -555,16 +555,22 @@ async function transitionOrObserve( /** * Tombstones the subject and retires its non-terminal runs. For each run that - * reached `pending`/`running` — and therefore carries an active - * `assessment-pending` label — the negation is issued BEFORE the terminal - * cancellation, so a failed or paused negation (signing mid-rotation) leaves the - * run non-terminal and re-discoverable by `listNonTerminalAssessmentsForUri` on - * redelivery. Cancelling first would drop the run from that set, stranding the - * pending label live forever once the message acks. The invariant: a run is - * cancelled only after its pending negation has committed, and the message - * cannot ack while any pending/running run is still un-negated (a throw - * propagates to the delete handler's mutation-phase catch, which always retries), - * so no active `assessment-pending` survives an acked delete. + * committed a positive `assessment-pending` label, the negation is issued BEFORE + * the terminal cancellation, so a failed or paused negation (signing + * mid-rotation) leaves the run non-terminal and re-discoverable by + * `listNonTerminalAssessmentsForUri` on redelivery. Cancelling first would drop + * the run from that set, stranding the pending label live forever once the + * message acks. + * + * The negation is keyed on the run having committed a live positive — NOT on its + * lifecycle state — because an operator rerun issues its positive + * `assessment-pending` while the run is still `observed` (the advance to `pending` + * is deferred). Keying on state would let a delete cancel that `observed` run + * without negating its already-live positive. The invariant: a run is cancelled + * only after any positive it committed has been negated, and the message cannot + * ack while a negation is still owed (a throw propagates to the delete handler's + * mutation-phase catch, which always retries), so no active `assessment-pending` + * survives an acked delete. */ async function applyDiscoveryDelete( deps: DiscoveryConsumerDeps, @@ -581,11 +587,13 @@ async function applyDiscoveryDelete( run.state !== "running" ) continue; - // observed/verifying runs never issued a pending label (it is issued once a - // run reaches `pending`); pending/running runs carry one, so negate before - // the cancellation retires the run out of the recovery set. - if (run.state === "pending" || run.state === "running") - await negateRunPendingLabel(deps, run, now); + // Negate before cancelling any run that committed a positive pending, + // regardless of lifecycle state (a rerun's is live while `observed`). + const positive = await readIssuedLabelByActionKey( + deps.db, + automatedIdempotencyKey(run.runKey, "assessment-pending", false), + ); + if (positive) await negateRunPendingLabel(deps, run, now); try { await transitionAssessmentState(deps.db, { id: run.id, diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index c9f4149597..57501189be 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -463,13 +463,14 @@ export async function prepareAutomatedLabelIssuance( action: AutomatedIssuanceAction, proposal: AutomatedLabelProposal, now: Date, + options: BuildIssuanceOptions = {}, ): Promise { if (signer.issuerDid !== config.labelerDid) throw new TypeError("signer issuer does not match the configured labeler DID"); validateKeyVersion(config.signingKeyVersion); validateAutomatedAction(action); validateAutomatedProposal(proposal); - return buildIssuanceStatements(db, config, signer, action, proposal, now, true); + return buildIssuanceStatements(db, config, signer, action, proposal, now, true, options); } export interface OverrideIssuanceSpec { diff --git a/apps/labeler/test/console-assessment-mutations.test.ts b/apps/labeler/test/console-assessment-mutations.test.ts index 6521f0a3ce..e1d1955850 100644 --- a/apps/labeler/test/console-assessment-mutations.test.ts +++ b/apps/labeler/test/console-assessment-mutations.test.ts @@ -17,6 +17,7 @@ import { computeRunKey, initialTriggerId } from "../src/assessment-lifecycle.js" import { createAssessmentRun, createSubject, + deleteSubjectsByUri, getActiveLabelState, getAssessment, getCurrentAssessment, @@ -375,6 +376,40 @@ describe("rerun", () => { expect(await getCurrentAssessment(testEnv.DB, { src: LABELER_DID, uri, cid: CID })).toBeNull(); }); + it("commits no live pending and does not dispatch when a delete tombstoned the subject first (rerun-path ordering race)", async () => { + const { id, uri } = await seedRun("rerun-delete-ordering"); + // The concurrent discovery-delete already tombstoned the subject (its + // non-terminal-run snapshot predates the rerun's run, so it never sees it). + await deleteSubjectsByUri(testEnv.DB, { uri }); + + const { deps, workflow, settle } = captureDeferred(); + const response = await handleConsoleMutation( + post(`/admin/api/assessments/${id}/rerun`, { + confirmation: CID, + reason: "rerun races a delete", + idempotencyKey: nextKey(), + }), + deps, + ); + + // The subject-guarded positive no-op'd, so the phantom-success check fails + // (503) and the deferred tail — advance, dispatch, publish — never runs. + expect(response.status).toBe(503); + await settle(); + expect(workflow.created.length).toBe(0); + + // No active positive assessment-pending survives on the tombstoned subject, + // and no positive row committed at all. + const winners = await getActiveLabelState(testEnv.DB, { src: LABELER_DID, uri, cid: CID }); + expect(winners.get("assessment-pending")?.active ?? false).toBe(false); + expect( + await countRows( + `SELECT COUNT(*) n FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' AND neg = 0`, + uri, + ), + ).toBe(0); + }); + it("a second rerun creates another distinct run", async () => { const { id } = await seedRun("rerun-twice"); const first = await bodyData<{ runId: string }>( diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index adec284218..27d266cea6 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -16,8 +16,14 @@ import { automatedIdempotencyKey, computeRunKey, initialTriggerId, + operatorTriggerId, } from "../src/assessment-lifecycle.js"; -import { getAssessmentByRunKey, getCurrentAssessment } from "../src/assessment-store.js"; +import { + createAssessmentRun, + createSubject, + getAssessmentByRunKey, + getCurrentAssessment, +} from "../src/assessment-store.js"; import { buildAutomationPauseUpdate } from "../src/automation-state.js"; import { bestEffortPublisher, @@ -625,14 +631,88 @@ describe("processDiscoveryMessage: create racing delete (Blocker 1 create-path)" ), ).toBe(false); - // The active label state carries no live pending — only the delete's negation. + // No active positive assessment-pending survives. Here the positive never + // committed (guarded no-op) and the delete issued no negation — it correctly + // negates only runs that committed a positive — so the stream winner is either + // absent or a negation, never a live positive. const winner = await testEnv.DB.prepare( `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' ORDER BY sequence DESC LIMIT 1`, ) .bind(uriFor(job)) .first<{ neg: number }>(); - expect(winner?.neg).toBe(1); + expect(winner === null || winner.neg === 1).toBe(true); + }); + + it("negates an operator rerun's live positive pending even while its run is still observed (rerun observed-state gap)", async () => { + // Reconstruct the rerun's committed state: an `observed` run already carrying + // a live positive assessment-pending (the rerun issues its positive before the + // deferred advance to `pending`). + const job = await jobFor({ rkey: rkey() }); + const uri = uriFor(job); + await createSubject(testEnv.DB, { + uri, + cid: job.cid, + did: PUBLISHER_DID, + collection: RELEASE_COLLECTION, + rkey: job.rkey, + }); + const triggerId = operatorTriggerId("op-observed-gap"); + const runKey = await computeRunKey({ + uri, + cid: job.cid, + policyVersion: MODERATION_POLICY.policyVersion, + modelId: "unassigned", + promptHash: "unassigned", + scannerSetVersion: "unassigned", + triggerId, + }); + const { assessment } = await createAssessmentRun(testEnv.DB, { + runKey, + uri, + cid: job.cid, + trigger: "operator", + triggerId, + policyVersion: MODERATION_POLICY.policyVersion, + coverageJson: "{}", + }); + expect(assessment.state).toBe("observed"); + await issueAutomatedAssessmentLabel( + testEnv.DB, + config, + await signer(), + { + actor: LABELER_DID, + type: "automated-assessment", + assessmentId: assessment.id, + reason: "operator rerun", + idempotencyKey: automatedIdempotencyKey(runKey, "assessment-pending", false), + }, + { uri, cid: job.cid, val: "assessment-pending" }, + ); + + const latestPending = () => + testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uri) + .first<{ neg: number }>(); + expect((await latestPending())?.neg).toBe(0); + + // A discovery delete arrives while the run is still `observed`. + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + const msg = new FakeMessage(); + await processDiscoveryMessage(deleteJob, msg, { + ...(await buildDeps()), + confirmDeleted: () => Promise.resolve(true), + }); + + expect(msg.acked).toBe(1); + // The observed run is cancelled AND its live positive is negated — no stale + // positive survives (pre-fix the state-based negation skipped observed runs). + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("cancelled"); + expect((await latestPending())?.neg).toBe(1); }); }); From 398b4f202aa15638502bc474079a5fb384e99670 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sun, 19 Jul 2026 07:51:52 +0100 Subject: [PATCH 14/15] fix(labeler): structural delete-generation guard closing the create/verify/rerun-vs-delete race class Adds a monotonic delete_generation to subjects (migration 0013). Every delete increments it; create/verify/rerun capture it before reading state/verifying and CAS-guard their subject-undelete, run creation, and label issuance on the generation not having advanced. A decision made before a delete is rejected obsolete; one that captured the post-delete generation still works (republish). Closes three race seams: (1) createSubject un-delete is generation-gated so a stale verify can't resurrect a deleted subject; (2) the delete cleanup now scans pending-bearing runs INCLUDING terminal stale and negates any committed positive; (3) the console rerun's run creation is generation-gated so no orphan observed run. Implementation checkpoint; barrier tests for each seam + a republish test follow. Refs Sol round-5 (systemic close). --- .../0013_subject_delete_generation.sql | 9 + apps/labeler/src/assessment-store.ts | 170 ++++++++++++++---- apps/labeler/src/console-mutation-api.ts | 26 ++- apps/labeler/src/discovery-consumer.ts | 123 ++++++++----- apps/labeler/src/service.ts | 22 ++- 5 files changed, 263 insertions(+), 87 deletions(-) create mode 100644 apps/labeler/migrations/0013_subject_delete_generation.sql diff --git a/apps/labeler/migrations/0013_subject_delete_generation.sql b/apps/labeler/migrations/0013_subject_delete_generation.sql new file mode 100644 index 0000000000..507f11bbc4 --- /dev/null +++ b/apps/labeler/migrations/0013_subject_delete_generation.sql @@ -0,0 +1,9 @@ +-- Monotonic tombstone counter on subjects, the structural close of the +-- delete-vs-{create,verify,rerun} race class. Every delete of a `(uri, cid)` +-- increments `delete_generation`; a create/verify/rerun captures the generation +-- when it reads state / verifies and CAS-guards its subject-undelete, run +-- creation, and label issuance on the generation not having advanced. A stale +-- operation (older generation) is rejected as obsolete; an operation that began +-- AFTER the delete reads the new generation and proceeds (delete-then-republish +-- still works). Backfilled to 0 for existing rows (never-deleted baseline). +ALTER TABLE subjects ADD COLUMN delete_generation INTEGER NOT NULL DEFAULT 0; diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index 0d5a19fa39..6b55b6d553 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/apps/labeler/src/assessment-store.ts @@ -68,6 +68,14 @@ export interface CreateSubjectInput { collection: string; rkey: string; now?: Date; + /** + * The delete-generation the caller captured before verifying. When set, a + * re-observation reactivates the tombstoned row ONLY if the generation still + * matches — a delete that advanced it in the meantime leaves the row tombstoned, + * so a stale verify cannot resurrect a subject deleted after it read state. A + * brand-new row is inserted at generation 0 regardless (no prior delete to race). + */ + expectedGeneration?: number; } /** @@ -75,10 +83,25 @@ export interface CreateSubjectInput { * A verified re-observation reactivates a tombstoned row (the create path * only reaches here after the PDS confirms the record exists, so clearing * `deleted_at` is correct — it closes the delete-then-recreate race and - * handles a genuine republish of the same rkey+cid). + * handles a genuine republish of the same rkey+cid). When `expectedGeneration` + * is supplied the reactivation is gated on the tombstone counter not having + * advanced past the captured value, so a delete concurrent with a slow verify + * cannot be undone by the verify's re-observation. */ export async function createSubject(db: D1Database, input: CreateSubjectInput): Promise { const now = input.now ?? new Date(); + const generationGuard = + input.expectedGeneration === undefined ? "" : `\n\t\t\t WHERE subjects.delete_generation = ?`; + const binds: unknown[] = [ + input.uri, + input.cid, + input.did, + input.collection, + input.rkey, + now.toISOString(), + now.getTime(), + ]; + if (input.expectedGeneration !== undefined) binds.push(input.expectedGeneration); await db .prepare( `INSERT INTO subjects (uri, cid, did, collection, rkey, observed_at, observed_at_epoch_ms) @@ -90,17 +113,9 @@ export async function createSubject(db: D1Database, input: CreateSubjectInput): observed_at = excluded.observed_at, observed_at_epoch_ms = excluded.observed_at_epoch_ms, deleted_at = NULL, - deleted_at_epoch_ms = NULL`, - ) - .bind( - input.uri, - input.cid, - input.did, - input.collection, - input.rkey, - now.toISOString(), - now.getTime(), + deleted_at_epoch_ms = NULL${generationGuard}`, ) + .bind(...binds) .run(); } @@ -111,7 +126,8 @@ export async function deleteSubject( const now = input.now ?? new Date(); await db .prepare( - `UPDATE subjects SET deleted_at = ?, deleted_at_epoch_ms = ? + `UPDATE subjects SET deleted_at = ?, deleted_at_epoch_ms = ?, + delete_generation = delete_generation + 1 WHERE uri = ? AND cid = ? AND deleted_at IS NULL`, ) .bind(now.toISOString(), now.getTime(), input.uri, input.cid) @@ -131,13 +147,53 @@ export async function deleteSubjectsByUri( const now = input.now ?? new Date(); await db .prepare( - `UPDATE subjects SET deleted_at = ?, deleted_at_epoch_ms = ? + `UPDATE subjects SET deleted_at = ?, deleted_at_epoch_ms = ?, + delete_generation = delete_generation + 1 WHERE uri = ? AND deleted_at IS NULL`, ) .bind(now.toISOString(), now.getTime(), input.uri) .run(); } +/** + * The subject's monotonic tombstone counter for `(uri, cid)`, or 0 when the + * subject has never been observed. A create/verify/rerun captures this at the + * point it reads state / verifies, then CAS-guards its subject-undelete, run + * creation, and label issuance on the generation not having advanced — so a + * delete landing after the capture rejects the operation as obsolete, while an + * operation that captured the post-delete generation still proceeds. + */ +export async function readDeleteGeneration( + db: D1Database, + input: { uri: string; cid: string }, +): Promise { + const row = await db + .prepare(`SELECT delete_generation FROM subjects WHERE uri = ? AND cid = ?`) + .bind(input.uri, input.cid) + .first<{ delete_generation: number }>(); + return row?.delete_generation ?? 0; +} + +/** + * True when a non-tombstoned subject row exists for `(uri, cid)` at exactly the + * captured `generation` — the same predicate the generation-guarded issuance and + * run-creation use. Classifies a guarded-issuance miss precisely: a subject the + * caller's generation no longer matches (a newer delete) reads as obsolete. + */ +export async function subjectMatchesGeneration( + db: D1Database, + input: { uri: string; cid: string; generation: number }, +): Promise { + const row = await db + .prepare( + `SELECT 1 FROM subjects + WHERE uri = ? AND cid = ? AND deleted_at IS NULL AND delete_generation = ?`, + ) + .bind(input.uri, input.cid, input.generation) + .first(); + return row !== null; +} + /** * True when a non-tombstoned subject row exists for this exact `(uri, cid)` — the * same condition the `requireSubjectNotDeleted` issuance guard checks. Distinct @@ -200,6 +256,13 @@ export interface CreateAssessmentRunInput { promptHash?: string; coverageJson: string; now?: Date; + /** + * When set, the run row is created only if the subject `(uri, cid)` is + * undeleted at exactly this captured generation. A delete that advanced the + * generation after the caller verified makes the INSERT match no row (no orphan + * run), the same all-or-nothing guard the issuance uses. + */ + requireSubjectGeneration?: number; } export interface CreateAssessmentRunResult { @@ -219,6 +282,30 @@ export function buildAssessmentRunStatement( input: CreateAssessmentRunInput & { id: string }, ): D1PreparedStatement { const now = input.now ?? new Date(); + const generationGuard = + input.requireSubjectGeneration === undefined + ? "" + : `\n\t\t\t AND EXISTS (SELECT 1 FROM subjects + WHERE uri = ? AND cid = ? AND deleted_at IS NULL AND delete_generation = ?)`; + const binds: unknown[] = [ + input.id, + input.runKey, + input.uri, + input.cid, + input.artifactId ?? null, + input.artifactChecksum ?? null, + input.trigger, + input.triggerId, + input.policyVersion, + input.modelId ?? null, + input.promptHash ?? null, + input.coverageJson, + now.toISOString(), + now.getTime(), + input.runKey, + ]; + if (input.requireSubjectGeneration !== undefined) + binds.push(input.uri, input.cid, input.requireSubjectGeneration); return db .prepare( `INSERT INTO assessments @@ -226,25 +313,9 @@ export function buildAssessmentRunStatement( policy_version, model_id, prompt_hash, coverage_json, created_at, created_at_epoch_ms) SELECT ?, ?, ?, ?, ?, ?, 'observed', ?, ?, ?, ?, ?, ?, ?, ? - WHERE NOT EXISTS (SELECT 1 FROM assessments WHERE run_key = ?)`, + WHERE NOT EXISTS (SELECT 1 FROM assessments WHERE run_key = ?)${generationGuard}`, ) - .bind( - input.id, - input.runKey, - input.uri, - input.cid, - input.artifactId ?? null, - input.artifactChecksum ?? null, - input.trigger, - input.triggerId, - input.policyVersion, - input.modelId ?? null, - input.promptHash ?? null, - input.coverageJson, - now.toISOString(), - now.getTime(), - input.runKey, - ); + .bind(...binds); } /** @@ -316,6 +387,43 @@ export async function listNonTerminalAssessmentsForUri( return (rows.results ?? []).map(rowToAssessment); } +/** States a run can be in and still carry a live, un-negated positive + * `assessment-pending`: every non-terminal state, plus terminal `stale` (which, + * unlike the other terminals, does not negate its own pending on transition). + * `cancelled` is excluded — the delete negates before cancelling; the decision + * outcomes negate their own pending at finalization. */ +const PENDING_BEARING_STATES: readonly AssessmentState[] = [ + "observed", + "verifying", + "pending", + "running", + "stale", +]; + +/** + * Runs for a URI that could still carry a live positive `assessment-pending` + * label — the delete cleanup's scan set (spec §9.1). Widens + * `listNonTerminalAssessmentsForUri` to include terminal `stale` runs: a run that + * self-transitioned to `stale` on detecting a deleted/superseded subject keeps + * its committed positive, so the delete must still reach it to negate. + */ +export async function listPendingBearingAssessmentsForUri( + db: D1Database, + uri: string, +): Promise { + const rows = await db + .prepare( + `SELECT id, run_key, uri, cid, artifact_id, artifact_checksum, state, trigger, trigger_id, + policy_version, model_id, prompt_hash, public_summary, coverage_json, + supersedes_assessment_id, started_at, completed_at, created_at + FROM assessments + WHERE uri = ? AND state IN (${PENDING_BEARING_STATES.map(() => "?").join(", ")})`, + ) + .bind(uri, ...PENDING_BEARING_STATES) + .all(); + return (rows.results ?? []).map(rowToAssessment); +} + export interface TransitionAssessmentInput { id: string; from: AssessmentState; diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index 3103e31e68..b0f55613a4 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -35,6 +35,7 @@ import { buildAssessmentRunStatement, getActiveLabelState, getAssessment, + readDeleteGeneration, type Assessment, } from "./assessment-store.js"; import { buildAutomationPauseUpdate } from "./automation-state.js"; @@ -698,6 +699,15 @@ async function runRerun( const assessment = await loadAssessment(deps.db, id); assertConfirmationCid(ctx.body.confirmation, assessment.cid); + // Capture the subject's tombstone generation before minting the run, so the run + // creation AND the pending issuance below are both gated on no delete having + // advanced it — a concurrent discovery-delete makes both no-op (no orphan run, + // no resurrected positive) rather than leaving a stranded `observed` run. + const generation = await readDeleteGeneration(deps.db, { + uri: assessment.uri, + cid: assessment.cid, + }); + const triggerId = operatorTriggerId(ctx.actionId); const runKey = await rerunRunKey(assessment.uri, assessment.cid, triggerId); const runId = `asmt_${ulid()}`; @@ -715,6 +725,9 @@ async function runRerun( promptHash: RERUN_PROMPT_HASH, coverageJson: "{}", now: ctx.now, + // No orphan `observed` run when a delete removed the subject: the run row + // commits only if the subject is undeleted at the captured generation. + requireSubjectGeneration: generation, }); const signer = await deps.createSigner(); const pending = await prepareAutomatedLabelIssuance( @@ -731,14 +744,15 @@ async function runRerun( { uri: assessment.uri, cid: assessment.cid, val: "assessment-pending" }, ctx.now, // Gate the rerun's positive assessment-pending on the run (created `observed` - // in this batch) and the subject still being undeleted at commit time, so a - // concurrent discovery-delete that tombstoned the subject makes the positive - // no-op instead of resurrecting a live label on a deleted release. On a miss - // the label does not persist -> `assertIssuancePersisted` below aborts before - // `deferRerunTail`, so nothing is published, dispatched, or advanced. + // in this batch) and the subject still being undeleted at the captured + // generation, so a concurrent discovery-delete that tombstoned the subject + // makes the positive no-op instead of resurrecting a live label on a deleted + // release. On a miss the label does not persist -> `assertIssuancePersisted` + // below aborts before `deferRerunTail`, so nothing is published, dispatched, + // or advanced, and the guarded run row above never committed. { requireAssessmentState: "observed", - requireSubjectNotDeleted: { uri: assessment.uri, cid: assessment.cid }, + requireSubjectNotDeleted: { uri: assessment.uri, cid: assessment.cid, generation }, }, ); diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index be04c51e78..f481e26663 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/apps/labeler/src/discovery-consumer.ts @@ -33,6 +33,7 @@ import { } from "@atcute/identity-resolver"; import type { LabelSigner } from "@emdash-cms/registry-moderation"; import { cloudflareDohResolver, type DnsResolver } from "emdash/security/ssrf"; +import { ulid } from "ulidx"; import { AssessmentDispatchError, @@ -46,12 +47,14 @@ import { initialTriggerId, } from "./assessment-lifecycle.js"; import { - createAssessmentRun, + buildAssessmentRunStatement, createSubject, deleteSubjectsByUri, getAssessment, - listNonTerminalAssessmentsForUri, - subjectIsUndeleted, + getAssessmentByRunKey, + listPendingBearingAssessmentsForUri, + readDeleteGeneration, + subjectMatchesGeneration, transitionAssessmentState, type Assessment, } from "./assessment-store.js"; @@ -385,6 +388,13 @@ async function verifyAndCreateRun( deps: DiscoveryConsumerDeps, now: Date, ): Promise { + // Capture the subject's tombstone generation BEFORE verifying. Every commit + // below (subject-undelete, run creation, label issuance) is gated on the + // generation not having advanced past this — so a delete that lands during the + // (slow) verify is seen as a newer generation and the whole create is rejected + // obsolete, while a create that captured the post-delete generation proceeds. + const generation = await readDeleteGeneration(deps.db, { uri, cid: job.cid }); + const verifyFn = deps.verify ?? fetchAndVerifyExactRecord; // Propagates PdsVerificationError / RecordVerificationError untouched — // the caller classifies retry vs dead-letter. @@ -396,6 +406,9 @@ async function verifyAndCreateRun( ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); + // Generation-gated: a re-observation only clears `deleted_at` if no delete has + // advanced the generation since the capture. A stale verify cannot resurrect a + // subject deleted after it read state. await createSubject(deps.db, { uri, cid: job.cid, @@ -403,6 +416,7 @@ async function verifyAndCreateRun( collection: job.collection, rkey: job.rkey, now, + expectedGeneration: generation, }); const triggerId = initialTriggerId(job.cid); @@ -416,7 +430,10 @@ async function verifyAndCreateRun( triggerId, }); - const { assessment } = await createAssessmentRun(deps.db, { + // Generation-gated run creation: no orphan run for a subject the delete removed + // between the capture and here. + await buildAssessmentRunStatement(deps.db, { + id: `asmt_${ulid()}`, runKey, uri, cid: job.cid, @@ -427,17 +444,32 @@ async function verifyAndCreateRun( promptHash: DISCOVERY_PROMPT_HASH, coverageJson: "{}", now, - }); + requireSubjectGeneration: generation, + }).run(); + const assessment = await getAssessmentByRunKey(deps.db, runKey); + if (!assessment) { + // The run insert matched no row: a delete advanced the generation before it + // committed. Obsolete — nothing to advance, issue, or dispatch. + return; + } await advanceToPending(deps.db, assessment, now); - const outcome = await issueInitialPendingLabel(deps, assessment.id, runKey, uri, job.cid, now); + const outcome = await issueInitialPendingLabel( + deps, + assessment.id, + runKey, + uri, + job.cid, + generation, + now, + ); if (outcome === "obsolete") { - // A concurrent delete tombstoned the subject (or cancelled the run) before - // the positive assessment-pending could commit — the run is moot. The - // issuance no-op'd, so there is nothing to publish and nothing to assess; - // the delete owns tombstone + cancel + negation. Do not dispatch a Workflow - // for a label that never committed. + // A concurrent delete tombstoned the subject (advancing the generation) or + // cancelled the run before the positive assessment-pending could commit — the + // run is moot. The issuance no-op'd, so there is nothing to publish and + // nothing to assess; the delete owns tombstone + cancel + negation. Do not + // dispatch a Workflow for a label that never committed. return; } @@ -472,6 +504,7 @@ async function issueInitialPendingLabel( runKey: string, uri: string, cid: string, + generation: number, now: Date, ): Promise<"issued" | "obsolete"> { const idempotencyKey = automatedIdempotencyKey(runKey, "assessment-pending", false); @@ -500,7 +533,10 @@ async function issueInitialPendingLabel( proposal, now, deps.publisher !== undefined, - { requireAssessmentState: "pending", requireSubjectNotDeleted: { uri, cid } }, + { + requireAssessmentState: "pending", + requireSubjectNotDeleted: { uri, cid, generation }, + }, ); await deps.db.batch(statements); @@ -511,10 +547,15 @@ async function issueInitialPendingLabel( } // The guarded insert matched no row. If the run is no longer `pending` or the - // subject was tombstoned, a concurrent delete won — a benign no-op. Anything - // else (the signing guard no-op'ing on a mid-batch pause/rotation) is retryable. + // subject was tombstoned / re-deleted (generation advanced), a concurrent delete + // won — a benign no-op. Anything else (the signing guard no-op'ing on a mid-batch + // pause/rotation) is retryable. const run = await getAssessment(deps.db, assessmentId); - if (!run || run.state !== "pending" || !(await subjectIsUndeleted(deps.db, { uri, cid }))) + if ( + !run || + run.state !== "pending" || + !(await subjectMatchesGeneration(deps.db, { uri, cid, generation })) + ) return "obsolete"; throw new LabelIssuanceUnavailableError("initial pending label did not persist"); } @@ -554,23 +595,23 @@ async function transitionOrObserve( } /** - * Tombstones the subject and retires its non-terminal runs. For each run that - * committed a positive `assessment-pending` label, the negation is issued BEFORE - * the terminal cancellation, so a failed or paused negation (signing - * mid-rotation) leaves the run non-terminal and re-discoverable by - * `listNonTerminalAssessmentsForUri` on redelivery. Cancelling first would drop - * the run from that set, stranding the pending label live forever once the - * message acks. + * Tombstones the subject (advancing its `delete_generation`) and retires every + * run that could still carry a live positive `assessment-pending` — including + * terminal `stale` runs, which self-transition on detecting a deleted subject and + * do NOT negate their own pending. For each such run the negation is issued + * BEFORE any cancellation, so a failed or paused negation (signing mid-rotation) + * leaves a non-terminal run non-terminal and re-discoverable on redelivery; + * cancelling first would drop it from the scan set, stranding the pending live + * once the message acks. * * The negation is keyed on the run having committed a live positive — NOT on its - * lifecycle state — because an operator rerun issues its positive - * `assessment-pending` while the run is still `observed` (the advance to `pending` - * is deferred). Keying on state would let a delete cancel that `observed` run - * without negating its already-live positive. The invariant: a run is cancelled - * only after any positive it committed has been negated, and the message cannot - * ack while a negation is still owed (a throw propagates to the delete handler's - * mutation-phase catch, which always retries), so no active `assessment-pending` - * survives an acked delete. + * lifecycle state — because an operator rerun issues its positive while the run is + * still `observed`, and a stale run keeps its positive after going terminal. + * Cancellation applies only to non-terminal runs (a stale run is already + * terminal). The invariant: a run is cancelled only after any positive it + * committed has been negated, and the message cannot ack while a negation is still + * owed (a throw propagates to the delete handler's mutation-phase catch, which + * always retries), so no active `assessment-pending` survives an acked delete. */ async function applyDiscoveryDelete( deps: DiscoveryConsumerDeps, @@ -578,22 +619,22 @@ async function applyDiscoveryDelete( now: Date, ): Promise { await deleteSubjectsByUri(deps.db, { uri, now }); - const runs = await listNonTerminalAssessmentsForUri(deps.db, uri); + const runs = await listPendingBearingAssessmentsForUri(deps.db, uri); for (const run of runs) { - if ( - run.state !== "observed" && - run.state !== "verifying" && - run.state !== "pending" && - run.state !== "running" - ) - continue; - // Negate before cancelling any run that committed a positive pending, - // regardless of lifecycle state (a rerun's is live while `observed`). + // Negate (before any cancel) any run — non-terminal OR terminal `stale` — + // that committed a positive pending and has not already been negated. const positive = await readIssuedLabelByActionKey( deps.db, automatedIdempotencyKey(run.runKey, "assessment-pending", false), ); - if (positive) await negateRunPendingLabel(deps, run, now); + if (positive) { + const negated = await readIssuedLabelByActionKey( + deps.db, + automatedIdempotencyKey(run.runKey, "assessment-pending", true), + ); + if (!negated) await negateRunPendingLabel(deps, run, now); + } + if (run.state === "stale") continue; // already terminal — negated, nothing to cancel try { await transitionAssessmentState(deps.db, { id: run.id, diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index 57501189be..7c3fb62dde 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -122,14 +122,16 @@ export interface BuildIssuanceOptions { requireAssessmentState?: AssessmentState; /** * Gate the action insert (and thus its label) additionally on the subject - * `(uri, cid)` still being non-tombstoned at commit time. The initial discovery - * issuance pairs this with `requireAssessmentState: "pending"` so a concurrent - * delete that tombstones the subject or cancels the run in the gap before this - * commit makes the positive `assessment-pending` no-op — no live label is - * resurrected for a deleted release. Gating the action (not the label) leaves no - * orphan label, same as the state guard. + * `(uri, cid)` still being non-tombstoned at commit time, at exactly the + * captured `generation`. The initial discovery issuance and the operator rerun + * pair this with `requireAssessmentState` so a concurrent delete that tombstones + * the subject (advancing its `delete_generation`) or cancels the run in the gap + * before this commit makes the positive `assessment-pending` no-op — no live + * label is resurrected for a deleted release, even if the create path itself + * cleared `deleted_at` (a generation bump the create cannot undo). Gating the + * action (not the label) leaves no orphan label, same as the state guard. */ - requireSubjectNotDeleted?: { uri: string; cid: string }; + requireSubjectNotDeleted?: { uri: string; cid: string; generation: number }; } /** @@ -228,7 +230,8 @@ export async function buildIssuanceStatements( const subjectGuardSql = requireSubject === undefined ? "" - : `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects WHERE uri = ? AND cid = ? AND deleted_at IS NULL)`; + : `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects + WHERE uri = ? AND cid = ? AND deleted_at IS NULL AND delete_generation = ?)`; const actionBinds: unknown[] = [ action.actor, action.type, @@ -248,7 +251,8 @@ export async function buildIssuanceStatements( proposal.val, ]; if (requireState !== undefined) actionBinds.push(assessmentId, requireState); - if (requireSubject !== undefined) actionBinds.push(requireSubject.uri, requireSubject.cid); + if (requireSubject !== undefined) + actionBinds.push(requireSubject.uri, requireSubject.cid, requireSubject.generation); const statements: D1PreparedStatement[] = [ db From f09d33ea32c3042d5a31f0e4deca5ddbe8dfb28e Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sun, 19 Jul 2026 07:59:13 +0100 Subject: [PATCH 15/15] test(labeler): barrier tests for the delete-generation close (three seams + republish) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Seam 1: a stale verify (barrier during verify, full delete completes) cannot resurrect the tombstoned subject — createSubject's generation-guarded undelete no-ops, no run, no positive, no dispatch. Seam 2: the delete negates a terminal stale run's stranded positive (widened scan). Seam 3: a rerun after a concurrent tombstone leaves no orphan observed operator run. Plus a delete-then-republish test proving a new revision after a delete assesses cleanly (the generation does not over-block). Each verified failing pre-fix. Refs Sol round-5 (systemic close). --- .../test/console-assessment-mutations.test.ts | 8 + apps/labeler/test/discovery-consumer.test.ts | 162 ++++++++++++++++++ 2 files changed, 170 insertions(+) diff --git a/apps/labeler/test/console-assessment-mutations.test.ts b/apps/labeler/test/console-assessment-mutations.test.ts index e1d1955850..262f39df87 100644 --- a/apps/labeler/test/console-assessment-mutations.test.ts +++ b/apps/labeler/test/console-assessment-mutations.test.ts @@ -408,6 +408,14 @@ describe("rerun", () => { uri, ), ).toBe(0); + // Seam 3: the generation-guarded run creation left NO orphan `observed` + // operator run (reconciliation would ignore an `observed` run forever). + expect( + await countRows( + `SELECT COUNT(*) n FROM assessments WHERE uri = ? AND trigger = 'operator'`, + uri, + ), + ).toBe(0); }); it("a second rerun creates another distinct run", async () => { diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 27d266cea6..0aacd97439 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -23,6 +23,7 @@ import { createSubject, getAssessmentByRunKey, getCurrentAssessment, + transitionAssessmentState, } from "../src/assessment-store.js"; import { buildAutomationPauseUpdate } from "../src/automation-state.js"; import { @@ -1206,3 +1207,164 @@ describe("processDiscoveryMessage: live publication (Sol follow-up)", () => { expect(published).toContain(row.sequence); }); }); + +describe("processDiscoveryMessage: systemic delete-generation (round-5 close)", () => { + const latestPending = (uri: string) => + testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uri) + .first<{ neg: number }>(); + + it("seam 1: a stale verify cannot resurrect a subject a concurrent delete tombstoned", async () => { + const job = await jobFor({ rkey: rkey() }); + const uri = uriFor(job); + // The subject already exists, undeleted, at generation 0 (a prior observation). + await createSubject(testEnv.DB, { + uri, + cid: job.cid, + did: PUBLISHER_DID, + collection: RELEASE_COLLECTION, + rkey: job.rkey, + }); + + const workflow = new FakeAssessmentWorkflow(); + const baseDeps = { ...(await buildDeps()), assessmentWorkflow: workflow }; + + // Barrier: during THIS create's verify (after it captured generation 0), a + // full concurrent delete completes — tombstone + generation bump + ack. + let deleteRan = false; + const barrierVerify = async (): Promise => { + if (!deleteRan) { + deleteRan = true; + await processDiscoveryMessage({ ...job, operation: "delete", cid: "" }, new FakeMessage(), { + ...baseDeps, + confirmDeleted: () => Promise.resolve(true), + }); + } + return verifiedFor(job)(); + }; + + const msg = new FakeMessage(); + await processDiscoveryMessage(job, msg, { ...baseDeps, verify: barrierVerify }); + + expect(msg.acked).toBe(1); + // createSubject's generation-guarded undelete no-op'd (captured gen 0, subject + // now gen 1): the subject stays tombstoned, no run was created, no positive + // issued, nothing dispatched. No resurrection. + const subject = await testEnv.DB.prepare( + `SELECT deleted_at, delete_generation FROM subjects WHERE uri = ? AND cid = ?`, + ) + .bind(uri, job.cid) + .first<{ deleted_at: string | null; delete_generation: number }>(); + expect(subject?.deleted_at).not.toBeNull(); + expect(subject?.delete_generation).toBe(1); + expect(await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job))).toBeNull(); + expect(workflow.created.length).toBe(0); + expect( + await testEnv.DB.prepare( + `SELECT COUNT(*) AS n FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' AND neg = 0`, + ) + .bind(uri) + .first<{ n: number }>(), + ).toEqual({ n: 0 }); + }); + + it("seam 2: the delete negates a stale run's stranded positive", async () => { + const job = await jobFor({ rkey: rkey() }); + const uri = uriFor(job); + await createSubject(testEnv.DB, { + uri, + cid: job.cid, + did: PUBLISHER_DID, + collection: RELEASE_COLLECTION, + rkey: job.rkey, + }); + const runKey = await runKeyFor(job); + const { assessment } = await createAssessmentRun(testEnv.DB, { + runKey, + uri, + cid: job.cid, + trigger: "initial", + triggerId: initialTriggerId(job.cid), + policyVersion: MODERATION_POLICY.policyVersion, + coverageJson: "{}", + }); + await issueAutomatedAssessmentLabel( + testEnv.DB, + config, + await signer(), + { + actor: LABELER_DID, + type: "automated-assessment", + assessmentId: assessment.id, + reason: "initial discovery", + idempotencyKey: automatedIdempotencyKey(runKey, "assessment-pending", false), + }, + { uri, cid: job.cid, val: "assessment-pending" }, + ); + // Drive the run to terminal `stale`, as the orchestrator would on detecting a + // non-current subject — WITHOUT negating its own pending. + for (const [from, to] of [ + ["observed", "verifying"], + ["verifying", "pending"], + ["pending", "running"], + ["running", "stale"], + ] as const) { + await transitionAssessmentState(testEnv.DB, { id: assessment.id, from, to }); + } + expect((await latestPending(uri))?.neg).toBe(0); + + // A delete arrives. The stale run is terminal, so the old non-terminal-only + // scan would miss it; the widened scan reaches it and negates its positive. + await processDiscoveryMessage({ ...job, operation: "delete", cid: "" }, new FakeMessage(), { + ...(await buildDeps()), + confirmDeleted: () => Promise.resolve(true), + }); + + expect((await latestPending(uri))?.neg).toBe(1); + expect((await getAssessmentByRunKey(testEnv.DB, runKey))?.state).toBe("stale"); + }); + + it("delete-then-republish: a new revision after a delete assesses cleanly (generation does not over-block)", async () => { + const job1 = await jobFor({ rkey: rkey() }); + const uri = uriFor(job1); + await processDiscoveryMessage(job1, new FakeMessage(), { + ...(await buildDeps()), + verify: verifiedFor(job1), + }); + await processDiscoveryMessage({ ...job1, operation: "delete", cid: "" }, new FakeMessage(), { + ...(await buildDeps()), + confirmDeleted: () => Promise.resolve(true), + }); + expect((await latestPending(uri))?.neg).toBe(1); + + // A genuine republish: a new revision (same rkey, new cid) discovered AFTER the + // delete. Its subject row is fresh (generation 0), so the capture-and-guard + // admits it — the delete of the old revision must not block the new one. + const job2 = { ...job1, cid: await cid(`${job1.rkey}-v2`) }; + const workflow = new FakeAssessmentWorkflow(); + const msg = new FakeMessage(); + await processDiscoveryMessage(job2, msg, { + ...(await buildDeps()), + assessmentWorkflow: workflow, + verify: verifiedFor(job2), + }); + + expect(msg.acked).toBe(1); + const run = await getAssessmentByRunKey(testEnv.DB, await runKeyFor(job2)); + expect(run?.state).toBe("pending"); + expect(workflow.created.length).toBe(1); + const positive = await testEnv.DB.prepare( + `SELECT l.neg FROM issued_labels l + JOIN issuance_actions a ON a.id = l.action_id + JOIN assessments r ON r.id = a.assessment_id + WHERE r.run_key = ? AND l.val = 'assessment-pending' + ORDER BY l.sequence DESC LIMIT 1`, + ) + .bind(await runKeyFor(job2)) + .first<{ neg: number }>(); + expect(positive?.neg).toBe(0); + }); +});