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/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-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 72f5fff6fa..f577d1f146 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/apps/labeler/src/assessment-orchestrator.ts @@ -31,9 +31,13 @@ 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 { getSigningStatusIfInitialized } from "./signing-rotation.js"; +import type { LabelPublisher } from "./subscribe-labels.js"; /** * A stage's finding is the canonical normalized contract (`findings.ts`). @@ -106,6 +110,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 +130,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 +143,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 { @@ -258,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", @@ -267,10 +286,18 @@ export class AssessmentOrchestrator { cid: assessment.cid, now, ...(coverageJson !== undefined ? { coverageJson } : {}), + signingGuard: { + 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]; - const postCommits: Array<() => Promise> = []; + const postCommits: Array<() => Promise> = []; const issue = async ( val: string, @@ -298,7 +325,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 }, @@ -347,11 +378,20 @@ 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. + // + // 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 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); @@ -382,12 +422,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-store.ts b/apps/labeler/src/assessment-store.ts index e821b32972..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,71 @@ 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 + * 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 @@ -182,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 { @@ -201,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 @@ -208,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); } /** @@ -298,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; @@ -474,6 +600,27 @@ 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 }; + /** + * 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 { @@ -502,6 +649,39 @@ 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); + } + // 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( @@ -510,18 +690,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}${subjectGuardSql}`, ) - .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/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/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index 4e869e9f94..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"; @@ -291,6 +292,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); } @@ -693,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()}`; @@ -710,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( @@ -725,6 +743,17 @@ 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 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, generation }, + }, ); const descriptor: RerunDescriptor = { @@ -1040,6 +1069,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/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index 18395e7b7b..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,11 +47,14 @@ import { initialTriggerId, } from "./assessment-lifecycle.js"; import { - createAssessmentRun, + buildAssessmentRunStatement, createSubject, deleteSubjectsByUri, getAssessment, - listNonTerminalAssessmentsForUri, + getAssessmentByRunKey, + listPendingBearingAssessmentsForUri, + readDeleteGeneration, + subjectMatchesGeneration, transitionAssessmentState, type Assessment, } from "./assessment-store.js"; @@ -71,8 +75,16 @@ 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"; /** * Stub identifiers for the model/prompt/scanner-set components of the run @@ -94,6 +106,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; /** Resolves each PDS hop's hostname for the SSRF egress guard; defaults to * the DoH resolver used by artifact acquisition. */ @@ -214,34 +233,49 @@ 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 } : {}), ...(deps.resolveHostname ? { resolveHostname: deps.resolveHostname } : {}), }); - if (!absent) { - await writeDeadLetter( - deps.db, - job, - "DELETE_RECORD_PRESENT", - "record still resolves", - now(), - ); - controller.ack(); - return; - } - const cancelled = await applyDiscoveryDelete(deps.db, uri, now()); - await negatePendingForDeletedRuns(deps, cancelled, now()); - controller.ack(); } 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) { + // 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; } @@ -354,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. @@ -365,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, @@ -372,6 +416,7 @@ async function verifyAndCreateRun( collection: job.collection, rkey: job.rkey, now, + expectedGeneration: generation, }); const triggerId = initialTriggerId(job.cid); @@ -385,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, @@ -396,24 +444,34 @@ 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); - 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" }, + const outcome = await issueInitialPendingLabel( + deps, + assessment.id, + runKey, + uri, + job.cid, + generation, now, ); + if (outcome === "obsolete") { + // 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; + } // 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 @@ -426,6 +484,82 @@ 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, + generation: number, + 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, generation }, + }, + ); + 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 / 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 subjectMatchesGeneration(deps.db, { uri, cid, generation })) + ) + 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 @@ -460,24 +594,54 @@ 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 (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 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, + uri: string, + now: Date, +): Promise { + await deleteSubjectsByUri(deps.db, { uri, now }); + 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; - if (run.state === "pending" || run.state === "running") hadPending.push(run); + // 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) { + 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(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 @@ -485,37 +649,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, - ); - } + 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 { @@ -572,6 +735,7 @@ async function createProductionDiscoveryDeps(env: Env): Promise 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/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/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..7c3fb62dde 100644 --- a/apps/labeler/src/service.ts +++ b/apps/labeler/src/service.ts @@ -120,6 +120,18 @@ 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, 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; generation: number }; } /** @@ -214,6 +226,12 @@ 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 AND delete_generation = ?)`; const actionBinds: unknown[] = [ action.actor, action.type, @@ -233,6 +251,8 @@ export async function buildIssuanceStatements( proposal.val, ]; if (requireState !== undefined) actionBinds.push(assessmentId, requireState); + if (requireSubject !== undefined) + actionBinds.push(requireSubject.uri, requireSubject.cid, requireSubject.generation); const statements: D1PreparedStatement[] = [ db @@ -257,7 +277,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), @@ -447,13 +467,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 { @@ -599,7 +620,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 0b71af5c8c..9e71aecf29 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"; @@ -43,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), }; } @@ -208,15 +219,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/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index ae9d1b7245..f4c2634590 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/apps/labeler/test/assessment-orchestrator.test.ts @@ -33,8 +33,13 @@ 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 { initializeSigningState } from "../src/signing-rotation.js"; +import { issueManualLabel, type IssuedLabel } from "../src/service.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"; interface TestEnv { @@ -47,8 +52,48 @@ 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. */ +function recordingPublisher(managesPublicationState: boolean): { + publisher: LabelPublisher; + published: IssuedLabel[]; +} { + const published: IssuedLabel[] = []; + return { + published, + publisher: { + managesPublicationState, + async publish(issued) { + published.push(issued); + }, + }, + }; +} + +/** 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); await initializeSigningState(testEnv.DB, { @@ -1119,3 +1164,226 @@ 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); + }); +}); + +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); + }); +}); + +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); + }); +}); diff --git a/apps/labeler/test/console-assessment-mutations.test.ts b/apps/labeler/test/console-assessment-mutations.test.ts index 6521f0a3ce..262f39df87 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,48 @@ 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); + // 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 () => { const { id } = await seedRun("rerun-twice"); const first = await bodyData<{ runId: string }>( diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index e98706172d..4f8f0ce98c 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(); diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 5325c4d221..0aacd97439 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -16,10 +16,18 @@ import { automatedIdempotencyKey, computeRunKey, initialTriggerId, + operatorTriggerId, } from "../src/assessment-lifecycle.js"; -import { getAssessmentByRunKey, getCurrentAssessment } from "../src/assessment-store.js"; +import { + createAssessmentRun, + createSubject, + getAssessmentByRunKey, + getCurrentAssessment, + transitionAssessmentState, +} from "../src/assessment-store.js"; import { buildAutomationPauseUpdate } from "../src/automation-state.js"; import { + bestEffortPublisher, type DiscoveryConsumerDeps, type MessageController, processDiscoveryMessage, @@ -27,12 +35,14 @@ import { import type { DiscoveryJob } from "../src/env.js"; import { PdsVerificationError, type VerifiedPdsRecord } from "../src/pds-verify.js"; import { MODERATION_POLICY } from "../src/policy.js"; +import { sweepPendingPublications } from "../src/reconciliation.js"; 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"; interface TestEnv { DB: D1Database; @@ -558,6 +568,155 @@ 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); + + // 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 === 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); + }); +}); + describe("processDiscoveryMessage: delete", () => { it("tombstones the subject and cancels non-terminal runs", async () => { const job = await jobFor({ rkey: rkey() }); @@ -631,6 +790,116 @@ 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("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(); @@ -838,3 +1107,264 @@ describe("processDiscoveryMessage: automation kill-switch", () => { 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); + }); +}); + +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); + }); +}); diff --git a/apps/labeler/test/reconciliation.test.ts b/apps/labeler/test/reconciliation.test.ts index 37bef9b4e1..ab6ba52599 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,109 @@ 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); + }); + + 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"); + }); +}); 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"); + }); +});