chore: 完善捞瓶重试配置与超时

.env.example

@@ -21,4 +21,10 @@ REFRESH_ATTEMPT_IP_WINDOW_SECONDS=60
 REFRESH_RATE_LIMIT=60
 SMS_CODE_TTL_SECONDS=300
 REDIS_KEY_PREFIX=drift:auth:
+# Overall pick retry budget and attempt cap; keep bounded to protect API latency/DB load.
+MATCH_PICK_RETRY_BUDGET_MS=5000
+MATCH_PICK_MAX_ATTEMPTS=64
+# Each pick/winner-read transaction is capped and further reduced to its remaining budget.
+MATCH_PICK_TRANSACTION_MAX_WAIT_MS=1000
+MATCH_PICK_TRANSACTION_TIMEOUT_MS=2000
 TRUST_PROXY=false

apps/api/src/match/match.service.spec.ts

@@ -0,0 +1,116 @@
+import { Prisma } from "@prisma/client";
+import { describe, expect, it, vi } from "vitest";
+import { ErrorCode } from "@drift/contracts";
+import { DomainException } from "../common/domain.exception.js";
+import type { PrismaService } from "../database/prisma.service.js";
+import type { CandidateRepository } from "./candidate.repository.js";
+import { MatchService } from "./match.service.js";
+
+const pickRequestRace = () =>
+  new Prisma.PrismaClientKnownRequestError("test", {
+    code: "P2002",
+    clientVersion: "test",
+    meta: { target: ["picker_id", "request_id"] },
+  });
+
+const transactionTimeout = () =>
+  new Prisma.PrismaClientKnownRequestError("test", {
+    code: "P2028",
+    clientVersion: "test",
+  });
+
+describe("MatchService winner reads", () => {
+  it.each([
+    ["idempotency race", pickRequestRace(), ErrorCode.SERVICE_UNAVAILABLE, 503],
+    [
+      "empty-pool visibility bridge",
+      new DomainException(
+        ErrorCode.BOTTLE_POOL_EMPTY,
+        "Bottle pool empty",
+        404,
+      ),
+      ErrorCode.BOTTLE_POOL_EMPTY,
+      404,
+    ],
+  ])(
+    "bounds a stalled winner query after an %s",
+    async (_case, initialError, expectedCode, expectedStatus) => {
+      const previousEnvironment = {
+        budget: process.env.MATCH_PICK_RETRY_BUDGET_MS,
+        attempts: process.env.MATCH_PICK_MAX_ATTEMPTS,
+        maxWait: process.env.MATCH_PICK_TRANSACTION_MAX_WAIT_MS,
+        timeout: process.env.MATCH_PICK_TRANSACTION_TIMEOUT_MS,
+      };
+      process.env.MATCH_PICK_RETRY_BUDGET_MS = "40";
+      process.env.MATCH_PICK_MAX_ATTEMPTS = "1";
+      process.env.MATCH_PICK_TRANSACTION_MAX_WAIT_MS = "20";
+      process.env.MATCH_PICK_TRANSACTION_TIMEOUT_MS = "20";
+
+      try {
+        const transactionOptions: Array<{ maxWait: number; timeout: number }> =
+          [];
+        const rootRead = vi.fn(() => {
+          throw new Error("winner read escaped its bounded transaction");
+        });
+        const transaction = vi
+          .fn()
+          .mockRejectedValueOnce(initialError)
+          .mockImplementationOnce(
+            async (
+              callback: (tx: unknown) => Promise<unknown>,
+              options: { maxWait: number; timeout: number },
+            ) => {
+              transactionOptions.push(options);
+              void callback({
+                bottlePickRequest: { findUnique: () => new Promise(() => {}) },
+              });
+              await new Promise((resolve) =>
+                setTimeout(resolve, options.timeout),
+              );
+              throw transactionTimeout();
+            },
+          );
+        const prisma = {
+          $transaction: transaction,
+          bottlePickRequest: { findUnique: rootRead },
+        } as unknown as PrismaService;
+        const service = new MatchService(
+          prisma,
+          {} as unknown as CandidateRepository,
+        );
+
+        const startedAt = performance.now();
+        await expect(service.pick("picker", "request")).rejects.toMatchObject({
+          code: expectedCode,
+          status: expectedStatus,
+        });
+        const elapsed = performance.now() - startedAt;
+
+        expect(rootRead).not.toHaveBeenCalled();
+        expect(transaction).toHaveBeenCalledTimes(2);
+        expect(transactionOptions).toHaveLength(1);
+        expect(transactionOptions[0]!.maxWait).toBeGreaterThan(0);
+        expect(transactionOptions[0]!.timeout).toBeGreaterThan(0);
+        expect(
+          transactionOptions[0]!.maxWait + transactionOptions[0]!.timeout,
+        ).toBeLessThanOrEqual(40);
+        expect(elapsed).toBeLessThan(250);
+      } finally {
+        const restore = (name: string, value: string | undefined) => {
+          if (value === undefined) delete process.env[name];
+          else process.env[name] = value;
+        };
+        restore("MATCH_PICK_RETRY_BUDGET_MS", previousEnvironment.budget);
+        restore("MATCH_PICK_MAX_ATTEMPTS", previousEnvironment.attempts);
+        restore(
+          "MATCH_PICK_TRANSACTION_MAX_WAIT_MS",
+          previousEnvironment.maxWait,
+        );
+        restore(
+          "MATCH_PICK_TRANSACTION_TIMEOUT_MS",
+          previousEnvironment.timeout,
+        );
+      }
+    },
+  );
+});

apps/api/src/match/match.service.ts

@@ -119,12 +119,18 @@ export class MatchService {
       } catch (error) {
         if (error instanceof DomainException) {
           if (error.code === ErrorCode.BOTTLE_POOL_EMPTY) {
-            const winner = await this.readWinnerUntil(
+            try {
-              pickerId,
+              const winner = await this.readWinnerUntil(
-              requestId,
+                pickerId,
-              Math.min(deadline, performance.now() + 250),
+                requestId,
-            );
+                Math.min(deadline, performance.now() + 250),
-            if (winner) return winner;
+              );
+              if (winner) return winner;
+            } catch (readError) {
+              if (classifyPickError(readError) === "SERVICE_UNAVAILABLE")
+                throw error;
+              throw readError;
+            }
           }
           throw error;
         }
@@ -133,7 +139,7 @@ export class MatchService {
         if (action === "SERVICE_UNAVAILABLE") break;
         if (action === "READ_WINNER") {
           try {
-            const winner = await this.readWinner(pickerId, requestId);
+            const winner = await this.readWinner(pickerId, requestId, deadline);
             if (winner) return winner;
           } catch (readError) {
             if (classifyPickError(readError) === "SERVICE_UNAVAILABLE") break;
@@ -180,8 +186,17 @@ export class MatchService {
     });
   }
 
-  private async readWinner(pickerId: string, requestId: string) {
+  private async readWinner(
-    const previous = await this.findPrevious(this.prisma, pickerId, requestId);
+    pickerId: string,
+    requestId: string,
+    deadline: number,
+  ) {
+    const limits = transactionLimits(this.retry, performance.now(), deadline);
+    if (!limits) return null;
+    const previous = await this.prisma.$transaction(
+      (tx) => this.findPrevious(tx, pickerId, requestId),
+      { isolationLevel: "ReadCommitted", ...limits },
+    );
     if (previous?.lease) {
       try {
         const token = decryptLeaseToken(
@@ -206,7 +221,7 @@ export class MatchService {
     deadline: number,
   ) {
     while (performance.now() < deadline) {
-      const winner = await this.readWinner(pickerId, requestId);
+      const winner = await this.readWinner(pickerId, requestId, deadline);
       if (winner) return winner;
       await new Promise((resolve) => setTimeout(resolve, 5));
     }