services/retl-runner/src/progress.test.ts
Rama newjitsu · réplica de lectura del repositorio original.
import { describe, expect, it, vi } from "vitest";
import type { ArtifactHead, BatchHead } from "./artifacts/state";
import { batchStatistics, deliveryTotals, RunProgress } from "./progress";
const batch = (extra: Partial<BatchHead> = {}): BatchHead => ({
id: "private-batch",
action: "upsert",
first: 1,
last: 64,
status: "acknowledged",
data: { key: "private-key", sha256: "a".repeat(64), bytes: 1, compressedBytes: 1 },
effectBytes: 0,
accepted: 0,
staged: 64,
rejected: 0,
reservedEntries: 0,
reservedBytes: 0,
resultBudget: 0,
submittedRecords: 64,
...extra,
});
const head = (): ArtifactHead => ({
version: 1,
runId: "private-run",
baseline: [],
batches: [batch()],
snapshot: {
sealed: true,
parts: [],
keys: 4601,
entries: 4601,
bytes: 0,
page: 1,
pageHash: "",
refreshBefore: null,
summary: {
baselineMembers: 4183,
uniqueMembers: 4244,
newMembers: 64,
changedMembers: 0,
refreshMembers: 0,
unchangedMembers: 4180,
removals: 3,
projectedMembers: 4601,
excludedRows: 0,
},
},
});
describe("redacted run progress", () => {
it("reports sealed snapshot aggregates without treating them as a delivery change", async () => {
const h = head();
h.snapshot!.sealed = false;
expect(batchStatistics(h).snapshot).toBeUndefined();
const writeStats = vi.fn().mockResolvedValue(undefined);
const progress = new RunProgress(async () => {}, writeStats);
await progress.observe(h, true);
h.snapshot!.sealed = true;
await progress.observe(h);
expect(writeStats.mock.calls.at(-1)![0].snapshot).toEqual({
sourceRows: 4601,
uniqueMembers: 4244,
projectedMembers: 4601,
excludedRows: 0,
});
expect(progress.deliveryChanged).toBe(false);
h.batches[0] = batch({ accepted: 64, staged: 0 });
await progress.observe(h);
expect(progress.deliveryChanged).toBe(true);
});
it("counts each batch once per action, with pending and mixed outcomes distinct", () => {
const h = head();
h.batches = [
batch({ status: "prepared", staged: 0 }),
batch({ status: "unknown", staged: 0 }),
batch({ accepted: 10, staged: 50, rejected: 4 }),
batch({ accepted: 64, staged: 0 }),
batch({ rejected: 64, staged: 0 }),
batch({ accepted: 60, rejected: 4, staged: 0 }),
batch({ status: "cancelled", accepted: 10, staged: 0 }),
batch({ action: "remove", accepted: 64, staged: 0 }),
];
const stats = batchStatistics(h);
expect(stats.upsert).toEqual({
total: 7,
prepared: 1,
unconfirmed: 1,
pending: 1,
accepted: 1,
rejected: 1,
partial: 1,
cancelled: 1,
});
expect(stats.remove).toMatchObject({ total: 1, accepted: 1, partial: 0, pending: 0 });
expect(stats.records).toEqual({ accepted: 208, pending: 50, rejected: 72 });
expect(stats.recordCounts).toEqual({
upsert: { total: 448, prepared: 64, unconfirmed: 64, pending: 50, accepted: 144, rejected: 72, cancelled: 54 },
remove: { total: 64, prepared: 0, unconfirmed: 0, pending: 0, accepted: 64, rejected: 0, cancelled: 0 },
});
for (const { total, ...outcomes } of Object.values(stats.recordCounts!)) {
expect(Object.values(outcomes).reduce((sum, value) => sum + value, 0)).toBe(total);
}
expect(stats.replacement).toBeUndefined();
expect(JSON.stringify(stats)).not.toMatch(/private-batch|private-key|sha256|submittedRecords/);
h.snapshot!.strategy = "native-replace";
h.snapshot!.replacementStatus = "pending";
expect(batchStatistics(h)).toMatchObject({ replacement: "pending", remove: { total: 1 } });
});
it("persists aggregates on restore and change, retries outages, and never fails delivery", async () => {
const stats = vi.fn().mockRejectedValueOnce(new Error("private-token")).mockResolvedValue(undefined);
const stderr = vi.spyOn(process.stderr, "write").mockReturnValue(true);
try {
const progress = new RunProgress(async () => {}, stats);
const h = head();
await expect(progress.observe(h, true)).resolves.toBeUndefined();
await progress.observe(h);
await progress.summarize(h);
expect(stats).toHaveBeenCalledTimes(2);
h.batches[0] = batch({ accepted: 64, staged: 0 });
await progress.observe(h);
expect(stats).toHaveBeenCalledTimes(3);
expect(stats.mock.calls[2][0]).toMatchObject({
upsert: { accepted: 1, pending: 0 },
recordCounts: { upsert: { total: 64, accepted: 64, pending: 0 } },
});
expect(stderr.mock.calls.flat().join(" ")).not.toContain("private-token");
} finally {
stderr.mockRestore();
}
});
it.each([undefined, "prepared", "pending", "accepted"] as const)(
"reports native cleanup %s without inventing removals or replaying stages on restore",
async status => {
const messages: string[] = [];
const progress = new RunProgress(async m => {
messages.push(m);
});
const h = head();
h.snapshot!.strategy = "native-replace";
h.snapshot!.replacementStatus = status;
await progress.observe(h, true);
await progress.observe(h);
expect(messages).toHaveLength(1);
expect(messages[0]).toContain("All 4244 members will be uploaded");
expect(messages[0]).toContain("357 duplicates collapsed");
expect(messages[0]).not.toContain("unchanged skipped");
await progress.summarize(h);
await progress.summarize(h);
expect(messages).toHaveLength(2);
expect(messages[1]).toContain("full-snapshot uploads: 64 confirmed submitted");
expect(messages[1]).toContain(
`Full-audience cleanup: ${status === "prepared" ? "prepared/unconfirmed" : status ?? "not started"}`
);
expect(messages[1]).toContain("removed-member count is not provided by Google");
expect(messages[1]).not.toContain("Removals: 0");
expect(messages[1]).toContain("4180 snapshot members");
expect(messages[1].includes("may have reached Google")).toBe(status === "prepared");
}
);
it("reports the complete mirror plan and distinguishes submission from acceptance", async () => {
const messages: string[] = [];
const progress = new RunProgress(async m => {
messages.push(m);
});
await progress.observe(head(), true);
expect(messages.some(m => m.startsWith("Delivery totals"))).toBe(false);
await progress.summarize(head());
const text = messages.join("\n");
for (const expected of [
"4183 baseline members",
"64 new",
"4180 unchanged skipped",
"3 to remove",
"4601 source rows",
"4244 unique members",
"4601 projected members",
"357 duplicates collapsed",
"64 confirmed submitted in 1 batches",
"0 accepted, 64 pending",
"3 removals",
"Removals are blocked",
])
expect(text).toContain(expected);
expect(text).not.toContain("private-");
await progress.observe(head());
await progress.summarize(head());
expect(messages).toHaveLength(2);
});
it("logs short stages and one final total as acceptance and removals progress", async () => {
const messages: string[] = [];
const progress = new RunProgress(async m => {
messages.push(m);
});
const h = head();
await progress.observe(h, true);
h.batches[0] = batch({ accepted: 64, staged: 0 });
await progress.observe(h);
expect(messages.at(-1)).toBe(
"Status updated for 64 additions/upserts: 64 accepted, 0 pending processing, 0 rejected in this batch."
);
h.batches.push(
batch({
id: "private-removal",
action: "remove",
first: 65,
last: 67,
status: "prepared",
submittedRecords: undefined,
staged: 0,
})
);
await progress.observe(h);
expect(messages.at(-1)).toBe("Preparing 3 removals.");
expect(messages.join("\n")).not.toMatch(/may have reached|Confirmation missing/);
h.batches[1] = { ...h.batches[1], status: "acknowledged", submittedRecords: 3, staged: 3 };
await progress.observe(h);
expect(messages.at(-1)).toBe("Submitted 3 removals; 0 accepted, 3 pending processing, 0 rejected in this batch.");
const beforeRepeat = messages.length;
await progress.observe(h);
expect(messages).toHaveLength(beforeRepeat);
expect(messages.some(m => m.startsWith("Delivery totals"))).toBe(false);
await progress.summarize(h);
await progress.summarize(h);
expect(messages.at(-1)).toContain("Removals: 3 confirmed submitted in 1 batches; 0 accepted, 3 pending");
expect(messages.at(-1)).not.toContain("Removals are blocked");
expect(messages.filter(m => m.startsWith("Delivery totals"))).toHaveLength(1);
expect(messages.filter(m => m.startsWith("Mirror comparison"))).toHaveLength(1);
const resumed: string[] = [];
const nextAttempt = new RunProgress(async m => {
resumed.push(m);
});
await nextAttempt.observe(h, true);
await nextAttempt.observe(h);
expect(resumed).toHaveLength(1);
await nextAttempt.summarize(h);
expect(resumed[0]).toContain("4183 baseline members");
expect(resumed[1]).toContain("including earlier attempts");
});
it.each(["prepared", "unknown"] as const)(
"warns about unresolved %s delivery at the end of an attempt",
async status => {
const messages: string[] = [];
const progress = new RunProgress(async m => {
messages.push(m);
});
const h = head();
h.batches = [batch({ status: "prepared", submittedRecords: undefined, staged: 0 })];
await progress.observe(h);
expect(messages.at(-1)).toBe("Preparing 64 additions/upserts.");
if (status === "unknown") {
h.batches[0].status = "unknown";
await progress.observe(h);
expect(messages.at(-1)).toContain("Confirmation missing for 64 additions/upserts");
}
await progress.summarize(h);
expect(messages.at(-1)).toContain("64 prepared/unconfirmed");
expect(messages.at(-1)).toContain("may have reached the destination");
expect(messages.filter(m => m.startsWith("Delivery totals"))).toHaveLength(1);
}
);
it("summarizes an empty extraction once", async () => {
const write = vi.fn().mockResolvedValue(undefined);
const progress = new RunProgress(write);
const h = { ...head(), snapshot: undefined, batches: [] };
await progress.observe(h, true);
expect(write).not.toHaveBeenCalled();
await progress.summarize(h);
await progress.summarize(h);
expect(write).toHaveBeenCalledTimes(1);
expect(write.mock.calls[0][0]).toContain("0 confirmed submitted");
});
it("does not label reconciliation of a restored unknown batch as a new submission", async () => {
const messages: string[] = [];
const progress = new RunProgress(async m => {
messages.push(m);
});
const h = head();
h.batches = [batch({ status: "unknown", submittedRecords: undefined, staged: 0 })];
await progress.observe(h, true);
h.batches[0] = batch({ accepted: 64, staged: 0 });
await progress.observe(h);
expect(messages.at(-1)).toContain("Status updated for 64 additions/upserts: 64 accepted");
expect(messages.join("\n")).not.toContain("Submitted 64");
await progress.summarize(h);
expect(messages.at(-1)).toContain("64 confirmed submitted");
});
it("never labels prepared or unknown records as confirmed submitted", () => {
const h = head();
h.batches = [
batch({ status: "unknown", submittedRecords: undefined, staged: 0 }),
batch({ status: "prepared", submittedRecords: undefined, staged: 0 }),
];
expect(deliveryTotals(h, "upsert")).toMatchObject({ submitted: 0, accepted: 0, unconfirmed: 128 });
});
it("distinguishes rejection before submission, provider rejection and cancellation", () => {
const h = head();
h.batches = [batch({ submittedRecords: 0, rejected: 64, staged: 0 })];
expect(deliveryTotals(h, "upsert")).toMatchObject({ submitted: 0, rejected: 64 });
h.batches[0].submittedRecords = 64;
expect(deliveryTotals(h, "upsert").submitted).toBe(64);
h.batches = [batch({ status: "cancelled", staged: 0, accepted: 10 })];
expect(deliveryTotals(h, "upsert")).toMatchObject({ submitted: 64, accepted: 10, cancelled: 54, pending: 0 });
});
it("handles older artifacts without inventing a plan and tolerates log outages", async () => {
const h = head();
delete h.snapshot!.summary;
delete h.batches[0].submittedRecords;
const write = vi.fn().mockRejectedValueOnce(new Error("private-token")).mockResolvedValue(undefined);
const stderr = vi.spyOn(process.stderr, "write").mockReturnValue(true);
try {
const progress = new RunProgress(write);
await expect(progress.observe(h, true)).resolves.toBeUndefined();
expect(stderr).toHaveBeenCalledWith('{"event":"reverse_etl_progress_unavailable"}\n');
await progress.observe(h);
await progress.summarize(h);
expect(write.mock.calls.flat().join(" ")).toContain("counts are unavailable for this older run");
expect(write.mock.calls.flat().join(" ")).toContain("64 confirmed submitted");
expect(write.mock.calls.flat().join(" ")).toContain("confirmed submission totals are a lower bound");
} finally {
stderr.mockRestore();
}
});
it("retries a failed summary write without duplicating a successful one", async () => {
const write = vi.fn().mockRejectedValueOnce(new Error("private-token")).mockResolvedValue(undefined);
const stderr = vi.spyOn(process.stderr, "write").mockReturnValue(true);
try {
const progress = new RunProgress(write);
await expect(progress.summarize(head())).resolves.toBeUndefined();
await progress.summarize(head());
await progress.summarize(head());
expect(write).toHaveBeenCalledTimes(2);
expect(stderr.mock.calls.flat().join(" ")).not.toContain("private-token");
} finally {
stderr.mockRestore();
}
});
});