From 5591d797fa2bf44d5c6bfd27041863287371adf8 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 15 Sep 2026 10:10:32 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8A=A0=E5=9B=BA=20Outbox=20=E7=A7=9F?= =?UTF-8?q?=E7=BA=A6=E4=B8=8E=20Worker=20=E7=94=9F=E5=91=BD=E5=91=A8?= =?UTF-8?q?=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/package.json | 2 +- apps/worker/src/main.spec.ts | 81 ++++++ apps/worker/src/main.ts | 98 ++++++-- apps/worker/src/moderation-worker.spec.ts | 154 +++++++++++- apps/worker/src/moderation-worker.ts | 235 ++++++++++++++---- .../0007_outbox_lease_ownership/migration.sql | 7 + prisma/schema.prisma | 5 +- 7 files changed, 502 insertions(+), 80 deletions(-) create mode 100644 apps/worker/src/main.spec.ts create mode 100644 prisma/migrations/0007_outbox_lease_ownership/migration.sql diff --git a/apps/worker/package.json b/apps/worker/package.json index 125f589..3e7cd8e 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -7,7 +7,7 @@ "build": "tsc -p tsconfig.build.json", "start": "node dist/main.js", "run:once": "tsx src/main.ts --once", - "test": "vitest run src/moderation-worker.spec.ts --no-file-parallelism", + "test": "vitest run --no-file-parallelism", "typecheck": "tsc --noEmit -p tsconfig.json" }, "dependencies": { diff --git a/apps/worker/src/main.spec.ts b/apps/worker/src/main.spec.ts new file mode 100644 index 0000000..7dc6595 --- /dev/null +++ b/apps/worker/src/main.spec.ts @@ -0,0 +1,81 @@ +import { describe, expect, it, vi } from "vitest"; +import { handleTopLevelError, logTopLevelError, runWorker } from "./main.js"; + +describe("worker lifecycle", () => { + it("waits for the current run before disconnecting on SIGTERM", async () => { + let finish!: () => void; + const runOnce = vi.fn( + () => + new Promise((resolve) => { + finish = () => resolve(true); + }), + ); + const disconnect = vi.fn(() => Promise.resolve()); + let handler: (() => void) | undefined; + const signals = { + once: vi.fn((_signal: string, callback: () => void) => { + handler = callback; + }), + removeListener: vi.fn(), + }; + const running = runWorker({ + worker: { runOnce }, + connect: vi.fn(() => Promise.resolve()), + disconnect, + once: false, + sleep: vi.fn(() => Promise.resolve()), + signals, + }); + await vi.waitFor(() => expect(runOnce).toHaveBeenCalledTimes(1)); + handler?.(); + expect(disconnect).not.toHaveBeenCalled(); + finish(); + await running; + expect(runOnce).toHaveBeenCalledTimes(1); + expect(disconnect).toHaveBeenCalledTimes(1); + expect(signals.once).toHaveBeenCalledTimes(2); + expect(signals.removeListener).toHaveBeenCalledTimes(2); + }); + + it("disconnects in once mode when handling throws", async () => { + const failure = new Error("boom"); + const disconnect = vi.fn(() => Promise.resolve()); + await expect( + runWorker({ + worker: { + runOnce: vi.fn(() => Promise.reject(failure)), + }, + connect: vi.fn(() => Promise.resolve()), + disconnect, + once: true, + }), + ).rejects.toBe(failure); + expect(disconnect).toHaveBeenCalledTimes(1); + }); + + it("logs top-level errors without leaking their message", () => { + const logger = { error: vi.fn() }; + + logTopLevelError(new Error("secret payload"), logger); + + expect(logger.error).toHaveBeenCalledTimes(1); + const output = String(logger.error.mock.calls[0]?.[0]); + expect(output).toContain('"scope":"moderation-worker"'); + expect(output).toContain('"errorClass":"Error"'); + expect(output).not.toContain("secret payload"); + expect(output).not.toContain("message"); + }); + + it("marks the process as failed after a top-level error", () => { + const logger = { error: vi.fn() }; + const processState: { exitCode?: number } = {}; + + handleTopLevelError(new Error("secret payload"), { + logger, + processState, + }); + + expect(processState.exitCode).toBe(1); + expect(logger.error).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index 9d6de84..666c8f8 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -1,23 +1,91 @@ +import { pathToFileURL } from "node:url"; import { PrismaClient } from "@prisma/client"; import { ModerationWorker } from "./moderation-worker.js"; -const prisma = new PrismaClient(); +type SignalSource = { + once(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown; + removeListener(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown; +}; -async function main() { - await prisma.$connect(); - const worker = new ModerationWorker(prisma); - if (process.argv.includes("--once")) { - await worker.runOnce(); - return; +type RunWorkerOptions = { + worker: { runOnce(): Promise }; + connect: () => Promise; + disconnect: () => Promise; + once: boolean; + sleep?: () => Promise; + signals?: SignalSource; +}; + +export async function runWorker(options: RunWorkerOptions) { + const signals = options.signals ?? process; + const sleep = + options.sleep ?? + (() => new Promise((resolve) => setTimeout(resolve, 1000))); + let stopping = false; + const stop = () => { + stopping = true; + }; + if (!options.once) { + signals.once("SIGTERM", stop); + signals.once("SIGINT", stop); } - for (;;) { - const handled = await worker.runOnce(); - if (!handled) await new Promise((resolve) => setTimeout(resolve, 1000)); + try { + await options.connect(); + do { + const handled = await options.worker.runOnce(); + if (options.once || stopping) break; + if (!handled) await sleep(); + } while (!stopping); + } finally { + if (!options.once) { + signals.removeListener("SIGTERM", stop); + signals.removeListener("SIGINT", stop); + } + await options.disconnect(); } } -main() - .catch(() => { - process.exitCode = 1; - }) - .finally(() => prisma.$disconnect()); +export async function main() { + const prisma = new PrismaClient(); + const worker = new ModerationWorker(prisma); + await runWorker({ + worker, + connect: () => prisma.$connect(), + disconnect: () => prisma.$disconnect(), + once: process.argv.includes("--once"), + }); +} + +export function logTopLevelError( + error: unknown, + logger: Pick = console, +) { + logger.error( + JSON.stringify({ + scope: "moderation-worker", + errorClass: + error instanceof Error && error.name ? error.name : "UnknownError", + }), + ); +} + +type TopLevelErrorOptions = { + logger?: Pick; + processState?: { exitCode?: number }; +}; + +export function handleTopLevelError( + error: unknown, + options: TopLevelErrorOptions = {}, +) { + logTopLevelError(error, options.logger); + (options.processState ?? process).exitCode = 1; +} + +const isEntrypoint = + process.argv[1] !== undefined && + import.meta.url === pathToFileURL(process.argv[1]).href; + +if (isEntrypoint) { + main().catch(handleTopLevelError); +} diff --git a/apps/worker/src/moderation-worker.spec.ts b/apps/worker/src/moderation-worker.spec.ts index e6325c1..9367336 100644 --- a/apps/worker/src/moderation-worker.spec.ts +++ b/apps/worker/src/moderation-worker.spec.ts @@ -1,5 +1,5 @@ import { PrismaClient } from "@prisma/client"; -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; import { assertSafeTestDatabaseUrl } from "../../../prisma/database-safety.js"; import { ModerationWorker, decideModeration } from "./moderation-worker.js"; @@ -133,8 +133,104 @@ describe("moderation worker with real PostgreSQL", () => { ).toBe("PUBLISHED"); }); - it("publishes a replay when the matching bottle task is already completed", async () => { + it("does not let an expired lease report failure after a new owner publishes", async () => { + const { bottle, event } = await fixture("ordinary"); + let rejectFirst!: (reason: Error) => void; + const firstDecision = new Promise>( + (_resolve, reject) => { + rejectFirst = reject; + }, + ); + const firstRun = new ModerationWorker( + prisma, + () => firstDecision, + 1000, + ).runOnce(); + await waitForProcessing(event.id); + await prisma.outboxEvent.update({ + where: { id: event.id }, + data: { lockedAt: new Date(Date.now() - 60_000) }, + }); + await new ModerationWorker(prisma, undefined, 1000).runOnce(); + rejectFirst(new Error("secret-body-provider-error")); + await firstRun; + expect( + await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }), + ).toMatchObject({ status: "PUBLISHED" }); + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), + ).toMatchObject({ reviewStatus: "APPROVED", poolStatus: "IN_POOL" }); + }); + + it("does not let an expired lease publish or overwrite a new owner's decision", async () => { const { bottle, event, task } = await fixture("ordinary"); + let resolveFirst!: (decision: ReturnType) => void; + const firstDecision = new Promise>( + (resolve) => { + resolveFirst = resolve; + }, + ); + const firstRun = new ModerationWorker( + prisma, + () => firstDecision, + 1000, + ).runOnce(); + await waitForProcessing(event.id); + await prisma.outboxEvent.update({ + where: { id: event.id }, + data: { lockedAt: new Date(Date.now() - 60_000) }, + }); + await new ModerationWorker( + prisma, + () => ({ + reviewStatus: "APPROVED", + poolStatus: "IN_POOL", + labels: ["NEW_OWNER"], + }), + 1000, + ).runOnce(); + resolveFirst({ + reviewStatus: "REJECTED", + poolStatus: "CLOSED", + labels: ["OLD_OWNER"], + }); + await firstRun; + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), + ).toMatchObject({ reviewStatus: "APPROVED", poolStatus: "IN_POOL" }); + expect( + await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }), + ).toMatchObject({ decision: "APPROVED", riskLabels: ["NEW_OWNER"] }); + }); + + it("publishes a replay when the completed bottle task matches current state", async () => { + const { bottle, event, task } = await fixture("ordinary"); + await prisma.$transaction([ + prisma.moderationTask.update({ + where: { id: task.id }, + data: { + status: "COMPLETED", + decision: "APPROVED", + result: { decision: "APPROVED" }, + reviewedAt: new Date(), + }, + }), + prisma.bottle.update({ + where: { id: bottle.id }, + data: { reviewStatus: "APPROVED", poolStatus: "IN_POOL" }, + }), + ]); + const decide = vi.fn(decideModeration); + + expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true); + expect(decide).not.toHaveBeenCalled(); + expect( + await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }), + ).toMatchObject({ status: "PUBLISHED" }); + }); + + it("fails a completed bottle event whose result drifted from current state", async () => { + const { event, task } = await fixture("secret bottle text"); await prisma.moderationTask.update({ where: { id: task.id }, data: { @@ -144,16 +240,14 @@ describe("moderation worker with real PostgreSQL", () => { reviewedAt: new Date(), }, }); - const decide = vi.fn(decideModeration); - - expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true); - expect(decide).not.toHaveBeenCalled(); + const logs: unknown[] = []; + await new ModerationWorker(prisma, undefined, undefined, { + error: (entry) => logs.push(entry), + }).runOnce(); expect( await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }), - ).toMatchObject({ status: "PUBLISHED" }); - expect( - await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), - ).toMatchObject({ reviewStatus: "REVIEWING", poolStatus: "CLOSED" }); + ).toMatchObject({ status: "FAILED" }); + expect(JSON.stringify(logs)).not.toContain("secret bottle text"); }); it("fails the event and leaves the task open when bottle state drift prevents the update", async () => { @@ -266,7 +360,7 @@ describe("moderation worker with real PostgreSQL", () => { ).toMatchObject({ status: "PENDING", decision: null }); }); - it("publishes a replay when the matching profile task is already completed", async () => { + it("fails a replay when the completed profile task drifted from current state", async () => { const { profile, event, task } = await profileFixture("ordinary", "first"); await prisma.moderationTask.update({ where: { id: task.id }, @@ -282,7 +376,7 @@ describe("moderation worker with real PostgreSQL", () => { expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true); expect( await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }), - ).toMatchObject({ status: "PUBLISHED" }); + ).toMatchObject({ status: "FAILED" }); expect( await prisma.anonymousProfile.findUniqueOrThrow({ where: { id: profile.id }, @@ -293,6 +387,29 @@ describe("moderation worker with real PostgreSQL", () => { ).toMatchObject({ status: "COMPLETED", decision: "APPROVED" }); }); + it("logs a safe error category without payload or exception message", async () => { + const { event } = await fixture("secret正文"); + const entries: unknown[] = []; + await new ModerationWorker( + prisma, + () => { + throw new TypeError("provider leaked secret正文"); + }, + undefined, + { error: (entry) => entries.push(entry) }, + ).runOnce(); + expect(entries).toEqual([ + { + eventId: event.id, + eventType: "BOTTLE_MODERATION_REQUESTED", + attempt: 1, + errorClass: "TypeError", + }, + ]); + expect(JSON.stringify(entries)).not.toContain("secret正文"); + expect(JSON.stringify(entries)).not.toContain("provider leaked"); + }); + it("publishes an old profile event without letting it overwrite a newer version", async () => { const { createHash } = await import("node:crypto"); const { profile, event, task } = await profileFixture("ordinary", "first"); @@ -375,7 +492,7 @@ describe("moderation worker with real PostgreSQL", () => { targetType: "BOTTLE", targetId: bottle.id, provider: "SIMULATED", - payloadHash: "a".repeat(64), + payloadHash: createHash("sha256").update(contentText).digest("hex"), riskLabels: [], }, }); @@ -430,4 +547,15 @@ describe("moderation worker with real PostgreSQL", () => { }); return { profile, task, event }; } + + async function waitForProcessing(id: string) { + for (let attempt = 0; attempt < 100; attempt += 1) { + const event = await prisma.outboxEvent.findUniqueOrThrow({ + where: { id }, + }); + if (event.status === "PROCESSING") return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error("worker did not claim event"); + } }); diff --git a/apps/worker/src/moderation-worker.ts b/apps/worker/src/moderation-worker.ts index 21ec020..35a4459 100644 --- a/apps/worker/src/moderation-worker.ts +++ b/apps/worker/src/moderation-worker.ts @@ -1,4 +1,4 @@ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { Prisma, PrismaClient, @@ -18,8 +18,28 @@ type Claimed = { eventType: string; dedupeKey: string; payload: Prisma.JsonValue; + attempts: number; + lockToken: string; }; +export type WorkerLogEntry = { + eventId: string; + eventType: string; + attempt: number; + errorClass: string; +}; +export type WorkerLogger = { error(entry: WorkerLogEntry): void }; + const NEVER = new Date("9999-12-31T23:59:59.999Z"); +const defaultLogger: WorkerLogger = { + error: (entry) => console.error(JSON.stringify(entry)), +}; + +class LeaseLostError extends Error { + constructor() { + super("outbox lease lost"); + this.name = "LeaseLostError"; + } +} function positiveInt( value: string | undefined, @@ -32,6 +52,19 @@ function positiveInt( return parsed; } +function resultDecision(result: Prisma.JsonValue | null): string | null { + if (typeof result !== "object" || result === null || Array.isArray(result)) + return null; + const decision = (result as Record).decision; + return typeof decision === "string" ? decision : null; +} + +function expectedPoolStatus(decision: string): BottlePoolStatus | null { + if (decision === "APPROVED") return "IN_POOL"; + if (decision === "REJECTED" || decision === "MANUAL_REVIEW") return "CLOSED"; + return null; +} + export function decideModeration(text: string): Decision { const rejectWords = (process.env.MODERATION_REJECT_WORDS ?? "reject-word") .split(",") @@ -65,6 +98,7 @@ export class ModerationWorker { 30_000, "OUTBOX_LEASE_MS", ), + private readonly logger: WorkerLogger = defaultLogger, ) { this.maxAttempts = positiveInt( process.env.OUTBOX_MAX_ATTEMPTS, @@ -82,20 +116,31 @@ export class ModerationWorker { else if (event.eventType === "PROFILE_MODERATION_REQUESTED") await this.handleProfile(event); else throw new Error("unsupported moderation event"); - } catch { - const current = await this.prisma.outboxEvent.findUniqueOrThrow({ - where: { id: event.id }, - }); - const exhausted = current.attempts >= this.maxAttempts; + } catch (error) { + if (!(error instanceof LeaseLostError)) { + this.logger.error({ + eventId: event.id, + eventType: event.eventType, + attempt: event.attempts, + errorClass: + error instanceof Error && error.name ? error.name : "UnknownError", + }); + } + const exhausted = event.attempts >= this.maxAttempts; const delayMs = Math.min( 60_000, - 1000 * 2 ** Math.max(0, current.attempts - 1), + 1000 * 2 ** Math.max(0, event.attempts - 1), ); - await this.prisma.outboxEvent.update({ - where: { id: event.id }, + await this.prisma.outboxEvent.updateMany({ + where: { + id: event.id, + status: "PROCESSING", + lockToken: event.lockToken, + }, data: { status: "FAILED", lockedAt: null, + lockToken: null, nextRetryAt: exhausted ? NEVER : new Date(Date.now() + delayMs), }, }); @@ -119,37 +164,73 @@ export class ModerationWorker { ]); if (task.targetType !== "BOTTLE" || task.targetId !== bottle.id) throw new Error("mismatched moderation task"); + const hash = createHash("sha256").update(bottle.contentText).digest("hex"); if (task.status === "COMPLETED") { - await this.prisma.$transaction((tx) => this.publish(tx, event.id)); + if (!this.completedBottleMatches(bottle, task, hash)) + throw new Error("completed bottle moderation state drift"); + await this.prisma.$transaction((tx) => this.publish(tx, event)); return; } const decision = await this.decide(bottle.contentText); await this.prisma.$transaction(async (tx) => { - const currentTask = await tx.moderationTask.findUniqueOrThrow({ - where: { id: task.id }, - }); - if (currentTask.status !== "COMPLETED") { - const updated = await tx.bottle.updateMany({ - where: { - id: bottle.id, - reviewStatus: "REVIEWING", - poolStatus: "CLOSED", - }, - data: { - reviewStatus: decision.reviewStatus, - poolStatus: decision.poolStatus, - approvedAt: - decision.reviewStatus === "APPROVED" ? new Date() : null, - }, - }); - if (updated.count !== 1) - throw new Error("bottle moderation state drift"); - await this.completeTask(tx, task.id, decision); + await this.publish(tx, event); + const [currentBottle, currentTask] = await Promise.all([ + tx.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), + tx.moderationTask.findUniqueOrThrow({ where: { id: task.id } }), + ]); + const currentHash = createHash("sha256") + .update(currentBottle.contentText) + .digest("hex"); + if (currentTask.status === "COMPLETED") { + if ( + !this.completedBottleMatches(currentBottle, currentTask, currentHash) + ) + throw new Error("completed bottle moderation state drift"); + return; } - await this.publish(tx, event.id); + if (currentTask.payloadHash !== currentHash) + throw new Error("bottle moderation payload drift"); + const updated = await tx.bottle.updateMany({ + where: { + id: bottle.id, + reviewStatus: "REVIEWING", + poolStatus: "CLOSED", + }, + data: { + reviewStatus: decision.reviewStatus, + poolStatus: decision.poolStatus, + approvedAt: decision.reviewStatus === "APPROVED" ? new Date() : null, + }, + }); + if (updated.count !== 1) throw new Error("bottle moderation state drift"); + await this.completeTask(tx, task.id, decision); }); } + private completedBottleMatches( + bottle: { + contentText: string; + reviewStatus: ReviewStatus; + poolStatus: BottlePoolStatus; + }, + task: { + payloadHash: string; + decision: string | null; + result: Prisma.JsonValue | null; + }, + hash: string, + ) { + const poolStatus = task.decision && expectedPoolStatus(task.decision); + return ( + task.payloadHash === hash && + task.decision !== null && + resultDecision(task.result) === task.decision && + bottle.reviewStatus === task.decision && + poolStatus !== null && + bottle.poolStatus === poolStatus + ); + } + private async handleProfile(event: Claimed) { const payload = event.payload as { profileId?: unknown; @@ -180,6 +261,8 @@ export class ModerationWorker { const version = Number( event.dedupeKey.slice(event.dedupeKey.lastIndexOf(":") + 1), ); + if (!Number.isSafeInteger(version) || version <= 0) + throw new Error("invalid profile moderation version"); const text = JSON.stringify([ profile.nickname, profile.avatarColor, @@ -187,28 +270,47 @@ export class ModerationWorker { ]); const hash = createHash("sha256").update(text).digest("hex"); if (profile.version !== version || task.payloadHash !== hash) { - await this.prisma.outboxEvent.update({ - where: { id: event.id }, - data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null }, - }); + await this.prisma.$transaction((tx) => this.publish(tx, event)); return; } if (task.status === "COMPLETED") { - await this.prisma.$transaction((tx) => this.publish(tx, event.id)); + if (!this.completedProfileMatches(profile, task, hash, version)) + throw new Error("completed profile moderation state drift"); + await this.prisma.$transaction((tx) => this.publish(tx, event)); return; } const decision = await this.decide(text); await this.prisma.$transaction(async (tx) => { + await this.publish(tx, event); const [currentProfile, currentTask] = await Promise.all([ tx.anonymousProfile.findUniqueOrThrow({ where: { id: profile.id } }), tx.moderationTask.findUniqueOrThrow({ where: { id: task.id } }), ]); + const currentText = JSON.stringify([ + currentProfile.nickname, + currentProfile.avatarColor, + currentProfile.bio, + ]); + const currentHash = createHash("sha256") + .update(currentText) + .digest("hex"); if ( currentProfile.version !== version || - currentTask.payloadHash !== hash || - currentTask.status === "COMPLETED" - ) { - await this.publish(tx, event.id); + currentTask.payloadHash !== hash + ) + return; + if (currentHash !== hash) + throw new Error("profile moderation payload drift"); + if (currentTask.status === "COMPLETED") { + if ( + !this.completedProfileMatches( + currentProfile, + currentTask, + hash, + version, + ) + ) + throw new Error("completed profile moderation state drift"); return; } const updated = await tx.anonymousProfile.updateMany({ @@ -218,10 +320,28 @@ export class ModerationWorker { if (updated.count !== 1) throw new Error("profile moderation state drift"); await this.completeTask(tx, task.id, decision); - await this.publish(tx, event.id); }); } + private completedProfileMatches( + profile: { version: number; reviewStatus: ReviewStatus }, + task: { + payloadHash: string; + decision: string | null; + result: Prisma.JsonValue | null; + }, + hash: string, + version: number, + ) { + return ( + profile.version === version && + task.payloadHash === hash && + task.decision !== null && + resultDecision(task.result) === task.decision && + profile.reviewStatus === task.decision + ); + } + private completeTask( tx: Prisma.TransactionClient, id: string, @@ -239,11 +359,21 @@ export class ModerationWorker { }); } - private publish(tx: Prisma.TransactionClient, id: string) { - return tx.outboxEvent.update({ - where: { id }, - data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null }, + private async publish(tx: Prisma.TransactionClient, event: Claimed) { + const published = await tx.outboxEvent.updateMany({ + where: { + id: event.id, + status: "PROCESSING", + lockToken: event.lockToken, + }, + data: { + status: "PUBLISHED", + publishedAt: new Date(), + lockedAt: null, + lockToken: null, + }, }); + if (published.count !== 1) throw new LeaseLostError(); } private async claim(): Promise { @@ -255,11 +385,16 @@ export class ModerationWorker { lockedAt: { lt: staleBefore }, attempts: { gte: this.maxAttempts }, }, - data: { status: "FAILED", lockedAt: null, nextRetryAt: NEVER }, + data: { + status: "FAILED", + lockedAt: null, + lockToken: null, + nextRetryAt: NEVER, + }, }); - const rows = await tx.$queryRaw` + const rows = await tx.$queryRaw[]>` SELECT "id", "aggregate_id" AS "aggregateId", "event_type" AS "eventType", - "dedupe_key" AS "dedupeKey", "payload" + "dedupe_key" AS "dedupeKey", "payload", "attempts" FROM "outbox_events" WHERE "event_type" IN ('BOTTLE_MODERATION_REQUESTED', 'PROFILE_MODERATION_REQUESTED') AND "attempts" < ${this.maxAttempts} @@ -269,15 +404,17 @@ export class ModerationWorker { FOR UPDATE SKIP LOCKED LIMIT 1`; const row = rows[0]; if (!row) return null; + const lockToken = randomUUID(); await tx.outboxEvent.update({ where: { id: row.id }, data: { status: "PROCESSING", lockedAt: new Date(), + lockToken, attempts: { increment: 1 }, }, }); - return row; + return { ...row, attempts: row.attempts + 1, lockToken }; }); } } diff --git a/prisma/migrations/0007_outbox_lease_ownership/migration.sql b/prisma/migrations/0007_outbox_lease_ownership/migration.sql new file mode 100644 index 0000000..90c51c7 --- /dev/null +++ b/prisma/migrations/0007_outbox_lease_ownership/migration.sql @@ -0,0 +1,7 @@ +DROP INDEX IF EXISTS "outbox_events_processing_lease_idx"; + +ALTER TABLE "outbox_events" + ADD COLUMN "lock_token" UUID; + +CREATE INDEX "outbox_events_processing_lease_idx" + ON "outbox_events"("status", "locked_at", "lock_token"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 2a98fcc..170797c 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -428,7 +428,7 @@ model Notification { } model OutboxEvent { - /// Processing lease index outbox_events_processing_lease_idx is managed in 0005 SQL. + /// Processing lease index outbox_events_processing_lease_idx is managed in 0007 SQL. id String @id @default(uuid()) @db.Uuid aggregateType String @map("aggregate_type") @db.VarChar(100) aggregateId String @map("aggregate_id") @db.Uuid @@ -439,12 +439,13 @@ model OutboxEvent { attempts Int @default(0) nextRetryAt DateTime @default(now()) @map("next_retry_at") @db.Timestamptz(3) lockedAt DateTime? @map("locked_at") @db.Timestamptz(3) + lockToken String? @map("lock_token") @db.Uuid publishedAt DateTime? @map("published_at") @db.Timestamptz(3) createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3) @@index([status, nextRetryAt]) - @@index([status, lockedAt], map: "outbox_events_processing_lease_idx") + @@index([status, lockedAt, lockToken], map: "outbox_events_processing_lease_idx") @@index([aggregateType, aggregateId]) @@map("outbox_events") }