From 57b442c6b33d6c81a69ec36d69b47c5650e4067f Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Fri, 17 Jul 2026 22:27:58 +0100 Subject: [PATCH] fix(labeler): dispatch the assessment Workflow on operator rerun MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit runRerun minted a fresh assessment run, gated the release with assessment-pending, and advanced the run to pending in its deferred tail, but never dispatched the Cloudflare Workflow that executes it — stranding the release pending forever until the reconciliation stuck-run alert. ConsoleMutationDeps lacked the workflow binding entirely. Thread AssessmentWorkflowBinding into ConsoleMutationDeps (wired from env.ASSESSMENT_WORKFLOW), and dispatch the run in deferRerunTail after advancing to pending, keyed on the run's runKey. The operator trigger yields a fresh runKey, so the rerun gets its own instance; a re-driven tail or replay dedups onto it via the instance-id lock. --- apps/labeler/src/console-mutation-api.ts | 32 +++- apps/labeler/src/index.ts | 1 + .../test/console-assessment-mutations.test.ts | 138 ++++++++++++++++++ .../labeler/test/console-mutation-api.test.ts | 4 + .../test/console-reconsiderations.test.ts | 4 + 5 files changed, 173 insertions(+), 6 deletions(-) diff --git a/apps/labeler/src/console-mutation-api.ts b/apps/labeler/src/console-mutation-api.ts index 088c7b4253..4e869e9f94 100644 --- a/apps/labeler/src/console-mutation-api.ts +++ b/apps/labeler/src/console-mutation-api.ts @@ -21,6 +21,10 @@ import type { OperatorIdentity, OperatorRole, } from "./access-auth.js"; +import { + type AssessmentWorkflowBinding, + dispatchAssessmentWorkflow, +} from "./assessment-dispatch.js"; import { automatedIdempotencyKey, computeRunKey, @@ -162,6 +166,10 @@ export interface ConsoleMutationDeps { * cannot join the D1 batch, so it runs in the deferred tail; the discovery * consumer's `runKey` dedup absorbs a duplicate re-drive. */ sendDiscoveryJob: (job: DiscoveryJob) => Promise; + /** Dispatches a rerun's fresh assessment run to its Workflow instance (the same + * binding discovery uses). The instance id is the run's `runKey`, so a + * redelivered or replayed rerun dedups onto the same instance. */ + assessmentWorkflow: AssessmentWorkflowBinding; /** Publisher-notification deps (plan W10.5). When present, a label-affecting * operator action fires a post-commit publisher notice in the deferred tail * (never blocking or failing the label). Omitted in tests that don't exercise @@ -569,7 +577,19 @@ function deferRerunTail(deps: ConsoleMutationDeps, runId: string, pendingKey: st (async () => { try { const run = await getAssessment(deps.db, runId); - if (run) await advanceAssessmentToPending(deps.db, run, deps.now()); + if (run) { + await advanceAssessmentToPending(deps.db, run, deps.now()); + // The run's runKey is its Workflow instance id, so a re-driven tail + // (defer retry or replay branch) dedups onto the same instance. A + // dispatch failure here is logged and dropped — unlike discovery, + // which retries the queue message, this deferred tail has no retry + // lever, so a dropped dispatch strands the run `pending` until the + // reconciliation stuck-run sweep surfaces it for a manual re-trigger. + await dispatchAssessmentWorkflow(deps.assessmentWorkflow, { + runKey: run.runKey, + assessmentId: runId, + }); + } } catch (error) { console.error("[console-mutation] rerun advance failed", error); } @@ -639,11 +659,11 @@ function deferReconsiderationNotify( * `POST /admin/api/assessments/:id/rerun` — mints the immutable operator trigger * (`operator:`), creates a fresh run for the assessment's exact URI+CID * anchored to that trigger, and re-issues `assessment-pending`, all in one atomic - * batch with the audit row (spec §10/§11.2). Initial discovery now dispatches an - * assessment Workflow after `pending`; this rerun path still stops at `pending` - * (its own Workflow dispatch is a follow-on). The operator trigger yields a - * distinct `runKey`, so the rerun maps to its own Workflow instance id rather - * than colliding with the prior run's — the re-assessment is not stranded. + * batch with the audit row (spec §10/§11.2). The deferred tail then advances the + * fresh run to `pending` and dispatches its assessment Workflow, mirroring + * discovery. The operator trigger yields a distinct `runKey`, so the rerun maps + * to its own Workflow instance id rather than colliding with the prior run's — + * the re-assessment runs rather than stranding `pending`. */ async function runRerun( request: Request, diff --git a/apps/labeler/src/index.ts b/apps/labeler/src/index.ts index 13b31063d6..36cffc7efd 100644 --- a/apps/labeler/src/index.ts +++ b/apps/labeler/src/index.ts @@ -206,6 +206,7 @@ async function handleConsoleApiRequest( sendDiscoveryJob: async (job) => { await env.DISCOVERY_QUEUE.send(job); }, + assessmentWorkflow: env.ASSESSMENT_WORKFLOW, notify: await safeCreateNotifyDeps(env), }); } diff --git a/apps/labeler/test/console-assessment-mutations.test.ts b/apps/labeler/test/console-assessment-mutations.test.ts index f4fb21cb30..6521f0a3ce 100644 --- a/apps/labeler/test/console-assessment-mutations.test.ts +++ b/apps/labeler/test/console-assessment-mutations.test.ts @@ -9,6 +9,10 @@ import { generateKeyPair, SignJWT } from "jose"; import { beforeAll, describe, expect, it } from "vitest"; import type { AccessKeyResolver } from "../src/access-auth.js"; +import type { + AssessmentWorkflowBinding, + AssessmentWorkflowParams, +} from "../src/assessment-dispatch.js"; import { computeRunKey, initialTriggerId } from "../src/assessment-lifecycle.js"; import { createAssessmentRun, @@ -91,6 +95,37 @@ function testSigner() { }); } +interface FakeInstance { + id: string; + params: AssessmentWorkflowParams; +} + +/** In-memory stand-in for the assessment Workflow binding, mirroring the + * discovery consumer test's fake: `create` throws when the id is already taken + * (the run-key instance-id lock), `get` resolves it, and `createError` + * simulates an infrastructure failure. */ +class FakeAssessmentWorkflow implements AssessmentWorkflowBinding { + readonly instances = new Map(); + readonly created: FakeInstance[] = []; + createError: Error | undefined; + + create(options: { id: string; params: AssessmentWorkflowParams }): Promise<{ id: string }> { + if (this.createError) return Promise.reject(this.createError); + if (this.instances.has(options.id)) + return Promise.reject(new Error(`instance ${options.id} already exists`)); + const instance = { id: options.id, params: options.params }; + this.instances.set(options.id, instance); + this.created.push(instance); + return Promise.resolve({ id: options.id }); + } + + get(id: string): Promise<{ id: string }> { + const instance = this.instances.get(id); + if (!instance) return Promise.reject(new Error(`instance ${id} not found`)); + return Promise.resolve({ id }); + } +} + function mutationDeps(overrides: Partial = {}): ConsoleMutationDeps { return { db: testEnv.DB, @@ -109,10 +144,30 @@ function mutationDeps(overrides: Partial = {}): ConsoleMuta void work; }, sendDiscoveryJob: async () => {}, + assessmentWorkflow: new FakeAssessmentWorkflow(), ...overrides, }; } +/** Captures deferred tail work so a test can settle it and observe the rerun's + * Workflow dispatch — the default `mutationDeps` drops deferred work. */ +function captureDeferred(overrides: Partial = {}): { + deps: ConsoleMutationDeps; + workflow: FakeAssessmentWorkflow; + settle: () => Promise; +} { + const workflow = new FakeAssessmentWorkflow(); + const deferred: Promise[] = []; + const deps = mutationDeps({ + assessmentWorkflow: workflow, + defer: (work) => { + deferred.push(work); + }, + ...overrides, + }); + return { deps, workflow, settle: () => Promise.all(deferred.splice(0)) }; +} + function readDeps(overrides: Partial = {}): ConsoleApiDeps { return { db: testEnv.DB, @@ -425,6 +480,89 @@ describe("rerun", () => { ); expect(response.status).toBe(404); }); + + it("dispatches the rerun's fresh run to its own Workflow instance", async () => { + const { id } = await seedRun("rerun-dispatch"); + const { deps, workflow, settle } = captureDeferred(); + const response = await handleConsoleMutation( + post(`/admin/api/assessments/${id}/rerun`, { + confirmation: CID, + reason: "re-assess this release", + idempotencyKey: nextKey(), + }), + deps, + ); + expect(response.status).toBe(200); + const descriptor = await bodyData<{ runId: string }>(response); + + // Dispatch is off the response path; the deferred tail carries it. + expect(workflow.created).toHaveLength(0); + await settle(); + + const run = await getAssessment(testEnv.DB, descriptor.runId); + // The tail advances the fresh run to pending, then hands it to a Workflow + // instance whose id is the run's runKey — distinct from the original run. + expect(run?.state).toBe("pending"); + expect(workflow.created).toHaveLength(1); + expect(workflow.created[0]).toMatchObject({ + id: run!.runKey, + params: { assessmentId: descriptor.runId }, + }); + }); + + it("re-dispatches the same run-key idempotently on replay", async () => { + const { id } = await seedRun("rerun-dispatch-replay"); + const workflow = new FakeAssessmentWorkflow(); + const deferred: Promise[] = []; + const deps = mutationDeps({ + assessmentWorkflow: workflow, + defer: (work) => { + deferred.push(work); + }, + }); + const body = { confirmation: CID, reason: "replay me", idempotencyKey: nextKey() }; + + const first = await handleConsoleMutation( + post(`/admin/api/assessments/${id}/rerun`, body), + deps, + ); + expect(first.status).toBe(200); + const second = await handleConsoleMutation( + post(`/admin/api/assessments/${id}/rerun`, body), + deps, + ); + expect(second.status).toBe(200); + + // Both the proceed tail and the replay tail dispatch the same runKey; the + // instance-id lock dedups the second onto the existing instance. + await Promise.all(deferred.splice(0)); + expect(workflow.created).toHaveLength(1); + }); + + it("keeps the rerun a success when the Workflow dispatch fails in the tail", async () => { + const { id } = await seedRun("rerun-dispatch-fail"); + const { deps, workflow, settle } = captureDeferred(); + workflow.createError = new Error("workflow binding unavailable"); + const response = await handleConsoleMutation( + post(`/admin/api/assessments/${id}/rerun`, { + confirmation: CID, + reason: "re-assess this release", + idempotencyKey: nextKey(), + }), + deps, + ); + // The label committed and the response is a success; a dispatch failure lives + // only in the deferred tail and must not surface or throw. + expect(response.status).toBe(200); + const descriptor = await bodyData<{ runId: string }>(response); + await expect(settle()).resolves.not.toThrow(); + + // The run advanced to pending but nothing dispatched — it is stranded until + // the reconciliation stuck-run sweep surfaces it. + const run = await getAssessment(testEnv.DB, descriptor.runId); + expect(run?.state).toBe("pending"); + expect(workflow.created).toHaveLength(0); + }); }); describe("override", () => { diff --git a/apps/labeler/test/console-mutation-api.test.ts b/apps/labeler/test/console-mutation-api.test.ts index a7ed6d7d28..d194a55d49 100644 --- a/apps/labeler/test/console-mutation-api.test.ts +++ b/apps/labeler/test/console-mutation-api.test.ts @@ -127,6 +127,10 @@ function mutationDeps(overrides: Partial = {}): ConsoleMuta void work; }, sendDiscoveryJob: async () => {}, + assessmentWorkflow: { + create: (options) => Promise.resolve({ id: options.id }), + get: (id) => Promise.resolve({ id }), + }, ...overrides, }; } diff --git a/apps/labeler/test/console-reconsiderations.test.ts b/apps/labeler/test/console-reconsiderations.test.ts index 87c1dd49e5..9b6c671e44 100644 --- a/apps/labeler/test/console-reconsiderations.test.ts +++ b/apps/labeler/test/console-reconsiderations.test.ts @@ -148,6 +148,10 @@ function mutationDeps(overrides: Partial = {}): { deferred.push(work); }, sendDiscoveryJob: async () => {}, + assessmentWorkflow: { + create: (options) => Promise.resolve({ id: options.id }), + get: (id) => Promise.resolve({ id }), + }, ...overrides, }; return { deps, settle: async () => void (await Promise.allSettled(deferred)) };