fix(dn): harden staged lease and candidate reuse
This commit is contained in:
DEls
parent
1f988a9a7f
commit
e388e460d0
7 changed files with 249 additions and 105 deletions
|
|
@ -164,6 +164,17 @@ def self_test():
|
|||
assert merged == [old[0], replacement]
|
||||
assert changes == [("replace", replacement)]
|
||||
assert merge_indexes(merged, [replacement])[1] == []
|
||||
candidate_key = field("candidate_key")
|
||||
assert candidate_key == {"name": "candidate_key", "type": "text", "required": False}
|
||||
candidate_index = (
|
||||
"CREATE UNIQUE INDEX `idx_dn_batch_notes_candidate_key` "
|
||||
"ON `dn_batch_notes` (`batch`, `candidate_key`) "
|
||||
"WHERE `candidate_key` IS NOT NULL AND `candidate_key` != ''"
|
||||
)
|
||||
candidate_indexes, candidate_changes = merge_indexes([], [candidate_index])
|
||||
assert candidate_indexes == [candidate_index]
|
||||
assert candidate_changes == [("add", candidate_index)]
|
||||
assert merge_indexes(candidate_indexes, [candidate_index])[1] == []
|
||||
print("self-test: index replacement and idempotence passed")
|
||||
return 0
|
||||
|
||||
|
|
@ -187,10 +198,10 @@ def main():
|
|||
print(f"offline preview: {error}")
|
||||
print("would create: dn_batch_sources, dn_batch_pages, "
|
||||
"supplier_document_aliases, dn_review_events, dn_pipeline_locks")
|
||||
print("would extend: dn_batches, dn_batch_notes, delivery_notes")
|
||||
print("would extend: dn_batches, dn_batch_notes(candidate_key), delivery_notes")
|
||||
print("would add unique indexes: source(batch,file_hash), "
|
||||
"page(source,source_page,region), alias(normalised_key), "
|
||||
"lock(batch), delivery_notes(batch_note)")
|
||||
"lock(batch), candidate(batch,candidate_key), delivery_notes(batch_note)")
|
||||
print("would add partial unique capture-session index on dn_batches")
|
||||
print("\nDRY RUN. No schema changes written.")
|
||||
return 0
|
||||
|
|
@ -281,9 +292,14 @@ def main():
|
|||
values=["matching", "needs_review", "ready_to_post", "posted", "dismissed"]),
|
||||
field("assembly_revision", "number"), field("pages", "json", maxSize=2000000),
|
||||
relation("supplier", supplier_id), field("supplier_name_raw"),
|
||||
field("match_method"), field("match_evidence", "json", maxSize=2000000),
|
||||
field("match_method"), field("match_evidence", "json", maxSize=2000000), field("candidate_key"),
|
||||
field("posted_at", "date"),
|
||||
])
|
||||
add_indexes(token, "dn_batch_notes", [
|
||||
"CREATE UNIQUE INDEX `idx_dn_batch_notes_candidate_key` "
|
||||
"ON `dn_batch_notes` (`batch`, `candidate_key`) "
|
||||
"WHERE `candidate_key` IS NOT NULL AND `candidate_key` != ''",
|
||||
])
|
||||
add_fields(token, "delivery_notes", [relation("batch_note", notes_id)])
|
||||
|
||||
add_indexes(token, "delivery_notes", [
|
||||
|
|
|
|||
|
|
@ -89,9 +89,8 @@ describe("assembly revision route", () => {
|
|||
|
||||
it("reclaims an expired lease and releases it after a failed revision", async () => {
|
||||
lockCreate.mockRejectedValueOnce(new Error("duplicate lock")).mockResolvedValueOnce({ id: "new-lock" });
|
||||
lockGet
|
||||
.mockResolvedValueOnce({ id: "expired-lock", batch: id, worker: "dead-worker", expires_at: new Date(Date.now() - 60_000).toISOString() })
|
||||
.mockImplementationOnce(async () => lockCreate.mock.calls[1][0]);
|
||||
lockList.mockResolvedValueOnce([{ id: "expired-lock", batch: id, worker: "dead-worker", expires_at: new Date(Date.now() - 60_000).toISOString() }]);
|
||||
lockGet.mockImplementationOnce(async () => lockCreate.mock.calls[1][0]);
|
||||
notesUpdate.mockRejectedValueOnce(new Error("write failed"));
|
||||
const response = await POST(request({ revision: 1, groups: [{ pages: [1] }, { pages: [2] }], dismissedPages: [], reason: "try" }), { params: Promise.resolve({ id }) });
|
||||
expect(response.status).toBe(500);
|
||||
|
|
|
|||
|
|
@ -3,96 +3,109 @@ import { withBatchLease } from "./lock";
|
|||
|
||||
const BATCH_ID = "abcdefghijklmno";
|
||||
|
||||
function lockPb(options: {
|
||||
create?: (data: Record<string, unknown>) => Promise<Record<string, unknown>>;
|
||||
getOne?: (id: string) => Promise<Record<string, unknown>>;
|
||||
remove?: (id: string) => Promise<unknown>;
|
||||
}) {
|
||||
const create = vi.fn(options.create ?? (async (data) => ({ ...data })));
|
||||
const getOne = vi.fn(options.getOne ?? (async () => { throw new Error("not found"); }));
|
||||
const remove = vi.fn(options.remove ?? (async () => undefined));
|
||||
type Lock = Record<string, unknown> & { id: string; batch: string; worker: string; expires_at: string };
|
||||
|
||||
function lockPb(initial: Lock[] = []) {
|
||||
const records = new Map(initial.map((record) => [record.id, { ...record }]));
|
||||
const create = vi.fn(async (data: Record<string, unknown>) => {
|
||||
if ([...records.values()].some((record) => record.batch === data.batch)) {
|
||||
throw new Error("unique batch lock");
|
||||
}
|
||||
const record = data as Lock;
|
||||
records.set(record.id, { ...record });
|
||||
return { ...record };
|
||||
});
|
||||
const getOne = vi.fn(async (id: string) => {
|
||||
const record = records.get(id);
|
||||
if (!record) throw new Error("not found");
|
||||
return { ...record };
|
||||
});
|
||||
const getFullList = vi.fn(async () => [...records.values()].map((record) => ({ ...record })));
|
||||
const remove = vi.fn(async (id: string) => { records.delete(id); });
|
||||
return {
|
||||
records,
|
||||
create,
|
||||
getOne,
|
||||
getFullList,
|
||||
remove,
|
||||
pb: {
|
||||
collection: vi.fn(() => ({ create, getOne, delete: remove })),
|
||||
},
|
||||
pb: { collection: vi.fn(() => ({ create, getOne, getFullList, delete: remove })) },
|
||||
};
|
||||
}
|
||||
|
||||
describe("withBatchLease", () => {
|
||||
it("uses one deterministic PocketBase record id so only the create winner runs", async () => {
|
||||
const records = new Map<string, Record<string, unknown>>();
|
||||
const create = async (data: Record<string, unknown>) => {
|
||||
const id = String(data.id);
|
||||
if (records.has(id)) throw new Error("duplicate record id");
|
||||
records.set(id, { ...data });
|
||||
return { ...data };
|
||||
};
|
||||
const { pb, create: createMock, remove } = lockPb({
|
||||
create,
|
||||
getOne: async (id) => records.get(id) ?? Promise.reject(new Error("not found")),
|
||||
remove: async (id) => { records.delete(id); },
|
||||
});
|
||||
let finishWinner: ((value: string) => void) | undefined;
|
||||
const winner = vi.fn(async () => new Promise<string>((resolve) => { finishWinner = resolve; }));
|
||||
const loser = vi.fn(async () => "must not run");
|
||||
it("uses a distinct valid record id for each contender while UNIQUE(batch) admits only one", async () => {
|
||||
const { pb, create } = lockPb();
|
||||
let releaseFirst: (() => void) | undefined;
|
||||
const first = withBatchLease(pb as never, BATCH_ID, "worker-a", async () => new Promise<void>((resolve) => { releaseFirst = resolve; }));
|
||||
await vi.waitFor(() => expect(create).toHaveBeenCalledOnce());
|
||||
const secondFn = vi.fn(async () => "must not run");
|
||||
const second = await withBatchLease(pb as never, BATCH_ID, "worker-b", secondFn);
|
||||
releaseFirst!();
|
||||
|
||||
const first = withBatchLease(pb as never, BATCH_ID, "worker-a", winner);
|
||||
await vi.waitFor(() => expect(winner).toHaveBeenCalledOnce());
|
||||
const second = await withBatchLease(pb as never, BATCH_ID, "worker-b", loser);
|
||||
finishWinner!("ran");
|
||||
|
||||
expect(await first).toEqual({ acquired: true, value: "ran" });
|
||||
expect(await first).toEqual({ acquired: true, value: undefined });
|
||||
expect(second).toEqual({ acquired: false });
|
||||
expect(winner).toHaveBeenCalledOnce();
|
||||
expect(loser).not.toHaveBeenCalled();
|
||||
const claims = createMock.mock.calls.map((call) => call[0]);
|
||||
expect(claims[0].id).toMatch(/^[a-z0-9]{15}$/);
|
||||
expect(claims[1].id).toBe(claims[0].id);
|
||||
expect(remove).toHaveBeenCalledOnce();
|
||||
expect(secondFn).not.toHaveBeenCalled();
|
||||
expect(create.mock.calls[0][0].id).toMatch(/^[a-z0-9]{15}$/);
|
||||
expect(create.mock.calls[1][0].id).toMatch(/^[a-z0-9]{15}$/);
|
||||
expect(create.mock.calls[1][0].id).not.toBe(create.mock.calls[0][0].id);
|
||||
});
|
||||
|
||||
it("reclaims one expired deterministic lock then releases its own replacement", async () => {
|
||||
const expired = new Date(Date.now() - 1_000).toISOString();
|
||||
const { pb, create, getOne, remove } = lockPb({});
|
||||
create
|
||||
.mockRejectedValueOnce(new Error("duplicate record id"))
|
||||
.mockImplementationOnce(async (data: Record<string, unknown>) => ({ ...data }));
|
||||
getOne
|
||||
.mockResolvedValueOnce({ id: "expired-lock", worker: "dead-worker", expires_at: expired })
|
||||
.mockImplementationOnce(async () => create.mock.calls[1][0]);
|
||||
it("reclaims one expired lock into a new acquisition id", async () => {
|
||||
const old: Lock = {
|
||||
id: "oldacquisition1", batch: BATCH_ID, worker: "dead", expires_at: new Date(Date.now() - 1_000).toISOString(),
|
||||
};
|
||||
const { pb, create, remove } = lockPb([old]);
|
||||
|
||||
const result = await withBatchLease(pb as never, BATCH_ID, "worker-a", async () => 42);
|
||||
|
||||
expect(result).toEqual({ acquired: true, value: 42 });
|
||||
expect(create).toHaveBeenCalledTimes(2);
|
||||
expect(getOne).toHaveBeenCalledTimes(2);
|
||||
expect(remove).toHaveBeenNthCalledWith(1, "expired-lock");
|
||||
expect(create.mock.calls[0][0].id).toBe(create.mock.calls[1][0].id);
|
||||
expect(create.mock.calls[1][0].id).not.toBe(old.id);
|
||||
expect(remove).toHaveBeenNthCalledWith(1, old.id);
|
||||
expect(remove).toHaveBeenNthCalledWith(2, create.mock.calls[1][0].id);
|
||||
});
|
||||
|
||||
it("does not release a lock reclaimed by another owner before finally", async () => {
|
||||
let live: Record<string, unknown> | undefined;
|
||||
const { pb, create, getOne, remove } = lockPb({
|
||||
create: async (data) => {
|
||||
live = { ...data };
|
||||
return live;
|
||||
},
|
||||
getOne: async () => live ?? Promise.reject(new Error("not found")),
|
||||
remove: async () => { live = undefined; },
|
||||
it("does not delete a replacement created after expired-lock lookup but before old deletion", async () => {
|
||||
const old: Lock = {
|
||||
id: "oldacquisition1", batch: BATCH_ID, worker: "dead", expires_at: new Date(Date.now() - 1_000).toISOString(),
|
||||
};
|
||||
const replacement: Lock = {
|
||||
id: "replacementlock", batch: BATCH_ID, worker: "worker-c", expires_at: new Date(Date.now() + 60_000).toISOString(),
|
||||
};
|
||||
const fixture = lockPb([old]);
|
||||
fixture.remove.mockImplementationOnce(async (id: string) => {
|
||||
fixture.records.delete(old.id);
|
||||
fixture.records.set(replacement.id, { ...replacement });
|
||||
fixture.records.delete(id);
|
||||
});
|
||||
const fn = vi.fn(async () => "must not run");
|
||||
|
||||
const result = await withBatchLease(fixture.pb as never, BATCH_ID, "worker-b", fn);
|
||||
|
||||
expect(result).toEqual({ acquired: false });
|
||||
expect(fn).not.toHaveBeenCalled();
|
||||
expect(fixture.remove).toHaveBeenCalledWith(old.id);
|
||||
expect(fixture.records.get(replacement.id)).toEqual(replacement);
|
||||
});
|
||||
|
||||
it("does not delete a replacement created after final ownership read", async () => {
|
||||
const fixture = lockPb();
|
||||
const replacement: Lock = {
|
||||
id: "replacementlock", batch: BATCH_ID, worker: "worker-b", expires_at: new Date(Date.now() + 60_000).toISOString(),
|
||||
};
|
||||
fixture.getOne.mockImplementationOnce(async (id: string) => {
|
||||
const old = fixture.records.get(id)!;
|
||||
const snapshot = { ...old };
|
||||
fixture.records.delete(id);
|
||||
fixture.records.set(replacement.id, { ...replacement });
|
||||
return snapshot;
|
||||
});
|
||||
|
||||
const result = await withBatchLease(pb as never, BATCH_ID, "worker-a", async () => {
|
||||
live = { id: live!.id, worker: "worker-b:replacement", expires_at: new Date(Date.now() + 60_000).toISOString() };
|
||||
return "done";
|
||||
});
|
||||
const result = await withBatchLease(fixture.pb as never, BATCH_ID, "worker-a", async () => "done");
|
||||
|
||||
expect(result).toEqual({ acquired: true, value: "done" });
|
||||
expect(create).toHaveBeenCalledOnce();
|
||||
expect(getOne).toHaveBeenCalledOnce();
|
||||
expect(remove).not.toHaveBeenCalled();
|
||||
expect(live?.worker).toBe("worker-b:replacement");
|
||||
expect(fixture.remove).toHaveBeenCalledWith(fixture.create.mock.calls[0][0].id);
|
||||
expect(fixture.records.get(replacement.id)).toEqual(replacement);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@ const LEASE_MS = 5 * 60 * 1000;
|
|||
type LockRecord = { id?: string; worker?: string; expires_at?: string };
|
||||
type LeaseResult<T> = { acquired: true; value: T } | { acquired: false };
|
||||
|
||||
/** A valid, deterministic PocketBase record id for one batch's unique lease. */
|
||||
export function batchLockId(batchId: string): string {
|
||||
return createHash("sha256").update(`dn-pipeline-lock:${batchId}`).digest("hex").slice(0, 15);
|
||||
/** A valid record id deterministic for one fresh acquisition, never the batch alone. */
|
||||
export function batchLockId(batchId: string, owner: string): string {
|
||||
return createHash("sha256").update(`dn-pipeline-lock:${batchId}:${owner}`).digest("hex").slice(0, 15);
|
||||
}
|
||||
|
||||
function isExpired(lock: LockRecord, now: number): boolean {
|
||||
|
|
@ -19,10 +19,9 @@ function isExpired(lock: LockRecord, now: number): boolean {
|
|||
/**
|
||||
* Run `fn` only while this batch's create-based lease is held.
|
||||
*
|
||||
* The initial create is the atomic claim. Reading happens only after a failed
|
||||
* create, to distinguish a live owner from one expired row that may be
|
||||
* reclaimed once. The per-acquisition worker token prevents a worker that has
|
||||
* timed out from releasing another worker's replacement lease.
|
||||
* The initial create is the atomic claim, with UNIQUE(batch) as the exclusion
|
||||
* gate. Every acquisition receives a different id derived from its fresh owner
|
||||
* token, so deleting a previous acquisition id cannot delete a replacement.
|
||||
*/
|
||||
export async function withBatchLease<T>(
|
||||
pb: PocketBase,
|
||||
|
|
@ -30,8 +29,8 @@ export async function withBatchLease<T>(
|
|||
worker: string,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<LeaseResult<T>> {
|
||||
const id = batchLockId(batchId);
|
||||
const owner = `${worker}:${randomUUID()}`;
|
||||
const id = batchLockId(batchId, owner);
|
||||
const locks = pb.collection("dn_pipeline_locks");
|
||||
const payload = {
|
||||
id,
|
||||
|
|
@ -44,13 +43,17 @@ export async function withBatchLease<T>(
|
|||
try {
|
||||
claimed = (await locks.create(payload)) as unknown as LockRecord;
|
||||
} catch {
|
||||
let existing: LockRecord;
|
||||
let existing: LockRecord | undefined;
|
||||
try {
|
||||
existing = (await locks.getOne(id)) as unknown as LockRecord;
|
||||
const rows = (await locks.getFullList({
|
||||
filter: `batch = ${JSON.stringify(batchId)}`,
|
||||
fields: "id,batch,worker,expires_at",
|
||||
})) as unknown as LockRecord[];
|
||||
existing = rows[0];
|
||||
} catch {
|
||||
return { acquired: false };
|
||||
}
|
||||
if (!isExpired(existing, Date.now())) return { acquired: false };
|
||||
if (!existing || !isExpired(existing, Date.now())) return { acquired: false };
|
||||
try {
|
||||
await locks.delete(existing.id ?? id);
|
||||
claimed = (await locks.create(payload)) as unknown as LockRecord;
|
||||
|
|
@ -63,9 +66,8 @@ export async function withBatchLease<T>(
|
|||
try {
|
||||
return { acquired: true, value: await fn() };
|
||||
} finally {
|
||||
// An expiry/reclaim can replace the deterministic row id before this
|
||||
// worker reaches finally. Check the unique owner token first so this
|
||||
// loser never deletes the replacement lease.
|
||||
// A replacement acquisition has another record id. Retain the owner check
|
||||
// as a belt-and-braces guard before deleting this acquisition's own row.
|
||||
const current = await locks.getOne(claimedId).catch(() => null) as LockRecord | null;
|
||||
if (current?.worker === owner) {
|
||||
await locks.delete(claimedId).catch(() => undefined);
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ vi.mock("./extract", async (importOriginal) => {
|
|||
return { ...actual, extractPage: extractPageMock };
|
||||
});
|
||||
|
||||
import { runStagedBatch } from "./run-staged";
|
||||
import { candidateKey, runStagedBatch } from "./run-staged";
|
||||
|
||||
const BATCH_ID = "abcdefghijklmno";
|
||||
const SOURCE_ID = "source-00000001";
|
||||
|
|
@ -32,7 +32,7 @@ function read(page: number, number: string, po: string | null): PageRead {
|
|||
function stagedPb() {
|
||||
const batch: {
|
||||
id: string; status: string; stage: string; site: string; source_count: number;
|
||||
page_count?: number; open_pages?: unknown;
|
||||
page_count?: number; open_pages?: unknown; claimed_at?: string;
|
||||
} = { id: BATCH_ID, status: "inbox", stage: "queued", site: "site-1", source_count: 1 };
|
||||
const pages = new Map<string, Record<string, unknown>>();
|
||||
const candidates = new Map<string, Record<string, unknown>>();
|
||||
|
|
@ -58,7 +58,7 @@ function stagedPb() {
|
|||
const row: Record<string, unknown> = {
|
||||
id,
|
||||
batch: data.get("batch"), source: data.get("source"), source_page: Number(data.get("source_page")),
|
||||
region: Number(data.get("region")), image: data.get("image"), image_hash: data.get("image_hash"),
|
||||
region: Number(data.get("region")), image: `${id}.png`, image_hash: data.get("image_hash"),
|
||||
mime: data.get("mime"), width: Number(data.get("width")), height: Number(data.get("height")),
|
||||
rotation: Number(data.get("rotation")), read_status: data.get("read_status"),
|
||||
};
|
||||
|
|
@ -90,7 +90,7 @@ function stagedPb() {
|
|||
const locks = new Map<string, Record<string, unknown>>();
|
||||
const lockCollection = {
|
||||
create: vi.fn(async (data: Record<string, unknown>) => {
|
||||
if (locks.has(String(data.id))) throw new Error("duplicate id");
|
||||
if ([...locks.values()].some((lock) => lock.batch === data.batch)) throw new Error("duplicate batch");
|
||||
locks.set(String(data.id), { ...data });
|
||||
return data;
|
||||
}),
|
||||
|
|
@ -99,10 +99,13 @@ function stagedPb() {
|
|||
if (!row) throw new Error("not found");
|
||||
return row;
|
||||
}),
|
||||
getFullList: vi.fn(async () => [...locks.values()]),
|
||||
delete: vi.fn(async (id: string) => { locks.delete(id); }),
|
||||
};
|
||||
const filesGetURL = vi.fn((record: { id: string }, filename: string) => `http://files.test/${record.id}/${filename}`);
|
||||
const pb = {
|
||||
authStore: { token: "token" },
|
||||
files: { getURL: filesGetURL },
|
||||
filter: (raw: string, params: Record<string, unknown>) => raw.replace(/\{:(\w+)\}/g, (_match, key: string) => JSON.stringify(params[key])),
|
||||
collection: (name: string) => {
|
||||
if (name === "dn_batches") return {
|
||||
|
|
@ -119,7 +122,7 @@ function stagedPb() {
|
|||
throw new Error(`unexpected collection ${name}`);
|
||||
},
|
||||
};
|
||||
return { pb, batch, pages, candidates, updates, pageCollection, candidateCollection };
|
||||
return { pb, batch, pages, candidates, updates, pageCollection, candidateCollection, filesGetURL };
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
|
|
@ -132,6 +135,34 @@ beforeEach(() => {
|
|||
afterEach(() => vi.unstubAllEnvs());
|
||||
|
||||
describe("runStagedBatch", () => {
|
||||
it("keeps unknown document candidates page-specific while normalising known printed numbers", () => {
|
||||
const page = { id: "page-1", source: "source-1", source_page: 1, region: 0 };
|
||||
expect(candidateKey(BATCH_ID, page, "DN-001")).toBe(candidateKey(BATCH_ID, page, "dn 001"));
|
||||
expect(candidateKey(BATCH_ID, page, "")).not.toBe(candidateKey(BATCH_ID, { ...page, source_page: 2 }, ""));
|
||||
});
|
||||
|
||||
it("stamps claimed_at before reading and leaves an overlapping worker not_claimed", async () => {
|
||||
const fixture = stagedPb();
|
||||
let releaseNormalisation: (() => void) | undefined;
|
||||
const normalisationGate = new Promise<void>((resolve) => { releaseNormalisation = resolve; });
|
||||
normaliseSourceMock.mockImplementationOnce(async () => {
|
||||
await normalisationGate;
|
||||
return [{ sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 }];
|
||||
});
|
||||
extractPageMock.mockResolvedValue(read(1, "DN-1", "PO-1"));
|
||||
|
||||
const first = runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
await vi.waitFor(() => expect(fixture.batch.claimed_at).toEqual(expect.any(String)));
|
||||
const firstClaimedAt = fixture.batch.claimed_at;
|
||||
const second = await runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
releaseNormalisation!();
|
||||
|
||||
expect(second).toMatchObject({ notClaimed: true });
|
||||
expect(fixture.batch.claimed_at).toBe(firstClaimedAt);
|
||||
await expect(first).resolves.toMatchObject({ status: "processed" });
|
||||
expect(fixture.batch.claimed_at).toBe("");
|
||||
});
|
||||
|
||||
it("persists canonical pages, reuses successful reads, and leaves a clear sibling ready when another needs review", async () => {
|
||||
const fixture = stagedPb();
|
||||
normaliseSourceMock.mockResolvedValue([
|
||||
|
|
@ -156,6 +187,9 @@ describe("runStagedBatch", () => {
|
|||
]));
|
||||
expect(fixture.batch.stage).toBe("split_review");
|
||||
expect(fixture.batch.page_count).toBe(2);
|
||||
expect(typeof fixture.updates[0].data.claimed_at).toBe("string");
|
||||
expect(Date.parse(String(fixture.updates[0].data.claimed_at))).toBeGreaterThan(Date.now() - 5_000);
|
||||
expect(fixture.updates.at(-1)?.data.claimed_at).toBe("");
|
||||
expect(fixture.updates.map((update) => update.data.stage).filter(Boolean)).toEqual([
|
||||
"preparing", "reading", "split_review", "preparing", "reading", "split_review",
|
||||
]);
|
||||
|
|
@ -179,4 +213,48 @@ describe("runStagedBatch", () => {
|
|||
expect(fixture.batch.open_pages).toEqual(expect.arrayContaining([expect.objectContaining({ page: 2, reason: "read_failed" })]));
|
||||
expect([...fixture.pages.values()]).toEqual(expect.arrayContaining([expect.objectContaining({ source_page: 2, read_status: "failed", read_error: "vision unavailable" })]));
|
||||
});
|
||||
|
||||
it("re-extracts from the persisted canonical page file when normalisation changes on rerun", async () => {
|
||||
const fixture = stagedPb();
|
||||
const persistedPng = Buffer.from("persisted-canonical-png");
|
||||
vi.stubGlobal("fetch", vi.fn(async (url: string) => new Response(
|
||||
String(url).includes("page-1.png") ? persistedPng : Buffer.from("original-source"),
|
||||
{ headers: { "content-type": "image/png" } },
|
||||
)));
|
||||
normaliseSourceMock
|
||||
.mockResolvedValueOnce([{ sourcePage: 1, png: Buffer.from("first-normalisation"), mime: "image/png", width: 100, height: 200, rotation: 0 }])
|
||||
.mockResolvedValueOnce([{ sourcePage: 1, png: Buffer.from("changed-normalisation"), mime: "image/png", width: 100, height: 200, rotation: 0 }]);
|
||||
extractPageMock.mockResolvedValue(read(1, "DN-1", "PO-1"));
|
||||
|
||||
await runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
fixture.pages.get("page-1")!.prompt_version = "old";
|
||||
await runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
|
||||
expect(extractPageMock).toHaveBeenCalledTimes(2);
|
||||
expect(extractPageMock.mock.calls[1][0]).toEqual(persistedPng);
|
||||
expect(fixture.filesGetURL).toHaveBeenCalledWith(expect.objectContaining({ id: SOURCE_ID }), "original.jpg");
|
||||
expect(fixture.filesGetURL).toHaveBeenCalledWith(expect.objectContaining({ id: "page-1" }), "page-1.png");
|
||||
});
|
||||
|
||||
it("moves the existing DN-2 candidate by stable key when an earlier failed page later recovers as DN-1", async () => {
|
||||
const fixture = stagedPb();
|
||||
normaliseSourceMock.mockResolvedValue([
|
||||
{ sourcePage: 1, png: Buffer.from("one"), mime: "image/png", width: 100, height: 200, rotation: 0 },
|
||||
{ sourcePage: 2, png: Buffer.from("two"), mime: "image/png", width: 100, height: 200, rotation: 0 },
|
||||
]);
|
||||
extractPageMock.mockImplementation(async (_image: Buffer, page: number) => {
|
||||
if (page === 1) throw new Error("temporary vision failure");
|
||||
return read(page, "DN-2", "PO-1");
|
||||
});
|
||||
|
||||
await runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
extractPageMock.mockImplementation(async (_image: Buffer, page: number) => read(page, "DN-1", "PO-1"));
|
||||
await runStagedBatch(fixture.pb as never, BATCH_ID);
|
||||
|
||||
expect([...fixture.candidates.values()]).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: "candidate-1", delivery_note_number: "DN-2", note_index: 1, candidate_key: expect.any(String) }),
|
||||
expect.objectContaining({ id: "candidate-2", delivery_note_number: "DN-1", note_index: 0, candidate_key: expect.any(String) }),
|
||||
]));
|
||||
expect(fixture.candidates.size).toBe(2);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -26,8 +26,8 @@ export interface StagedRunResult {
|
|||
|
||||
type Batch = { id: string; site: string; status?: string; file?: string };
|
||||
type Source = { id: string; file: string; mime: string; ordinal?: number; delivery_session_id?: string };
|
||||
type PersistedPage = { id: string; source?: string; source_page?: number; region?: number };
|
||||
type ExistingCandidate = { id: string; note_index?: number; status?: string; stage?: string; posted_at?: string };
|
||||
type PersistedPage = { id: string; source?: string; source_page?: number; region?: number; image?: string };
|
||||
type ExistingCandidate = { id: string; candidate_key?: string; note_index?: number; status?: string; stage?: string; posted_at?: string };
|
||||
|
||||
/**
|
||||
* Run the non-financial stages for one batch. A successful run only produces
|
||||
|
|
@ -42,7 +42,9 @@ export async function runStagedBatch(pb: PocketBase, batchId: string): Promise<S
|
|||
|
||||
async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise<StagedRunResult> {
|
||||
const batch = (await pb.collection("dn_batches").getOne(batchId)) as unknown as Batch;
|
||||
await pb.collection("dn_batches").update(batchId, { status: "analysing", stage: "preparing", last_error: "" });
|
||||
await pb.collection("dn_batches").update(batchId, {
|
||||
status: "analysing", stage: "preparing", claimed_at: new Date().toISOString(), last_error: "",
|
||||
});
|
||||
|
||||
const sources = ((await pb.collection("dn_batch_sources").getFullList({
|
||||
filter: batchFilter(pb, batchId),
|
||||
|
|
@ -62,6 +64,7 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise<S
|
|||
const reads: PageRead[] = [];
|
||||
const failedPages: Array<{ page: number; reason: string }> = [];
|
||||
const pageDeliverySessionIds: Record<number, string> = {};
|
||||
const persistedPages = new Map<number, PersistedPage>();
|
||||
let pageCount = 0;
|
||||
let pageOffset = 0;
|
||||
|
||||
|
|
@ -85,7 +88,9 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise<S
|
|||
if (source.delivery_session_id) pageDeliverySessionIds[page] = source.delivery_session_id;
|
||||
try {
|
||||
const persisted = await createOrReusePage(pb, batchId, source.id, image);
|
||||
reads.push(await readOrReusePage(pb, { pageId: persisted.id, image: image.png, page }));
|
||||
persistedPages.set(page, persisted);
|
||||
const canonicalImage = await fetchPersistedPageImage(pb, persisted, image.png);
|
||||
reads.push(await readOrReusePage(pb, { pageId: persisted.id, image: canonicalImage, page }));
|
||||
} catch (error) {
|
||||
failedPages.push({ page, reason: boundedError(error) });
|
||||
}
|
||||
|
|
@ -116,7 +121,10 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise<S
|
|||
});
|
||||
const isReady = Boolean(resolution.purchaseOrderId) && resolution.autoResolved;
|
||||
if (isReady) ready++; else review++;
|
||||
await upsertCandidate(pb, batchId, existing.get(index), index, group, first, resolution, isReady);
|
||||
const pageIdentity = persistedPages.get(group.pages[0]);
|
||||
if (!pageIdentity) continue;
|
||||
const key = candidateKey(batchId, pageIdentity, first.doc_number);
|
||||
await upsertCandidate(pb, batchId, existing.get(key), key, index, group, first, resolution, isReady);
|
||||
}
|
||||
|
||||
const openPages = [
|
||||
|
|
@ -143,15 +151,26 @@ async function runClaimedStagedBatch(pb: PocketBase, batchId: string): Promise<S
|
|||
}
|
||||
|
||||
async function fetchSourceBytes(pb: PocketBase, source: Source): Promise<{ bytes: Buffer; mime: string }> {
|
||||
const base = process.env.NEXT_PUBLIC_POCKETBASE_URL;
|
||||
if (!base) throw new Error("NEXT_PUBLIC_POCKETBASE_URL is not set");
|
||||
const response = await fetch(`${base}/api/files/dn_batch_sources/${source.id}/${source.file}`, {
|
||||
const response = await fetch(pb.files.getURL(source as never, source.file), {
|
||||
headers: { Authorization: pb.authStore.token },
|
||||
});
|
||||
if (!response.ok) throw new Error(`source file fetch failed: ${response.status}`);
|
||||
return { bytes: Buffer.from(await response.arrayBuffer()), mime: response.headers.get("content-type") ?? source.mime };
|
||||
}
|
||||
|
||||
async function fetchPersistedPageImage(
|
||||
pb: PocketBase,
|
||||
page: PersistedPage,
|
||||
fallback: Buffer,
|
||||
): Promise<Buffer> {
|
||||
if (!page.image) return fallback;
|
||||
const response = await fetch(pb.files.getURL(page as never, page.image), {
|
||||
headers: { Authorization: pb.authStore.token },
|
||||
});
|
||||
if (!response.ok) throw new Error(`persisted page fetch failed: ${response.status}`);
|
||||
return Buffer.from(await response.arrayBuffer());
|
||||
}
|
||||
|
||||
async function createOrReusePage(
|
||||
pb: PocketBase,
|
||||
batchId: string,
|
||||
|
|
@ -160,7 +179,7 @@ async function createOrReusePage(
|
|||
): Promise<PersistedPage> {
|
||||
const pages = pb.collection("dn_batch_pages");
|
||||
const filter = pageFilter(pb, sourceId, image.sourcePage);
|
||||
const found = (await pages.getFullList({ filter, fields: "id,source,source_page,region" })) as unknown as PersistedPage[];
|
||||
const found = (await pages.getFullList({ filter, fields: "id,source,source_page,region,image" })) as unknown as PersistedPage[];
|
||||
if (found.length > 0) return found[0];
|
||||
|
||||
const form = new FormData();
|
||||
|
|
@ -183,7 +202,7 @@ async function createOrReusePage(
|
|||
} catch (error) {
|
||||
// The unique source/page/region index is the durable idempotency key. A
|
||||
// concurrent writer may have inserted the same page after our lookup.
|
||||
const raced = (await pages.getFullList({ filter, fields: "id,source,source_page,region" })) as unknown as PersistedPage[];
|
||||
const raced = (await pages.getFullList({ filter, fields: "id,source,source_page,region,image" })) as unknown as PersistedPage[];
|
||||
if (raced.length > 0) return raced[0];
|
||||
throw error;
|
||||
}
|
||||
|
|
@ -193,6 +212,7 @@ async function upsertCandidate(
|
|||
pb: PocketBase,
|
||||
batchId: string,
|
||||
existing: ExistingCandidate | undefined,
|
||||
candidateKey: string,
|
||||
noteIndex: number,
|
||||
group: PageGroup,
|
||||
first: PageRead,
|
||||
|
|
@ -204,6 +224,7 @@ async function upsertCandidate(
|
|||
if (existing && (existing.status === "dismissed" || existing.stage === "dismissed" || existing.stage === "posted" || Boolean(existing.posted_at))) return;
|
||||
const fields = {
|
||||
batch: batchId,
|
||||
candidate_key: candidateKey,
|
||||
note_index: noteIndex,
|
||||
pages: group.pages,
|
||||
page_from: Math.min(...group.pages),
|
||||
|
|
@ -226,12 +247,25 @@ async function upsertCandidate(
|
|||
else await pb.collection("dn_batch_notes").create(fields);
|
||||
}
|
||||
|
||||
async function loadCandidates(pb: PocketBase, batchId: string): Promise<Map<number, ExistingCandidate>> {
|
||||
async function loadCandidates(pb: PocketBase, batchId: string): Promise<Map<string, ExistingCandidate>> {
|
||||
const records = (await pb.collection("dn_batch_notes").getFullList({
|
||||
filter: batchFilter(pb, batchId),
|
||||
fields: "id,note_index,status,stage,posted_at",
|
||||
fields: "id,candidate_key,note_index,status,stage,posted_at",
|
||||
})) as unknown as ExistingCandidate[];
|
||||
return new Map(records.filter((record) => Number.isInteger(record.note_index)).map((record) => [record.note_index!, record]));
|
||||
return new Map(records.filter((record) => Boolean(record.candidate_key)).map((record) => [record.candidate_key!, record]));
|
||||
}
|
||||
|
||||
/**
|
||||
* The first persisted page is stable when an earlier failed sibling recovers,
|
||||
* while the normalised printed number distinguishes documents sharing a scan.
|
||||
* An unknown number remains safely page-specific rather than colliding with
|
||||
* another unknown page.
|
||||
*/
|
||||
export function candidateKey(batchId: string, page: PersistedPage, documentNumber: string): string {
|
||||
const pageIdentity = `${page.source ?? "unknown-source"}:${page.source_page ?? 0}:${page.region ?? 0}`;
|
||||
const documentIdentity = normalisePoNumber(documentNumber);
|
||||
const identity = documentIdentity ? `document:${documentIdentity}` : `unknown-page:${pageIdentity}`;
|
||||
return createHash("sha256").update(`${batchId}|${pageIdentity}|${identity}`).digest("hex");
|
||||
}
|
||||
|
||||
async function loadPurchaseOrders(pb: PocketBase, siteId: string): Promise<EntityPurchaseOrder[]> {
|
||||
|
|
|
|||
|
|
@ -452,6 +452,8 @@ export interface DnBatchNoteRecord extends RecordModel {
|
|||
/** Back-reference to the DELIVERY_NOTES row this note was promoted into. Empty while a candidate or blocked. */
|
||||
delivery_note: string;
|
||||
stage?: CandidateStage;
|
||||
/** Stable staged identity derived from first persisted page plus document identity. */
|
||||
candidate_key?: string;
|
||||
assembly_revision?: number;
|
||||
pages?: unknown;
|
||||
supplier?: string;
|
||||
|
|
|
|||
Loading…
Reference in a new issue