fix(webapp): tolerate a legacy or corrupt backfill cursor instead of throwing
The backfill cursor format changed from a bare run id to a composite <createdAt>_<id>, and decodeBackfillCursor threw on anything without the separator. A backfill in flight across that change hands the new decoder an old bare-id cursor, so it would throw on every batch. Treat an unparsable cursor (a legacy bare id, or corrupt) as "no cursor" and restart the window: re-backfill is idempotent (ClickHouse ReplacingMergeTree keyed by run id), so the in-flight job self-recovers instead of failing. Logs a warning. Adds cursor round-trip and legacy-format tests.
This commit is contained in:
@@ -47,6 +47,15 @@ export class RunsBackfillerService {
|
|||||||
// different ranges. RunStore merges the two tables only on a time-based
|
// different ranges. RunStore merges the two tables only on a time-based
|
||||||
// key, so order by createdAt and tiebreak on id within a timestamp.
|
// key, so order by createdAt and tiebreak on id within a timestamp.
|
||||||
const keyset = cursor ? decodeBackfillCursor(cursor) : undefined;
|
const keyset = cursor ? decodeBackfillCursor(cursor) : undefined;
|
||||||
|
if (cursor && !keyset) {
|
||||||
|
// Legacy/corrupt cursor: ignore it and restart the window (idempotent
|
||||||
|
// re-backfill). Self-recovers a backfill in flight across the cursor
|
||||||
|
// format change instead of throwing on every batch.
|
||||||
|
this.logger.warn(
|
||||||
|
"RunsBackfillerService: unparsable backfill cursor, restarting from window start",
|
||||||
|
{ cursor }
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
const runs = await runStore.findRuns(
|
const runs = await runStore.findRuns(
|
||||||
{
|
{
|
||||||
@@ -122,15 +131,19 @@ export function encodeBackfillCursor(createdAt: Date, id: string): string {
|
|||||||
return `${createdAt.toISOString()}${BACKFILL_CURSOR_SEPARATOR}${id}`;
|
return `${createdAt.toISOString()}${BACKFILL_CURSOR_SEPARATOR}${id}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
export function decodeBackfillCursor(cursor: string): { createdAt: Date; id: string } {
|
export function decodeBackfillCursor(cursor: string): { createdAt: Date; id: string } | undefined {
|
||||||
const separatorIndex = cursor.indexOf(BACKFILL_CURSOR_SEPARATOR);
|
const separatorIndex = cursor.indexOf(BACKFILL_CURSOR_SEPARATOR);
|
||||||
const createdAt = separatorIndex === -1 ? new Date(NaN) : new Date(cursor.slice(0, separatorIndex));
|
const createdAt = separatorIndex === -1 ? new Date(NaN) : new Date(cursor.slice(0, separatorIndex));
|
||||||
const id = separatorIndex === -1 ? "" : cursor.slice(separatorIndex + 1);
|
const id = separatorIndex === -1 ? "" : cursor.slice(separatorIndex + 1);
|
||||||
|
|
||||||
|
// A cursor with no separator is the pre-(createdAt, id) format (a bare run id,
|
||||||
|
// e.g. a backfill that was in flight across this change), or otherwise corrupt.
|
||||||
|
// The old id-only keyset can't be translated to the new (createdAt, id) order,
|
||||||
|
// so return undefined and let the caller restart the window. Re-backfilling is
|
||||||
|
// idempotent (ClickHouse ReplacingMergeTree keyed by run id), so the only cost
|
||||||
|
// is redoing the already-done portion once.
|
||||||
if (Number.isNaN(createdAt.getTime()) || id.length === 0) {
|
if (Number.isNaN(createdAt.getTime()) || id.length === 0) {
|
||||||
throw new Error(
|
return undefined;
|
||||||
`RunsBackfillerService: malformed cursor "${cursor}" (expected "<createdAt>_<id>")`
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return { createdAt, id };
|
return { createdAt, id };
|
||||||
|
|||||||
@@ -9,13 +9,38 @@ vi.mock("~/db.server", () => ({
|
|||||||
import { ClickHouse } from "@internal/clickhouse";
|
import { ClickHouse } from "@internal/clickhouse";
|
||||||
import { replicationContainerTest } from "@internal/testcontainers";
|
import { replicationContainerTest } from "@internal/testcontainers";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
import { RunsBackfillerService } from "~/services/runsBackfiller.server";
|
import {
|
||||||
|
RunsBackfillerService,
|
||||||
|
decodeBackfillCursor,
|
||||||
|
encodeBackfillCursor,
|
||||||
|
} from "~/services/runsBackfiller.server";
|
||||||
import { RunsReplicationService } from "~/services/runsReplicationService.server";
|
import { RunsReplicationService } from "~/services/runsReplicationService.server";
|
||||||
import { createInMemoryTracing } from "./utils/tracing";
|
import { createInMemoryTracing } from "./utils/tracing";
|
||||||
import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
|
import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory";
|
||||||
|
|
||||||
vi.setConfig({ testTimeout: 60_000 });
|
vi.setConfig({ testTimeout: 60_000 });
|
||||||
|
|
||||||
|
describe("backfill cursor", () => {
|
||||||
|
it("round-trips createdAt + id", () => {
|
||||||
|
const createdAt = new Date("2026-06-23T00:00:00.000Z");
|
||||||
|
const decoded = decodeBackfillCursor(encodeBackfillCursor(createdAt, "cmqpwioyy0009unul63v3mxw2"));
|
||||||
|
expect(decoded?.createdAt.toISOString()).toBe(createdAt.toISOString());
|
||||||
|
expect(decoded?.id).toBe("cmqpwioyy0009unul63v3mxw2");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("treats a legacy bare-id cursor (no separator) as undefined so the window restarts", () => {
|
||||||
|
// Pre-(createdAt, id) format: a bare run id. Decoding must not throw — it
|
||||||
|
// returns undefined so an in-flight backfill restarts the window instead of
|
||||||
|
// failing every batch after the cursor-format change.
|
||||||
|
expect(decodeBackfillCursor("cmqpwioyy0009unul63v3mxw2")).toBeUndefined();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns undefined for a corrupt cursor instead of throwing", () => {
|
||||||
|
expect(decodeBackfillCursor("not-a-date_cmqpwioyy0009unul63v3mxw2")).toBeUndefined();
|
||||||
|
expect(decodeBackfillCursor("_cmqpwioyy0009unul63v3mxw2")).toBeUndefined();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("RunsBackfillerService", () => {
|
describe("RunsBackfillerService", () => {
|
||||||
replicationContainerTest(
|
replicationContainerTest(
|
||||||
"should backfill completed runs to clickhouse",
|
"should backfill completed runs to clickhouse",
|
||||||
|
|||||||
Reference in New Issue
Block a user