fix(ocr): live busy pre-check before posting a candidate's escrow
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled
Some checks failed
CI / typecheck + vitest (Node 20) (push) Has been cancelled
runOcrOneShot posted the escrow before calling the supplier, so a busy supplier's 409 stranded an already-paid escrow until reclaim - and the buyer-side busy-retry multiplies this into several stranded escrows per page. Add a live /status check per candidate, immediately before its escrow posts, mirroring sessions.ts's startChat pre-check: "working" skips to the next candidate without posting, anything else (free, unreachable, malformed, slow) is treated as usable so the check is an optimisation, never a gate. Bounded to a 2s timeout so a stalled supplier cannot meaningfully delay the request. Existing candidate-fallback, escrow-ref capture, usage recording, and receipt labels are unchanged.
This commit is contained in:
@@ -4,8 +4,9 @@
|
||||
* One escrow per page against a model-scoped `ocr.page.extract.<model>.v1`
|
||||
* supplier (config.ocrCapabilityId). Lifecycle mirrors the one-shot chat
|
||||
* path, all inside the per-key mutex: parse → select supplier → preflight →
|
||||
* submitOcr (escrow → supplier call → receipt verify) → resolve Submitted
|
||||
* ref → Accept (awaited) → record usage → wallet-health → respond.
|
||||
* per-candidate live busy pre-check → submitOcr (escrow → supplier call →
|
||||
* receipt verify) → resolve Submitted ref → Accept (awaited) → record usage
|
||||
* → wallet-health → respond.
|
||||
*
|
||||
* Response (200):
|
||||
* {
|
||||
@@ -38,7 +39,7 @@ import {
|
||||
ALLOWED_OCR_OUTPUT_FORMATS,
|
||||
MAX_OCR_IMAGE_B64_CHARS,
|
||||
} from "@marketplace/shared/tx";
|
||||
import type { ProgressEvent } from "@marketplace/buyer/sdk";
|
||||
import { SupplierError, type ProgressEvent } from "@marketplace/buyer/sdk";
|
||||
import type { GatewayDeps } from "../deps.js";
|
||||
import type { GatewayStore } from "../db/store.js";
|
||||
import { requireKey } from "../middleware/apiKeyAuth.js";
|
||||
@@ -59,6 +60,10 @@ const B64_RE = /^[A-Za-z0-9+/]+={0,2}$/;
|
||||
const HEX64_RE = /^[0-9a-fA-F]{64}$/;
|
||||
const VENUE_RE = /^[a-z0-9-]{1,32}$/;
|
||||
|
||||
/** Bound on the live busy pre-check below - cheap enough that a stalled
|
||||
* supplier never meaningfully delays the request. */
|
||||
const SUPPLIER_STATUS_CHECK_TIMEOUT_MS = 2_000;
|
||||
|
||||
interface ParsedOcrRequest {
|
||||
image_b64: string;
|
||||
mime: string;
|
||||
@@ -122,6 +127,35 @@ export function venueAttribution(req: Request): { venue: string; runUserHash: st
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Live busy pre-check, run immediately before a candidate's escrow posts.
|
||||
* Mirrors sessions.ts's startChat pre-check: same signal ("working" on the
|
||||
* supplier's own /status), same best-effort posture. Posting an escrow only
|
||||
* to have the supplier reject as busy strands the escrow until reclaim and
|
||||
* burns the tx fee, so this is worth a bounded, cheap HTTP round trip.
|
||||
*
|
||||
* Never a gate: the indexer already filters candidates on advert status, so
|
||||
* a timeout, network error, non-2xx, or malformed body all resolve to "not
|
||||
* busy" (unknown, not blocking) - a briefly-unreachable /status must never
|
||||
* stop the candidate from being attempted.
|
||||
*/
|
||||
async function isSupplierBusy(
|
||||
endpointUrl: string,
|
||||
fetchFn: typeof globalThis.fetch,
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const res = await fetchFn(`${endpointUrl.replace(/\/+$/, "")}/status`, {
|
||||
signal: AbortSignal.timeout(SUPPLIER_STATUS_CHECK_TIMEOUT_MS),
|
||||
});
|
||||
if (!res.ok) return false;
|
||||
const body = (await res.json()) as unknown;
|
||||
const status = body && typeof body === "object" ? (body as { status?: unknown }).status : undefined;
|
||||
return status === "working";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function recordOcrUsage(
|
||||
store: GatewayStore,
|
||||
p: {
|
||||
@@ -238,6 +272,17 @@ async function runOcrOneShot(
|
||||
// submitOcr's return value (I2 follow-up).
|
||||
let lastErr: unknown;
|
||||
for (const cand of candidates) {
|
||||
// Live pre-check, run right here (not once at the top of the request):
|
||||
// skip a candidate that is busy right now instead of paying to find
|
||||
// out. See isSupplierBusy's docstring for the exact signal + posture.
|
||||
if (await isSupplierBusy(cand.endpointUrl, fetchFn)) {
|
||||
lastErr = new SupplierError("supplier_busy", {
|
||||
status: 409,
|
||||
message: "supplier is busy with another job",
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const onProgress = (ev: unknown): void => {
|
||||
const e = ev as ProgressEvent;
|
||||
if (e.type === "escrow_posted" && e.escrow_ref) escrowRefStr = e.escrow_ref;
|
||||
|
||||
@@ -400,3 +400,173 @@ describe("POST /v1/ocr/extract — escrow ref survives a post-escrow SupplierErr
|
||||
expect(failedLine[0].escrow_ref).toBe(knownEscrowRef);
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Busy pre-check - never post an escrow to a candidate that is busy ──────
|
||||
//
|
||||
// runOcrOneShot now checks each candidate's live /status immediately before
|
||||
// calling submitOcr for it (mirrors sessions.ts's startChat pre-check):
|
||||
// "working" means busy and moves on to the next candidate WITHOUT posting an
|
||||
// escrow; anything else (free, unreachable, malformed, slow) is treated as
|
||||
// usable. submitOcr is the escrow-posting path in these tests (same
|
||||
// monkeypatch pattern as the suites above), so asserting its call count is
|
||||
// exactly how "no escrow was posted" is observed.
|
||||
|
||||
const OCR_SUPPLIER_A = {
|
||||
...OCR_SUPPLIER,
|
||||
utxo_ref: `${"aa".repeat(32)}#0`,
|
||||
supplier_pkh: "sA",
|
||||
endpoint_url: "http://sup-a",
|
||||
};
|
||||
const OCR_SUPPLIER_B = {
|
||||
...OCR_SUPPLIER,
|
||||
utxo_ref: `${"bb".repeat(32)}#0`,
|
||||
supplier_pkh: "sB",
|
||||
endpoint_url: "http://sup-b",
|
||||
};
|
||||
|
||||
/** Indexer /suppliers lookup plus a per-endpoint /status responder, on one
|
||||
* fetchFn - everything else falls back to the suite's usual empty array. */
|
||||
function makeFetchWithStatus(
|
||||
suppliers: Array<typeof OCR_SUPPLIER>,
|
||||
statusByEndpoint: Record<string, () => Response>,
|
||||
): typeof globalThis.fetch {
|
||||
return (async (url: unknown) => {
|
||||
const u = String(url);
|
||||
if (u.includes("/suppliers")) {
|
||||
return new Response(JSON.stringify(suppliers), { status: 200 });
|
||||
}
|
||||
for (const [endpoint, respond] of Object.entries(statusByEndpoint)) {
|
||||
if (u === `${endpoint}/status`) return respond();
|
||||
}
|
||||
return new Response("[]", { status: 200 });
|
||||
}) as unknown as typeof globalThis.fetch;
|
||||
}
|
||||
|
||||
/** Same successful submitOcr shape as the happy-path suite above, reused
|
||||
* across the busy pre-check tests. */
|
||||
function makeSubmitOcrStub() {
|
||||
return vi.fn(async (_opts: { advertRef: { txHash: string; index: number } }) => ({
|
||||
output_format: "markdown",
|
||||
content: "extracted text",
|
||||
usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 },
|
||||
receipt: {
|
||||
prompt_hash: "1b".repeat(32),
|
||||
response_hash: "2c".repeat(32),
|
||||
model: "chandra-ocr-2",
|
||||
prompt_tokens: 10,
|
||||
completion_tokens: 5,
|
||||
wallclock_ms: Date.now(),
|
||||
supplier_pkh: "s3",
|
||||
escrow_ref: `${"d".repeat(64)}#0`,
|
||||
},
|
||||
receiptSignature: "sig".repeat(20),
|
||||
escrowRef: { txHash: "d".repeat(64), index: 0 },
|
||||
}));
|
||||
}
|
||||
|
||||
describe("POST /v1/ocr/extract - busy pre-check", () => {
|
||||
it("a busy supplier is never sent an escrow; the route returns the busy response", async () => {
|
||||
const fetchFn = makeFetchWithStatus([OCR_SUPPLIER], {
|
||||
"http://sup": () => new Response(JSON.stringify({ status: "working" }), { status: 200 }),
|
||||
});
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const rawKey = makeKey(deps, 0);
|
||||
const keyRow = deps.store.getKeyByHash(hashApiKey(rawKey))!;
|
||||
|
||||
const ctx = deps.registry.getContext(keyRow);
|
||||
const submitOcr = makeSubmitOcrStub();
|
||||
(ctx.sdk as { submitOcr: unknown }).submitOcr = submitOcr;
|
||||
|
||||
const res = await request(createApp(deps))
|
||||
.post("/v1/ocr/extract")
|
||||
.set("Authorization", `Bearer ${rawKey}`)
|
||||
.send(validBody());
|
||||
|
||||
// The escrow-posting path was never invoked.
|
||||
expect(submitOcr).not.toHaveBeenCalled();
|
||||
// toGatewayError maps SupplierError("supplier_busy") to 503 overloaded -
|
||||
// the same busy response a post-escrow 409 would have produced.
|
||||
expect(res.status).toBe(503);
|
||||
expect(res.body.error?.code).toBe("overloaded");
|
||||
|
||||
const rows = deps.store.listUsage(keyRow.id, 10);
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0].status).toBe("failed");
|
||||
expect(rows[0].escrow_ref).toBeNull();
|
||||
});
|
||||
|
||||
it("a free supplier proceeds to post and settle exactly as before", async () => {
|
||||
const fetchFn = makeFetchWithStatus([OCR_SUPPLIER], {
|
||||
"http://sup": () => new Response(JSON.stringify({ status: "free" }), { status: 200 }),
|
||||
});
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const rawKey = makeKey(deps, 0);
|
||||
const keyRow = deps.store.getKeyByHash(hashApiKey(rawKey))!;
|
||||
|
||||
const ctx = deps.registry.getContext(keyRow);
|
||||
const submitOcr = makeSubmitOcrStub();
|
||||
(ctx.sdk as { submitOcr: unknown }).submitOcr = submitOcr;
|
||||
|
||||
const res = await request(createApp(deps))
|
||||
.post("/v1/ocr/extract")
|
||||
.set("Authorization", `Bearer ${rawKey}`)
|
||||
.send(validBody());
|
||||
|
||||
expect(submitOcr).toHaveBeenCalledTimes(1);
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body.x_vector.receipt_labels).toContain("first-party-brokered");
|
||||
});
|
||||
|
||||
it("falls back to the second candidate when the first is busy, posting exactly one escrow", async () => {
|
||||
const fetchFn = makeFetchWithStatus([OCR_SUPPLIER_A, OCR_SUPPLIER_B], {
|
||||
"http://sup-a": () => new Response(JSON.stringify({ status: "working" }), { status: 200 }),
|
||||
"http://sup-b": () => new Response(JSON.stringify({ status: "free" }), { status: 200 }),
|
||||
});
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const rawKey = makeKey(deps, 0);
|
||||
const keyRow = deps.store.getKeyByHash(hashApiKey(rawKey))!;
|
||||
|
||||
const ctx = deps.registry.getContext(keyRow);
|
||||
const submitOcr = makeSubmitOcrStub();
|
||||
(ctx.sdk as { submitOcr: unknown }).submitOcr = submitOcr;
|
||||
|
||||
const res = await request(createApp(deps))
|
||||
.post("/v1/ocr/extract")
|
||||
.set("Authorization", `Bearer ${rawKey}`)
|
||||
.send(validBody());
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
// Exactly one escrow posted, and it is the second (free) candidate's -
|
||||
// the busy first candidate was skipped, not retried after failing.
|
||||
expect(submitOcr).toHaveBeenCalledTimes(1);
|
||||
expect(submitOcr.mock.calls[0][0].advertRef).toEqual({ txHash: "bb".repeat(32), index: 0 });
|
||||
});
|
||||
|
||||
it("a status check that errors does not block the attempt", async () => {
|
||||
const fetchFn = makeFetchWithStatus([OCR_SUPPLIER], {
|
||||
"http://sup": () => {
|
||||
// Simulates both a network error and an aborted (timed-out) request -
|
||||
// isSupplierBusy's catch block treats every failure mode the same
|
||||
// way: unknown, not busy, proceed. A real AbortSignal timeout is not
|
||||
// exercised here to keep the suite fast; the outcome is identical.
|
||||
throw new Error("simulated network failure reaching /status");
|
||||
},
|
||||
});
|
||||
const deps = makeDeps(fetchFn, { chain: fundedChain });
|
||||
const rawKey = makeKey(deps, 0);
|
||||
const keyRow = deps.store.getKeyByHash(hashApiKey(rawKey))!;
|
||||
|
||||
const ctx = deps.registry.getContext(keyRow);
|
||||
const submitOcr = makeSubmitOcrStub();
|
||||
(ctx.sdk as { submitOcr: unknown }).submitOcr = submitOcr;
|
||||
|
||||
const res = await request(createApp(deps))
|
||||
.post("/v1/ocr/extract")
|
||||
.set("Authorization", `Bearer ${rawKey}`)
|
||||
.send(validBody());
|
||||
|
||||
// The candidate was still tried despite the failed pre-check.
|
||||
expect(submitOcr).toHaveBeenCalledTimes(1);
|
||||
expect(res.status).toBe(200);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user