fix: 加固 Outbox 租约与 Worker 生命周期

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": {

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<boolean>((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);
+  });
+});

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";
 
+type SignalSource = {
+  once(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown;
+  removeListener(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown;
+};
+
+type RunWorkerOptions = {
+  worker: { runOnce(): Promise<boolean> };
+  connect: () => Promise<unknown>;
+  disconnect: () => Promise<unknown>;
+  once: boolean;
+  sleep?: () => Promise<unknown>;
+  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);
+  }
+  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();
+  }
+}
+
+export async function main() {
   const prisma = new PrismaClient();
-
-async function main() {
-  await prisma.$connect();
   const worker = new ModerationWorker(prisma);
-  if (process.argv.includes("--once")) {
+  await runWorker({
-    await worker.runOnce();
+    worker,
-    return;
+    connect: () => prisma.$connect(),
-  }
+    disconnect: () => prisma.$disconnect(),
-  for (;;) {
+    once: process.argv.includes("--once"),
-    const handled = await worker.runOnce();
+  });
-    if (!handled) await new Promise((resolve) => setTimeout(resolve, 1000));
-  }
 }
 
-main()
+export function logTopLevelError(
-  .catch(() => {
+  error: unknown,
-    process.exitCode = 1;
+  logger: Pick<Console, "error"> = console,
-  })
+) {
-  .finally(() => prisma.$disconnect());
+  logger.error(
+    JSON.stringify({
+      scope: "moderation-worker",
+      errorClass:
+        error instanceof Error && error.name ? error.name : "UnknownError",
+    }),
+  );
+}
+
+type TopLevelErrorOptions = {
+  logger?: Pick<Console, "error">;
+  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);
+}

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<ReturnType<typeof decideModeration>>(
+      (_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<typeof decideModeration>) => void;
+    const firstDecision = new Promise<ReturnType<typeof decideModeration>>(
+      (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);
+    const logs: unknown[] = [];
-
+    await new ModerationWorker(prisma, undefined, undefined, {
-    expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true);
+      error: (entry) => logs.push(entry),
-    expect(decide).not.toHaveBeenCalled();
+    }).runOnce();
     expect(
       await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
-    ).toMatchObject({ status: "PUBLISHED" });
+    ).toMatchObject({ status: "FAILED" });
-    expect(
+    expect(JSON.stringify(logs)).not.toContain("secret bottle text");
-      await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
-    ).toMatchObject({ reviewStatus: "REVIEWING", poolStatus: "CLOSED" });
   });
 
   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");
+  }
 });

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<string, unknown>).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 {
+    } catch (error) {
-      const current = await this.prisma.outboxEvent.findUniqueOrThrow({
+      if (!(error instanceof LeaseLostError)) {
-        where: { id: event.id },
+        this.logger.error({
+          eventId: event.id,
+          eventType: event.eventType,
+          attempt: event.attempts,
+          errorClass:
+            error instanceof Error && error.name ? error.name : "UnknownError",
         });
-      const exhausted = current.attempts >= this.maxAttempts;
+      }
+      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({
+      await this.prisma.outboxEvent.updateMany({
-        where: { id: event.id },
+        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,16 +164,32 @@ 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({
+      await this.publish(tx, event);
-        where: { id: task.id },
+      const [currentBottle, currentTask] = await Promise.all([
-      });
+        tx.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
-      if (currentTask.status !== "COMPLETED") {
+        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;
+      }
+      if (currentTask.payloadHash !== currentHash)
+        throw new Error("bottle moderation payload drift");
       const updated = await tx.bottle.updateMany({
         where: {
           id: bottle.id,
@@ -138,18 +199,38 @@ export class ModerationWorker {
         data: {
           reviewStatus: decision.reviewStatus,
           poolStatus: decision.poolStatus,
-            approvedAt:
+          approvedAt: decision.reviewStatus === "APPROVED" ? new Date() : null,
-              decision.reviewStatus === "APPROVED" ? new Date() : null,
         },
       });
-        if (updated.count !== 1)
+      if (updated.count !== 1) throw new Error("bottle moderation state drift");
-          throw new Error("bottle moderation state drift");
       await this.completeTask(tx, task.id, decision);
-      }
-      await this.publish(tx, event.id);
     });
   }
 
+  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({
+      await this.prisma.$transaction((tx) => this.publish(tx, event));
-        where: { id: event.id },
-        data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null },
-      });
       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.payloadHash !== hash
-        currentTask.status === "COMPLETED"
+      )
-      ) {
+        return;
-        await this.publish(tx, event.id);
+      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) {
+  private async publish(tx: Prisma.TransactionClient, event: Claimed) {
-    return tx.outboxEvent.update({
+    const published = await tx.outboxEvent.updateMany({
-      where: { id },
+      where: {
-      data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null },
+        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<Claimed | null> {
@@ -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<Claimed[]>`
+      const rows = await tx.$queryRaw<Omit<Claimed, "lockToken">[]>`
         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 };
     });
   }
 }

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");

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")
 }