From 4c87b186b4d8b0ea400b363116f1fab8306d7f35 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 21:19:03 -0700 Subject: [PATCH 1/9] perf(ingest): prefetch frozen fact identities Refs TRA-276 --- .../agent-frozen-prefetch.test.ts | 93 +++++++++++++++++++ .../agent-frozen-verification.ts | 18 +++- 2 files changed, 107 insertions(+), 4 deletions(-) create mode 100644 scripts/ingest-recovery/agent-frozen-prefetch.test.ts diff --git a/scripts/ingest-recovery/agent-frozen-prefetch.test.ts b/scripts/ingest-recovery/agent-frozen-prefetch.test.ts new file mode 100644 index 00000000..be0c619c --- /dev/null +++ b/scripts/ingest-recovery/agent-frozen-prefetch.test.ts @@ -0,0 +1,93 @@ +import { expect, test } from 'bun:test'; +import type { CanonicalHashIndex } from './agent-canonical-index'; +import { verifyAllFrozenFacts } from './agent-frozen-verification'; +import type { AgentRecoveryClient } from './agent-transport'; + +const identities = ['first', 'second', 'third'].map((factId) => ({ + category: 'messages' as const, + factId, +})); +const metadata = identities.map((identity) => ({ + ...identity, + sourceHash: 'a'.repeat(16), + payloadBytes: 100, + eventDay: '2025-09-13', + ingestedAt: '2025-09-13 01:00:00.000', +})); +const index = { complete: true, oldestDay: '2025-09-14' } as CanonicalHashIndex; + +test('prefetch overlaps one identity page while preserving the serial verification digest', async () => { + const prefetched = Promise.withResolvers(); + let inFlight = 0; + let maximumInFlight = 0; + let pages = 0; + const recovery = { + async call(method: string, input: any) { + inFlight++; + maximumInFlight = Math.max(maximumInFlight, inFlight); + try { + if (method === 'listFrozenFacts') { + const position = pages++; + if (position === 1) prefetched.resolve(); + return { + facts: [identities[position]], + nextAfter: position === identities.length - 1 ? null : identities[position], + }; + } + expect(method).toBe('inspectFrozenFactSources'); + if (input.facts[0].factId === 'first') await prefetched.promise; + return metadata.filter((row) => row.factId === input.facts[0].factId); + } finally { + inFlight--; + } + }, + } as unknown as AgentRecoveryClient; + const serial = { + async call(method: string) { + return method === 'listFrozenFacts' ? { facts: identities, nextAfter: null } : metadata; + }, + } as unknown as AgentRecoveryClient; + const report = await verifyAllFrozenFacts(recovery, index); + expect(report).toEqual(await verifyAllFrozenFacts(serial, index)); + expect(report).toMatchObject({ total: 3, expired: 3, eligibleForLegacyRetirement: true }); + expect(maximumInFlight).toBe(2); + expect(inFlight).toBe(0); + expect(pages).toBe(3); +}); + +test('a failed prefetched page drains current verification before rejecting', async () => { + const failedPage = Promise.withResolvers(); + const releaseInspection = Promise.withResolvers(); + const failure = new Error('Identity page unavailable'); + let inspectionFinished = false; + let pages = 0; + let settled = false; + const recovery = { + async call(method: string) { + if (method === 'listFrozenFacts') { + if (pages++ === 0) return { facts: [identities[0]], nextAfter: identities[0] }; + failedPage.resolve(); + throw failure; + } + expect(method).toBe('inspectFrozenFactSources'); + await releaseInspection.promise; + inspectionFinished = true; + return [metadata[0]]; + }, + } as unknown as AgentRecoveryClient; + const result = verifyAllFrozenFacts(recovery, index).then( + () => { + settled = true; + throw new Error('Verification unexpectedly succeeded'); + }, + (error: unknown) => { + settled = true; + return error; + }, + ); + await failedPage.promise; + expect(settled).toBe(false); + releaseInspection.resolve(); + expect(await result).toBe(failure); + expect(inspectionFinished).toBe(true); +}); diff --git a/scripts/ingest-recovery/agent-frozen-verification.ts b/scripts/ingest-recovery/agent-frozen-verification.ts index a3737d26..a69325bb 100644 --- a/scripts/ingest-recovery/agent-frozen-verification.ts +++ b/scripts/ingest-recovery/agent-frozen-verification.ts @@ -38,20 +38,30 @@ export async function verifyAllFrozenFacts( if (!index.complete) throw new Error('Canonical hash index export is incomplete'); const report = emptyReport(); const digest = createHash('sha256'); - let after: { category: Category; factId: string } | undefined; + let page = await recovery.call('listFrozenFacts', { limit: 100 }); for (let pageNumber = 0; pageNumber < 100_000; pageNumber++) { - const page = await recovery.call('listFrozenFacts', { after, limit: 100 }); if (!Array.isArray(page?.facts) || page.facts.length > 100) { throw new Error('Invalid frozen fact page'); } const identities = page.facts.map(validatePageIdentity); - if (identities.length > 0) await verifyFrozenPage(recovery, index, identities, report, digest); if (page.nextAfter === null) { + if (identities.length > 0) + await verifyFrozenPage(recovery, index, identities, report, digest); report.eligibleForLegacyRetirement = report.missing === 0 && report.conflicts === 0; report.verificationSha256 = digest.digest('hex'); return report; } - after = validatePageIdentity(page.nextAfter); + const after = validatePageIdentity(page.nextAfter); + // Drain both reads on failure before the caller closes the canonical index. + const [verified, next] = await Promise.allSettled([ + identities.length > 0 + ? verifyFrozenPage(recovery, index, identities, report, digest) + : Promise.resolve(), + recovery.call('listFrozenFacts', { after, limit: 100 }), + ]); + if (verified.status === 'rejected') throw verified.reason; + if (next.status === 'rejected') throw next.reason; + page = next.value; } throw new Error('Frozen fact verification page bound exceeded'); } From 1e364bb10dc89e0f50c44f9ddee0106f5f353989 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 21:20:33 -0700 Subject: [PATCH 2/9] fix(ingest): bound baseline Copy date slices --- ...apability_snapshots_versions_baseline.pipe | 2 + ...r_agent_file_events_versions_baseline.pipe | 2 + ...pair_agent_messages_versions_baseline.pipe | 2 + ..._pull_request_links_versions_baseline.pipe | 2 + ...w_unit_attributions_versions_baseline.pipe | 2 + ...r_agent_tool_events_versions_baseline.pipe | 2 + scripts/ci/tinybird-local-fixture-tests.py | 30 +++-- .../ci/tinybird_baseline_version_fixtures.py | 118 ++++++++++++++---- 8 files changed, 122 insertions(+), 38 deletions(-) diff --git a/copies/repair_agent_capability_snapshots_versions_baseline.pipe b/copies/repair_agent_capability_snapshots_versions_baseline.pipe index 4a355ed2..c2346c7a 100644 --- a/copies/repair_agent_capability_snapshots_versions_baseline.pipe +++ b/copies/repair_agent_capability_snapshots_versions_baseline.pipe @@ -35,6 +35,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND EventAt >= toDateTime({{ Date(start_day) }}) AND EventAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND EventAt >= toDateTime({{ Date(chunk_start_day) }}) + AND EventAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, capability_snapshot_pk, IngestedAt) IN ( SELECT OrgId, session_pk, capability_snapshot_pk, max(IngestedAt) AS IngestedAt FROM agent_capability_snapshot_facts diff --git a/copies/repair_agent_file_events_versions_baseline.pipe b/copies/repair_agent_file_events_versions_baseline.pipe index 22055abd..0e0d6e44 100644 --- a/copies/repair_agent_file_events_versions_baseline.pipe +++ b/copies/repair_agent_file_events_versions_baseline.pipe @@ -31,6 +31,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND EventAt >= toDateTime({{ Date(start_day) }}) AND EventAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND EventAt >= toDateTime({{ Date(chunk_start_day) }}) + AND EventAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, file_event_pk, IngestedAt) IN ( SELECT OrgId, session_pk, file_event_pk, max(IngestedAt) AS IngestedAt FROM agent_file_event_facts diff --git a/copies/repair_agent_messages_versions_baseline.pipe b/copies/repair_agent_messages_versions_baseline.pipe index 3224db3f..4231376d 100644 --- a/copies/repair_agent_messages_versions_baseline.pipe +++ b/copies/repair_agent_messages_versions_baseline.pipe @@ -50,6 +50,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND EventAt >= toDateTime({{ Date(start_day) }}) AND EventAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND EventAt >= toDateTime({{ Date(chunk_start_day) }}) + AND EventAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, message_pk, IngestedAt) IN ( SELECT OrgId, session_pk, message_pk, max(IngestedAt) AS IngestedAt FROM agent_message_facts diff --git a/copies/repair_agent_pull_request_links_versions_baseline.pipe b/copies/repair_agent_pull_request_links_versions_baseline.pipe index 6438454d..02a4d3a2 100644 --- a/copies/repair_agent_pull_request_links_versions_baseline.pipe +++ b/copies/repair_agent_pull_request_links_versions_baseline.pipe @@ -36,6 +36,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND EventAt >= toDateTime({{ Date(start_day) }}) AND EventAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND EventAt >= toDateTime({{ Date(chunk_start_day) }}) + AND EventAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, pull_request_link_pk, IngestedAt) IN ( SELECT OrgId, session_pk, pull_request_link_pk, max(IngestedAt) AS IngestedAt FROM agent_pull_request_facts diff --git a/copies/repair_agent_review_unit_attributions_versions_baseline.pipe b/copies/repair_agent_review_unit_attributions_versions_baseline.pipe index 156600f7..8428aa8c 100644 --- a/copies/repair_agent_review_unit_attributions_versions_baseline.pipe +++ b/copies/repair_agent_review_unit_attributions_versions_baseline.pipe @@ -39,6 +39,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND DecidedAt >= toDateTime({{ Date(start_day) }}) AND DecidedAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND DecidedAt >= toDateTime({{ Date(chunk_start_day) }}) + AND DecidedAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, review_unit_attribution_pk, IngestedAt) IN ( SELECT OrgId, session_pk, review_unit_attribution_pk, max(IngestedAt) AS IngestedAt FROM agent_review_unit_attributions diff --git a/copies/repair_agent_tool_events_versions_baseline.pipe b/copies/repair_agent_tool_events_versions_baseline.pipe index c8b541c0..38d4b615 100644 --- a/copies/repair_agent_tool_events_versions_baseline.pipe +++ b/copies/repair_agent_tool_events_versions_baseline.pipe @@ -56,6 +56,8 @@ SQL > WHERE OrgId = {{ String(org_id) }} AND EventAt >= toDateTime({{ Date(start_day) }}) AND EventAt < toDateTime({{ Date(end_day) }}) + INTERVAL 1 DAY + AND EventAt >= toDateTime({{ Date(chunk_start_day) }}) + AND EventAt < toDateTime({{ Date(chunk_end_day) }}) + INTERVAL 1 DAY AND tuple(OrgId, session_pk, tool_use_pk, IngestedAt) IN ( SELECT OrgId, session_pk, tool_use_pk, max(IngestedAt) AS IngestedAt FROM agent_tool_event_facts diff --git a/scripts/ci/tinybird-local-fixture-tests.py b/scripts/ci/tinybird-local-fixture-tests.py index 2b9a9f16..5ebd715f 100644 --- a/scripts/ci/tinybird-local-fixture-tests.py +++ b/scripts/ci/tinybird-local-fixture-tests.py @@ -11,7 +11,10 @@ from tinybird.tb.modules.build_common import process as build_project from tinybird.tb.modules.local_common import get_tinybird_local_client from tinybird.tb.modules.project import Project -from tinybird_baseline_version_fixtures import verify_baseline_versions +from tinybird_baseline_version_fixtures import ( + bounded_day_chunks, + verify_baseline_versions, +) ROOT = Path.cwd() @@ -82,16 +85,21 @@ def seed_versioned_facts(client) -> None: """, ) for row in rows: - run_copy( - client, - f"repair_agent_{category}_versions_baseline", - { - "org_id": str(row["OrgId"]), - "start_day": str(row["StartDay"]), - "end_day": str(row["EndDay"]), - "copy_attempt": copy_attempt, - }, - ) + start_day = str(row["StartDay"]) + end_day = str(row["EndDay"]) + for chunk_start_day, chunk_end_day in bounded_day_chunks(start_day, end_day): + run_copy( + client, + f"repair_agent_{category}_versions_baseline", + { + "org_id": str(row["OrgId"]), + "start_day": start_day, + "end_day": end_day, + "chunk_start_day": chunk_start_day, + "chunk_end_day": chunk_end_day, + "copy_attempt": copy_attempt, + }, + ) def published_days(client) -> dict[str, list[str]]: diff --git a/scripts/ci/tinybird_baseline_version_fixtures.py b/scripts/ci/tinybird_baseline_version_fixtures.py index caa86350..9f2551b1 100644 --- a/scripts/ci/tinybird_baseline_version_fixtures.py +++ b/scripts/ci/tinybird_baseline_version_fixtures.py @@ -3,7 +3,7 @@ import json import re import subprocess -from datetime import datetime, timedelta, timezone +from datetime import date, datetime, timedelta, timezone from pathlib import Path @@ -20,6 +20,21 @@ } +def bounded_day_chunks(start_day: str, end_day: str) -> list[tuple[str, str]]: + start = date.fromisoformat(start_day) + end = date.fromisoformat(end_day) + if end < start: + raise ValueError("Baseline fixture window ends before it starts") + + chunks: list[tuple[str, str]] = [] + chunk_start = start + while chunk_start <= end: + chunk_end = min(chunk_start + timedelta(days=6), end) + chunks.append((chunk_start.isoformat(), chunk_end.isoformat())) + chunk_start = chunk_end + timedelta(days=1) + return chunks + + def verify_baseline_versions(client, sources, query_rows, run_copy) -> None: org = "org_baseline_version_regression" today = datetime.now(timezone.utc).replace(hour=12, minute=0, second=0, microsecond=0) @@ -77,37 +92,86 @@ def verify_baseline_versions(client, sources, query_rows, run_copy) -> None: ) if receipt.get("successful_rows") != 5 or receipt.get("quarantined_rows") != 0: raise RuntimeError(f"Baseline version fixture insert failed for {category}") - run_copy( - client, - f"repair_agent_{category}_versions_baseline", - { - "org_id": org, - "start_day": previous.strftime("%Y-%m-%d"), - "end_day": today.strftime("%Y-%m-%d"), - "copy_attempt": str(int(today.timestamp() * 1000)), - }, + start_day = previous.strftime("%Y-%m-%d") + end_day = today.strftime("%Y-%m-%d") + copy_attempt = str(int(today.timestamp() * 1000)) + + chunks = bounded_day_chunks(start_day, end_day) + + def copy_chunk(chunk_start_day: str, chunk_end_day: str) -> None: + run_copy( + client, + f"repair_agent_{category}_versions_baseline", + { + "org_id": org, + "start_day": start_day, + "end_day": end_day, + "chunk_start_day": chunk_start_day, + "chunk_end_day": chunk_end_day, + "copy_attempt": copy_attempt, + }, + ) + + columns = re.findall( + r"^\s+`([^`]+)`\s", + Path(f"datasources/{table}.datasource").read_text(), + re.M, ) - columns = re.findall(r"^\s+`([^`]+)`\s", Path(f"datasources/{table}.datasource").read_text(), re.M) projection = ",".join(f"`{column}`" for column in columns) - rows = query_rows( - client, - f"""SELECT UserId,toString(toDate({timestamp})) AS EventDay, + + def canonical_rows() -> list[dict]: + rows = query_rows( + client, + f"""SELECT {key} AS FactKey, UserId, + toString(toDate({timestamp})) AS EventDay, DeliverySequence, IsDeleted, ContentHash=lower(hex(SHA256(toJSONString(tuple({projection}))))) AS hash_matches {',isNull(cost_usd) AS null_preserved' if category == 'messages' else ''} - FROM {target} FINAL WHERE OrgId='{org}' AND {key}='baseline-fact'""", - ) - if len(rows) == 1: - rows[0]["DeliverySequence"] = int(rows[0]["DeliverySequence"]) - if len(rows) != 1 or rows[0] != { - "UserId": "newer", - "EventDay": today.strftime("%Y-%m-%d"), - "DeliverySequence": 1, - "IsDeleted": 0, - "hash_matches": 1, - **({"null_preserved": 1} if category == "messages" else {}), - }: - raise RuntimeError(f"Baseline Copy did not preserve the exact latest version in {category}") + FROM {target} FINAL + WHERE OrgId='{org}' + AND {key} IN ('baseline-fact', 'baseline-backdated-fact') + ORDER BY FactKey""", + ) + for row in rows: + row["DeliverySequence"] = int(row["DeliverySequence"]) + return rows + + def expected_row(fact_key: str, event_day: str) -> dict: + return { + "FactKey": fact_key, + "UserId": "newer", + "EventDay": event_day, + "DeliverySequence": 1, + "IsDeleted": 0, + "hash_matches": 1, + **({"null_preserved": 1} if category == "messages" else {}), + } + + copy_chunk(*chunks[0]) + earlier_chunk_rows = canonical_rows() + expected_earlier_rows = [expected_row("baseline-backdated-fact", start_day)] + if earlier_chunk_rows != expected_earlier_rows: + raise RuntimeError( + f"Baseline Copy narrowed latest-version selection to the earlier chunk in {category}" + ) + + for chunk in chunks[1:]: + copy_chunk(*chunk) + expected_union = [ + expected_row("baseline-backdated-fact", start_day), + expected_row("baseline-fact", end_day), + ] + if canonical_rows() != expected_union: + raise RuntimeError( + f"Baseline Copy chunks did not union to the exact latest versions in {category}" + ) + + for chunk in chunks: + copy_chunk(*chunk) + if canonical_rows() != expected_union: + raise RuntimeError( + f"Baseline Copy retry did not deduplicate identical chunk output in {category}" + ) count = query_rows(client, f"SELECT count() AS n FROM {table} WHERE OrgId='{org}'") if int(count[0]["n"]) != 5: raise RuntimeError(f"Baseline Copy changed preserved source rows in {category}") From 77e617c585cb4f8bd34bab49422c5d7806df91d4 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 21:21:51 -0700 Subject: [PATCH 3/9] fix(ingest): validate prefetched identity pages Refs TRA-276 --- .../agent-frozen-verification.ts | 34 +++++++++++++------ 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/scripts/ingest-recovery/agent-frozen-verification.ts b/scripts/ingest-recovery/agent-frozen-verification.ts index a69325bb..825f85b3 100644 --- a/scripts/ingest-recovery/agent-frozen-verification.ts +++ b/scripts/ingest-recovery/agent-frozen-verification.ts @@ -38,12 +38,9 @@ export async function verifyAllFrozenFacts( if (!index.complete) throw new Error('Canonical hash index export is incomplete'); const report = emptyReport(); const digest = createHash('sha256'); - let page = await recovery.call('listFrozenFacts', { limit: 100 }); + let page = await readFrozenPage(recovery); for (let pageNumber = 0; pageNumber < 100_000; pageNumber++) { - if (!Array.isArray(page?.facts) || page.facts.length > 100) { - throw new Error('Invalid frozen fact page'); - } - const identities = page.facts.map(validatePageIdentity); + const identities = page.facts; if (page.nextAfter === null) { if (identities.length > 0) await verifyFrozenPage(recovery, index, identities, report, digest); @@ -51,13 +48,12 @@ export async function verifyAllFrozenFacts( report.verificationSha256 = digest.digest('hex'); return report; } - const after = validatePageIdentity(page.nextAfter); // Drain both reads on failure before the caller closes the canonical index. const [verified, next] = await Promise.allSettled([ identities.length > 0 ? verifyFrozenPage(recovery, index, identities, report, digest) : Promise.resolve(), - recovery.call('listFrozenFacts', { after, limit: 100 }), + readFrozenPage(recovery, page.nextAfter), ]); if (verified.status === 'rejected') throw verified.reason; if (next.status === 'rejected') throw next.reason; @@ -66,10 +62,26 @@ export async function verifyAllFrozenFacts( throw new Error('Frozen fact verification page bound exceeded'); } +async function readFrozenPage( + recovery: AgentRecoveryClient, + after?: { category: Category; factId: string }, +) { + const value: unknown = await recovery.call('listFrozenFacts', { after, limit: 100 }); + if (!value || typeof value !== 'object') throw new Error('Invalid frozen fact page'); + const page = value as Record; + if (!Array.isArray(page.facts) || page.facts.length > 100) { + throw new Error('Invalid frozen fact page'); + } + return { + facts: page.facts.map(validatePageIdentity), + nextAfter: page.nextAfter === null ? null : validatePageIdentity(page.nextAfter), + }; +} + async function verifyFrozenPage( recovery: AgentRecoveryClient, index: CanonicalHashIndex, - identities: Array<{ category: Category; factId: string }>, + identities: { category: Category; factId: string }[], report: FrozenVerificationReport, digest: ReturnType, ): Promise { @@ -148,7 +160,7 @@ function validateMetadata(value: FrozenSourceMetadata): void { !Number.isSafeInteger(value.payloadBytes) || value.payloadBytes < 2 || !/^\d{4}-\d{2}-\d{2}$/.test(value.eventDay) || - !Number.isFinite(Date.parse(value.ingestedAt.replace(' ', 'T') + 'Z')) + !Number.isFinite(Date.parse(`${value.ingestedAt.replace(' ', 'T')}Z`)) ) { throw new Error('Invalid frozen source metadata'); } @@ -171,8 +183,8 @@ function* sourceBatches(sources: FrozenSourceMetadata[]): Generator, - returned: Array<{ category: Category; factId: string }>, + requested: { category: Category; factId: string }[], + returned: { category: Category; factId: string }[], ): void { const expected = requested.map(factKey).sort(); const actual = returned.map(factKey).sort(); From 6e4d323b0140ce0599e74712d9ec8d652b789c2c Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 21:36:58 -0700 Subject: [PATCH 4/9] fix(agent-consumer): bound baseline copy recovery --- .../src/__tests__/baseline-copy-retry.test.ts | 21 +- .../bounded-baseline-copy.integration.test.ts | 222 ++++++++++++ .../src/agent-delivery-coordinator.ts | 28 ++ .../src/baseline-copy-contract.ts | 82 ++++- .../src/baseline-copy-migration.ts | 16 +- apps/agent-consumer/src/baseline-copy-plan.ts | 166 +++++++++ .../src/bounded-baseline-copy.ts | 314 +++++++++++++++++ apps/agent-consumer/src/index.ts | 80 +++-- .../agent-baseline-copy-plan.ts | 62 ++++ .../agent-baseline-copy-retry-journal.ts | 4 +- .../agent-baseline-copy.test.ts | 191 +++++----- .../ingest-recovery/agent-baseline-copy.ts | 173 ++++----- .../agent-bounded-baseline-copy.test.ts | 328 ++++++++++++++++++ .../agent-bounded-baseline-copy.ts | 269 ++++++++++++++ .../agent-bounded-baseline-proof.ts | 158 +++++++++ .../agent-migration-proof.test.ts | 32 +- .../ingest-recovery/agent-migration-proof.ts | 50 ++- .../migrate-agent-ingestion.ts | 5 +- scripts/ingest-recovery/worker.mjs | 5 + 19 files changed, 1947 insertions(+), 259 deletions(-) create mode 100644 apps/agent-consumer/src/__tests__/bounded-baseline-copy.integration.test.ts create mode 100644 apps/agent-consumer/src/baseline-copy-plan.ts create mode 100644 apps/agent-consumer/src/bounded-baseline-copy.ts create mode 100644 scripts/ingest-recovery/agent-baseline-copy-plan.ts create mode 100644 scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts create mode 100644 scripts/ingest-recovery/agent-bounded-baseline-copy.ts create mode 100644 scripts/ingest-recovery/agent-bounded-baseline-proof.ts diff --git a/apps/agent-consumer/src/__tests__/baseline-copy-retry.test.ts b/apps/agent-consumer/src/__tests__/baseline-copy-retry.test.ts index edbde0bc..cb2006ed 100644 --- a/apps/agent-consumer/src/__tests__/baseline-copy-retry.test.ts +++ b/apps/agent-consumer/src/__tests__/baseline-copy-retry.test.ts @@ -64,6 +64,23 @@ describe('baseline Copy retry service guard', () => { ); expect(unfrozen.retry).not.toHaveBeenCalled(); }); + + it('applies the actual organization guards to bounded plan mutations', async () => { + const fixture = serviceFixture(); + const input = { category: 'tool_events' } as never; + await expect(fixture.recovery.beginBoundedBaselineCopy('org-1', input)).resolves.toEqual({ + armed: true, + }); + expect(fixture.beginBounded).toHaveBeenCalledWith(input); + + const seeded = serviceFixture({ + migration: { proofSha256: 'c'.repeat(64), complete: false }, + }); + await expect(seeded.recovery.beginBoundedBaselineCopy('org-1', input)).rejects.toThrow( + 'forbidden after migration seed', + ); + expect(seeded.beginBounded).not.toHaveBeenCalled(); + }); }); function serviceFixture( @@ -74,12 +91,13 @@ function serviceFixture( } = {}, ) { const retry = vi.fn(async () => ({ armed: true })); + const beginBounded = vi.fn(async () => ({ armed: true })); const names: string[] = []; const organization = { getIngestionMigrationState: vi.fn(async () => overrides.migration ?? null), getStats: vi.fn(async () => overrides.stats ?? emptyStats), }; - const baseline = { retryBaselineCopy: retry }; + const baseline = { retryBaselineCopy: retry, beginBoundedBaselineCopy: beginBounded }; const env = { AGENT_DELIVERY_COORDINATOR: { getByName(name: string) { @@ -104,6 +122,7 @@ function serviceFixture( } as unknown as AgentConsumerEnv; return { retry, + beginBounded, names, recovery: new TraceRecovery(createExecutionContext(), env), }; diff --git a/apps/agent-consumer/src/__tests__/bounded-baseline-copy.integration.test.ts b/apps/agent-consumer/src/__tests__/bounded-baseline-copy.integration.test.ts new file mode 100644 index 00000000..6f1dc4c6 --- /dev/null +++ b/apps/agent-consumer/src/__tests__/bounded-baseline-copy.integration.test.ts @@ -0,0 +1,222 @@ +import { env as workerEnv } from 'cloudflare:workers'; +import { runInDurableObject } from 'cloudflare:test'; +import { sha256Hex } from '@trace-flow/utils'; +import { describe, expect, it } from 'vitest'; +import type { AgentFactBatcherInstance } from '../fact-batcher'; +import type { BaselineCopyPlan } from '../baseline-copy-contract'; +import { + armBoundedBaselineCopyChunk, + beginBoundedBaselineCopy, + completeBoundedBaselineCopy, + completeBoundedBaselineCopyChunk, + confirmBoundedBaselineCopyChunk, +} from '../bounded-baseline-copy'; +import { baselineCopyPlanHashInput } from '../baseline-copy-plan'; +import { + beginBaselineCopy, + confirmBaselineCopy, + retryBaselineCopy, +} from '../baseline-copy-migration'; + +const env = workerEnv as unknown as { + AGENT_FACT_BATCHER: DurableObjectNamespace; +}; +const window = { startDay: '2026-09-01', endDay: '2026-09-10' }; +const base = { category: 'tool_events' as const, ...window, startedAt: 100 }; + +async function plan(): Promise { + const value: BaselineCopyPlan = { + sha256: '0'.repeat(64), + dailyStats: [ + { day: '2026-09-01', rows: 10, projectedBytes: 100 }, + { day: '2026-09-02', rows: 20, projectedBytes: 200 }, + { day: '2026-09-09', rows: 30, projectedBytes: 300 }, + ], + chunks: [ + { startDay: '2026-09-01', endDay: '2026-09-02', rows: 30, projectedBytes: 300 }, + { startDay: '2026-09-09', endDay: '2026-09-09', rows: 30, projectedBytes: 300 }, + ], + totalRows: 60, + totalProjectedBytes: 600, + }; + value.sha256 = await sha256Hex( + baselineCopyPlanHashInput(base.category, window.startDay, window.endDay, value), + ); + return value; +} + +function chunkInput(planSha256: string, copyAttempt = 101, chunkIndex = 0) { + return { category: base.category, planSha256, chunkIndex, copyAttempt }; +} + +describe('bounded baseline Copy checkpoint', () => { + it('serializes concurrent intents and rejects stale runners', async () => { + const host = env.AGENT_FACT_BATCHER.get(env.AGENT_FACT_BATCHER.newUniqueId()); + const copyPlan = await plan(); + await runInDurableObject(host, async (_instance, state) => { + await expect( + beginBoundedBaselineCopy(state.storage, { ...base, plan: copyPlan }), + ).resolves.toMatchObject({ mode: 'bounded', created: true, completedJobs: [] }); + const intents = await Promise.all([ + armBoundedBaselineCopyChunk(state.storage, chunkInput(copyPlan.sha256)), + armBoundedBaselineCopyChunk(state.storage, chunkInput(copyPlan.sha256)), + ]); + expect(intents.map((intent) => intent.created).sort()).toEqual([false, true]); + await expect( + armBoundedBaselineCopyChunk(state.storage, chunkInput(copyPlan.sha256, 102)), + ).rejects.toThrow('intent conflict'); + await expect( + armBoundedBaselineCopyChunk(state.storage, chunkInput(copyPlan.sha256, 101, 1)), + ).rejects.toThrow('cursor conflict'); + await expect( + confirmBoundedBaselineCopyChunk(state.storage, { + ...chunkInput(copyPlan.sha256, 102), + jobId: 'stale-job', + }), + ).rejects.toThrow('receipt conflict'); + for (const jobId of [undefined, null]) { + await expect( + confirmBoundedBaselineCopyChunk(state.storage, { + ...chunkInput(copyPlan.sha256), + jobId, + } as never), + ).rejects.toThrow('Invalid bounded baseline Copy job receipt'); + } + }); + }); + + it('resumes completed chunks in order and fences stale legacy confirmation', async () => { + const host = env.AGENT_FACT_BATCHER.get(env.AGENT_FACT_BATCHER.newUniqueId()); + const copyPlan = await plan(); + await runInDurableObject(host, async (_instance, state) => { + await beginBoundedBaselineCopy(state.storage, { ...base, plan: copyPlan }); + const first = chunkInput(copyPlan.sha256); + await armBoundedBaselineCopyChunk(state.storage, first); + await confirmBoundedBaselineCopyChunk(state.storage, { ...first, jobId: 'job-one' }); + const completed = await completeBoundedBaselineCopyChunk(state.storage, { + ...first, + jobId: 'job-one', + }); + expect(completed).toMatchObject({ completedJobs: [{ copyAttempt: 101, jobId: 'job-one' }] }); + expect(completed.activeJob).toBeUndefined(); + await expect( + completeBoundedBaselineCopyChunk(state.storage, { ...first, jobId: 'job-one' }), + ).resolves.toEqual(completed); + await expect( + confirmBaselineCopy(state.storage, { + category: base.category, + copyAttempt: 101, + jobId: 'job-one', + complete: true, + }), + ).rejects.toThrow('forbidden in bounded mode'); + await expect( + retryBaselineCopy(state.storage, { + category: base.category, + expectedJobId: 'job-one', + expectedCopyAttempt: 101, + nextCopyAttempt: 102, + observedAt: 102, + providerErrorSha256: 'a'.repeat(64), + journalSha256: 'b'.repeat(64), + }), + ).rejects.toThrow('retry conflict'); + + const second = chunkInput(copyPlan.sha256, 103, 1); + await armBoundedBaselineCopyChunk(state.storage, second); + await confirmBoundedBaselineCopyChunk(state.storage, { ...second, jobId: 'job-two' }); + await completeBoundedBaselineCopyChunk(state.storage, { ...second, jobId: 'job-two' }); + await expect( + completeBoundedBaselineCopy(state.storage, { + category: base.category, + planSha256: copyPlan.sha256, + proofSha256: 'c'.repeat(64), + completedAt: 104, + }), + ).resolves.toMatchObject({ + complete: true, + completion: { lastJobId: 'job-two', proofSha256: 'c'.repeat(64) }, + }); + }); + }); + + it('archives both failed whole-window receipts without completing either', async () => { + const host = env.AGENT_FACT_BATCHER.get(env.AGENT_FACT_BATCHER.newUniqueId()); + const copyPlan = await plan(); + await runInDurableObject(host, async (_instance, state) => { + await state.storage.put('baseline-copy:tool_events', { + ...base, + copyAttempt: 11, + jobId: 'dedicated-failed', + complete: false, + failedAttempt: { + copyAttempt: 10, + jobId: 'original-failed', + status: 'error', + observedAt: 10, + providerErrorSha256: 'a'.repeat(64), + journalSha256: 'b'.repeat(64), + }, + }); + const transitioned = await beginBoundedBaselineCopy(state.storage, { + ...base, + plan: copyPlan, + legacyFailure: { + expectedJobId: 'dedicated-failed', + expectedCopyAttempt: 11, + observedAt: 12, + providerErrorSha256: 'c'.repeat(64), + journalSha256: 'd'.repeat(64), + }, + }); + expect(transitioned.complete).toBe(false); + expect(transitioned.completedJobs).toEqual([]); + expect(transitioned.legacy).toMatchObject({ + checkpoint: { + jobId: 'dedicated-failed', + complete: false, + failedAttempt: { jobId: 'original-failed', status: 'error' }, + }, + currentFailure: { jobId: 'dedicated-failed', status: 'error' }, + }); + await expect(beginBaselineCopy(state.storage, { ...base, copyAttempt: 12 })).rejects.toThrow( + 'cannot replace bounded mode', + ); + }); + }); + + it('rejects invalid coverage, bounds, totals, hashes, and extra input fields', async () => { + const host = env.AGENT_FACT_BATCHER.get(env.AGENT_FACT_BATCHER.newUniqueId()); + const copyPlan = await plan(); + await runInDurableObject(host, async (_instance, state) => { + for (const invalid of [ + { ...copyPlan, totalRows: 61 }, + { ...copyPlan, chunks: [copyPlan.chunks[0]!] }, + { + ...copyPlan, + chunks: [ + { ...copyPlan.chunks[0]!, projectedBytes: 65 * 1024 * 1024 }, + copyPlan.chunks[1]!, + ], + }, + ]) { + await expect( + beginBoundedBaselineCopy(state.storage, { ...base, plan: invalid }), + ).rejects.toThrow(/totals|omits|chunk/); + } + await expect( + beginBoundedBaselineCopy(state.storage, { + ...base, + plan: copyPlan, + unexpected: 'provider payload', + } as never), + ).rejects.toThrow('unexpected fields'); + await expect( + beginBoundedBaselineCopy(state.storage, { + ...base, + plan: { ...copyPlan, sha256: 'f'.repeat(64) }, + }), + ).rejects.toThrow('plan hash mismatch'); + }); + }); +}); diff --git a/apps/agent-consumer/src/agent-delivery-coordinator.ts b/apps/agent-consumer/src/agent-delivery-coordinator.ts index 85ebd717..a5db2a47 100644 --- a/apps/agent-consumer/src/agent-delivery-coordinator.ts +++ b/apps/agent-consumer/src/agent-delivery-coordinator.ts @@ -10,6 +10,19 @@ import { type ConfirmBaselineCopyInput, type RetryBaselineCopyInput, } from './baseline-copy-migration'; +import { + armBoundedBaselineCopyChunk, + beginBoundedBaselineCopy, + completeBoundedBaselineCopy, + completeBoundedBaselineCopyChunk, + confirmBoundedBaselineCopyChunk, +} from './bounded-baseline-copy'; +import type { + BaselineCopyChunkInput, + BeginBoundedBaselineCopyInput, + CompleteBoundedBaselineCopyInput, + ConfirmBaselineCopyChunkInput, +} from './baseline-copy-contract'; import { initializeIngestionMigration, ingestionMigrationState, @@ -130,6 +143,21 @@ class AgentDeliveryCoordinatorBase extends DurableObject { retryBaselineCopy(input: RetryBaselineCopyInput) { return retryBaselineCopy(this.ctx.storage, input); } + beginBoundedBaselineCopy(input: BeginBoundedBaselineCopyInput) { + return beginBoundedBaselineCopy(this.ctx.storage, input); + } + armBoundedBaselineCopyChunk(input: BaselineCopyChunkInput) { + return armBoundedBaselineCopyChunk(this.ctx.storage, input); + } + confirmBoundedBaselineCopyChunk(input: ConfirmBaselineCopyChunkInput) { + return confirmBoundedBaselineCopyChunk(this.ctx.storage, input); + } + completeBoundedBaselineCopyChunk(input: ConfirmBaselineCopyChunkInput) { + return completeBoundedBaselineCopyChunk(this.ctx.storage, input); + } + completeBoundedBaselineCopy(input: CompleteBoundedBaselineCopyInput) { + return completeBoundedBaselineCopy(this.ctx.storage, input); + } beginBaselineMigrationWindow(input: BaselineMigrationWindow) { return beginBaselineMigrationWindow(this.ctx.storage, input); } diff --git a/apps/agent-consumer/src/baseline-copy-contract.ts b/apps/agent-consumer/src/baseline-copy-contract.ts index 25457160..9f95ea87 100644 --- a/apps/agent-consumer/src/baseline-copy-contract.ts +++ b/apps/agent-consumer/src/baseline-copy-contract.ts @@ -9,7 +9,7 @@ export interface BaselineCopyFailedAttempt { journalSha256: string; } -export interface BaselineCopyCheckpoint { +export interface LegacyBaselineCopyCheckpoint { category: Category; startDay: string; endDay: string; @@ -20,13 +20,58 @@ export interface BaselineCopyCheckpoint { failedAttempt?: BaselineCopyFailedAttempt; } +export interface BaselineCopyDailyStat { + day: string; + rows: number; + projectedBytes: number; +} + +export interface BaselineCopyChunk { + startDay: string; + endDay: string; + rows: number; + projectedBytes: number; +} + +export interface BaselineCopyPlan { + sha256: string; + dailyStats: BaselineCopyDailyStat[]; + chunks: BaselineCopyChunk[]; + totalRows: number; + totalProjectedBytes: number; +} + +export interface BaselineCopyJobReceipt { + copyAttempt: number; + jobId: string; +} + +export interface BoundedBaselineCopyCheckpoint { + mode: 'bounded'; + category: Category; + startDay: string; + endDay: string; + startedAt: number; + plan: BaselineCopyPlan; + completedJobs: BaselineCopyJobReceipt[]; + activeJob?: { copyAttempt: number; jobId?: string }; + complete: boolean; + completion?: { proofSha256: string; completedAt: number; lastJobId: string }; + legacy?: { + checkpoint: LegacyBaselineCopyCheckpoint; + currentFailure: BaselineCopyFailedAttempt; + }; +} + +export type BaselineCopyCheckpoint = LegacyBaselineCopyCheckpoint | BoundedBaselineCopyCheckpoint; + export interface BaselineMigrationWindow { startDay: string; endDay: string; } export type BeginBaselineCopyInput = Omit< - BaselineCopyCheckpoint, + LegacyBaselineCopyCheckpoint, 'jobId' | 'complete' | 'failedAttempt' >; @@ -46,3 +91,36 @@ export interface RetryBaselineCopyInput { providerErrorSha256: string; journalSha256: string; } + +export interface BeginBoundedBaselineCopyInput { + category: Category; + startDay: string; + endDay: string; + startedAt: number; + plan: BaselineCopyPlan; + legacyFailure?: { + expectedJobId: string; + expectedCopyAttempt: number; + observedAt: number; + providerErrorSha256: string; + journalSha256: string; + }; +} + +export interface BaselineCopyChunkInput { + category: Category; + planSha256: string; + chunkIndex: number; + copyAttempt: number; +} + +export interface ConfirmBaselineCopyChunkInput extends BaselineCopyChunkInput { + jobId: string; +} + +export interface CompleteBoundedBaselineCopyInput { + category: Category; + planSha256: string; + proofSha256: string; + completedAt: number; +} diff --git a/apps/agent-consumer/src/baseline-copy-migration.ts b/apps/agent-consumer/src/baseline-copy-migration.ts index 68fabc02..f3f22583 100644 --- a/apps/agent-consumer/src/baseline-copy-migration.ts +++ b/apps/agent-consumer/src/baseline-copy-migration.ts @@ -6,13 +6,19 @@ import type { BaselineMigrationWindow, BeginBaselineCopyInput, ConfirmBaselineCopyInput, + LegacyBaselineCopyCheckpoint, RetryBaselineCopyInput, } from './baseline-copy-contract'; export type { BaselineCopyCheckpoint, + BaselineCopyChunkInput, BaselineMigrationWindow, BeginBaselineCopyInput, + BeginBoundedBaselineCopyInput, + CompleteBoundedBaselineCopyInput, ConfirmBaselineCopyInput, + ConfirmBaselineCopyChunkInput, + LegacyBaselineCopyCheckpoint, RetryBaselineCopyInput, } from './baseline-copy-contract'; const key = (category: Category) => `baseline-copy:${category}`; @@ -59,7 +65,7 @@ export async function baselineCopyCheckpoint( export async function beginBaselineCopy( storage: DurableObjectStorage, input: BeginBaselineCopyInput, -): Promise { +): Promise { if ( !CATEGORIES.includes(input.category) || !Number.isSafeInteger(input.startedAt) || @@ -75,6 +81,7 @@ export async function beginBaselineCopy( return storage.transaction(async (transaction) => { const existing = await baselineCopyCheckpoint(transaction, input.category); if (existing) { + if ('mode' in existing) throw new Error('Legacy baseline Copy cannot replace bounded mode'); if (existing.startDay !== input.startDay || existing.endDay !== input.endDay) throw new Error('Baseline Copy window changed'); return { ...existing, created: false }; @@ -95,7 +102,7 @@ export async function beginBaselineCopy( export async function confirmBaselineCopy( storage: DurableObjectStorage, input: ConfirmBaselineCopyInput, -): Promise { +): Promise { assertExactKeys(input, ['category', 'copyAttempt', 'jobId', 'complete'], 'confirm baseline Copy'); if ( !Number.isSafeInteger(input.copyAttempt) || @@ -106,6 +113,8 @@ export async function confirmBaselineCopy( throw new Error('Invalid baseline Copy confirmation'); return storage.transaction(async (transaction) => { const existing = await baselineCopyCheckpoint(transaction, input.category); + if (existing && 'mode' in existing) + throw new Error('Legacy baseline Copy confirmation is forbidden in bounded mode'); if ( existing?.copyAttempt !== input.copyAttempt || (existing?.jobId !== undefined && existing.jobId !== input.jobId) @@ -124,7 +133,7 @@ export async function confirmBaselineCopy( export async function retryBaselineCopy( storage: DurableObjectStorage, input: RetryBaselineCopyInput, -): Promise { +): Promise { assertExactKeys( input, [ @@ -156,6 +165,7 @@ export async function retryBaselineCopy( const existing = await baselineCopyCheckpoint(transaction, input.category); if ( !existing || + 'mode' in existing || existing.complete || existing.failedAttempt || existing.jobId !== input.expectedJobId || diff --git a/apps/agent-consumer/src/baseline-copy-plan.ts b/apps/agent-consumer/src/baseline-copy-plan.ts new file mode 100644 index 00000000..7f2b368d --- /dev/null +++ b/apps/agent-consumer/src/baseline-copy-plan.ts @@ -0,0 +1,166 @@ +import type { + BaselineCopyCheckpoint, + BaselineCopyPlan, + BoundedBaselineCopyCheckpoint, +} from './baseline-copy-contract'; +import { CATEGORIES, type Category } from './facts'; +import { assertExactKeys } from './agent-delivery-coordinator-validation'; + +export const MAX_BASELINE_COPY_DAYS = 367; +export const MAX_BASELINE_COPY_CHUNK_DAYS = 7; +export const MAX_BASELINE_COPY_CHUNK_ROWS = 50_000; +export const MAX_BASELINE_COPY_CHUNK_BYTES = 64 * 1024 * 1024; +export const MAX_BASELINE_COPY_CHECKPOINT_BYTES = 128 * 1024; + +export function isBoundedBaselineCopy( + checkpoint: BaselineCopyCheckpoint, +): checkpoint is BoundedBaselineCopyCheckpoint { + return 'mode' in checkpoint && checkpoint.mode === 'bounded'; +} + +export function baselineCopyPlanHashInput( + category: Category, + startDay: string, + endDay: string, + plan: BaselineCopyPlan, +): string { + return JSON.stringify({ + version: 1, + category, + startDay, + endDay, + dailyStats: plan.dailyStats, + chunks: plan.chunks, + totalRows: plan.totalRows, + totalProjectedBytes: plan.totalProjectedBytes, + }); +} + +export function validateBaselineCopyPlan( + category: Category, + startDay: string, + endDay: string, + plan: BaselineCopyPlan, +): void { + assertExactKeys( + plan, + ['sha256', 'dailyStats', 'chunks', 'totalRows', 'totalProjectedBytes'], + 'bounded baseline Copy plan', + ); + const first = dayNumber(startDay); + const last = dayNumber(endDay); + if (!CATEGORIES.includes(category) || first > last || last - first >= MAX_BASELINE_COPY_DAYS) { + throw new Error('Invalid bounded baseline Copy window'); + } + if ( + !/^[0-9a-f]{64}$/.test(plan.sha256) || + !Array.isArray(plan.dailyStats) || + plan.dailyStats.length === 0 || + plan.dailyStats.length > MAX_BASELINE_COPY_DAYS || + !Array.isArray(plan.chunks) || + plan.chunks.length === 0 || + plan.chunks.length > MAX_BASELINE_COPY_DAYS + ) { + throw new Error('Invalid bounded baseline Copy plan'); + } + + let totalRows = 0; + let totalBytes = 0; + let previousDay = first - 1; + for (const stat of plan.dailyStats) { + assertExactKeys( + stat, + ['day', 'rows', 'projectedBytes'], + 'bounded baseline Copy daily statistic', + ); + const day = dayNumber(stat.day); + if ( + day <= previousDay || + day < first || + day > last || + !positiveBounded(stat.rows, MAX_BASELINE_COPY_CHUNK_ROWS) || + !positiveBounded(stat.projectedBytes, MAX_BASELINE_COPY_CHUNK_BYTES) + ) { + throw new Error('Invalid bounded baseline Copy daily statistics'); + } + previousDay = day; + totalRows += stat.rows; + totalBytes += stat.projectedBytes; + } + if (plan.totalRows !== totalRows || plan.totalProjectedBytes !== totalBytes) { + throw new Error('Bounded baseline Copy totals do not match daily statistics'); + } + + let dailyIndex = 0; + let previousEnd = first - 1; + for (const chunk of plan.chunks) { + assertExactKeys( + chunk, + ['startDay', 'endDay', 'rows', 'projectedBytes'], + 'bounded baseline Copy chunk', + ); + const chunkStart = dayNumber(chunk.startDay); + const chunkEnd = dayNumber(chunk.endDay); + if ( + chunkStart <= previousEnd || + chunkStart < first || + chunkEnd > last || + chunkEnd < chunkStart || + chunkEnd - chunkStart >= MAX_BASELINE_COPY_CHUNK_DAYS + ) { + throw new Error('Invalid bounded baseline Copy chunk range'); + } + const startIndex = dailyIndex; + let rows = 0; + let bytes = 0; + while ( + dailyIndex < plan.dailyStats.length && + dayNumber(plan.dailyStats[dailyIndex]!.day) <= chunkEnd + ) { + const stat = plan.dailyStats[dailyIndex]!; + if (dayNumber(stat.day) < chunkStart) { + throw new Error('Bounded baseline Copy chunks overlap daily statistics'); + } + rows += stat.rows; + bytes += stat.projectedBytes; + dailyIndex++; + } + if ( + dailyIndex === startIndex || + plan.dailyStats[startIndex]!.day !== chunk.startDay || + plan.dailyStats[dailyIndex - 1]!.day !== chunk.endDay || + chunk.rows !== rows || + chunk.projectedBytes !== bytes || + !positiveBounded(rows, MAX_BASELINE_COPY_CHUNK_ROWS) || + !positiveBounded(bytes, MAX_BASELINE_COPY_CHUNK_BYTES) + ) { + throw new Error('Bounded baseline Copy chunk totals do not match daily statistics'); + } + previousEnd = chunkEnd; + } + if (dailyIndex !== plan.dailyStats.length) { + throw new Error('Bounded baseline Copy plan omits daily statistics'); + } +} + +export function requireBoundedCheckpointSize(checkpoint: BoundedBaselineCopyCheckpoint): void { + if ( + new TextEncoder().encode(JSON.stringify(checkpoint)).byteLength > + MAX_BASELINE_COPY_CHECKPOINT_BYTES + ) { + throw new Error('Bounded baseline Copy checkpoint exceeds 128 KiB'); + } +} + +function positiveBounded(value: number, maximum: number): boolean { + return Number.isSafeInteger(value) && value > 0 && value <= maximum; +} + +function dayNumber(day: string): number { + if (!/^\d{4}-\d{2}-\d{2}$/.test(day)) throw new Error('Invalid bounded baseline Copy date'); + const milliseconds = Date.parse(`${day}T00:00:00.000Z`); + if (!Number.isFinite(milliseconds) || new Date(milliseconds).toISOString().slice(0, 10) !== day) { + throw new Error('Invalid bounded baseline Copy date'); + } + return milliseconds / 86_400_000; +} diff --git a/apps/agent-consumer/src/bounded-baseline-copy.ts b/apps/agent-consumer/src/bounded-baseline-copy.ts new file mode 100644 index 00000000..f58ffec1 --- /dev/null +++ b/apps/agent-consumer/src/bounded-baseline-copy.ts @@ -0,0 +1,314 @@ +import { sha256Hex } from '@trace-flow/utils'; +import { assertExactKeys } from './agent-delivery-coordinator-validation'; +import type { + BaselineCopyCheckpoint, + BaselineCopyChunkInput, + BeginBoundedBaselineCopyInput, + BoundedBaselineCopyCheckpoint, + CompleteBoundedBaselineCopyInput, + ConfirmBaselineCopyChunkInput, +} from './baseline-copy-contract'; +import { + baselineCopyPlanHashInput, + isBoundedBaselineCopy, + requireBoundedCheckpointSize, + validateBaselineCopyPlan, +} from './baseline-copy-plan'; +import { baselineCopyCheckpoint } from './baseline-copy-migration'; + +const key = (category: BaselineCopyCheckpoint['category']) => `baseline-copy:${category}`; + +export async function beginBoundedBaselineCopy( + storage: DurableObjectStorage, + input: BeginBoundedBaselineCopyInput, +): Promise { + validateBeginInput(input); + validateBaselineCopyPlan(input.category, input.startDay, input.endDay, input.plan); + if ( + (await sha256Hex( + baselineCopyPlanHashInput(input.category, input.startDay, input.endDay, input.plan), + )) !== input.plan.sha256 + ) { + throw new Error('Bounded baseline Copy plan hash mismatch'); + } + return storage.transaction(async (transaction) => { + const existing = await baselineCopyCheckpoint(transaction, input.category); + if (existing && isBoundedBaselineCopy(existing)) { + if ( + existing.startDay !== input.startDay || + existing.endDay !== input.endDay || + existing.plan.sha256 !== input.plan.sha256 + ) { + throw new Error('Bounded baseline Copy plan conflict'); + } + return { ...existing, created: false }; + } + + let legacy: BoundedBaselineCopyCheckpoint['legacy']; + if (existing) { + const failure = input.legacyFailure; + if ( + !failure || + existing.complete || + existing.jobId !== failure.expectedJobId || + existing.copyAttempt !== failure.expectedCopyAttempt || + existing.failedAttempt?.jobId === failure.expectedJobId || + existing.startDay !== input.startDay || + existing.endDay !== input.endDay || + existing.startedAt !== input.startedAt + ) { + throw new Error('Bounded baseline Copy legacy transition conflict'); + } + legacy = { + checkpoint: existing, + currentFailure: { + copyAttempt: failure.expectedCopyAttempt, + jobId: failure.expectedJobId, + status: 'error', + observedAt: failure.observedAt, + providerErrorSha256: failure.providerErrorSha256, + journalSha256: failure.journalSha256, + }, + }; + } else if (input.legacyFailure) { + throw new Error('Bounded baseline Copy has no legacy checkpoint to archive'); + } + + const checkpoint: BoundedBaselineCopyCheckpoint = { + mode: 'bounded', + category: input.category, + startDay: input.startDay, + endDay: input.endDay, + startedAt: input.startedAt, + plan: input.plan, + completedJobs: [], + complete: false, + ...(legacy ? { legacy } : {}), + }; + requireBoundedCheckpointSize(checkpoint); + await transaction.put(key(input.category), checkpoint); + return { ...checkpoint, created: true }; + }); +} + +export async function armBoundedBaselineCopyChunk( + storage: DurableObjectStorage, + input: BaselineCopyChunkInput, +): Promise { + validateChunkInput(input); + return storage.transaction(async (transaction) => { + const existing = await requireCurrentPlan(transaction, input); + if (existing.complete || input.chunkIndex !== existing.completedJobs.length) { + throw new Error('Bounded baseline Copy chunk cursor conflict'); + } + if (existing.activeJob) { + if (existing.activeJob.copyAttempt !== input.copyAttempt) { + throw new Error('Bounded baseline Copy chunk intent conflict'); + } + return { ...existing, created: false }; + } + const checkpoint = { ...existing, activeJob: { copyAttempt: input.copyAttempt } }; + requireBoundedCheckpointSize(checkpoint); + await transaction.put(key(input.category), checkpoint); + return { ...checkpoint, created: true }; + }); +} + +export async function confirmBoundedBaselineCopyChunk( + storage: DurableObjectStorage, + input: ConfirmBaselineCopyChunkInput, +): Promise { + validateChunkInput(input); + requireJobId(input.jobId); + return storage.transaction(async (transaction) => { + const existing = await requireCurrentPlan(transaction, input); + const completed = existing.completedJobs[input.chunkIndex]; + if (completed) { + if (completed.copyAttempt === input.copyAttempt && completed.jobId === input.jobId) { + return existing; + } + throw new Error('Bounded baseline Copy chunk completion conflict'); + } + if ( + existing.complete || + input.chunkIndex !== existing.completedJobs.length || + existing.activeJob?.copyAttempt !== input.copyAttempt || + (existing.activeJob.jobId !== undefined && existing.activeJob.jobId !== input.jobId) + ) { + throw new Error('Bounded baseline Copy chunk receipt conflict'); + } + const checkpoint = { ...existing, activeJob: { ...existing.activeJob, jobId: input.jobId } }; + requireBoundedCheckpointSize(checkpoint); + await transaction.put(key(input.category), checkpoint); + return checkpoint; + }); +} + +export async function completeBoundedBaselineCopyChunk( + storage: DurableObjectStorage, + input: ConfirmBaselineCopyChunkInput, +): Promise { + validateChunkInput(input); + requireJobId(input.jobId); + return storage.transaction(async (transaction) => { + const existing = await requireCurrentPlan(transaction, input); + const completed = existing.completedJobs[input.chunkIndex]; + if (completed) { + if (completed.copyAttempt === input.copyAttempt && completed.jobId === input.jobId) { + return existing; + } + throw new Error('Bounded baseline Copy chunk completion conflict'); + } + if ( + existing.complete || + input.chunkIndex !== existing.completedJobs.length || + existing.activeJob?.copyAttempt !== input.copyAttempt || + existing.activeJob.jobId !== input.jobId + ) { + throw new Error('Bounded baseline Copy chunk completion conflict'); + } + const checkpoint: BoundedBaselineCopyCheckpoint = { + ...existing, + completedJobs: [ + ...existing.completedJobs, + { copyAttempt: input.copyAttempt, jobId: input.jobId }, + ], + }; + delete checkpoint.activeJob; + requireBoundedCheckpointSize(checkpoint); + await transaction.put(key(input.category), checkpoint); + return checkpoint; + }); +} + +export async function completeBoundedBaselineCopy( + storage: DurableObjectStorage, + input: CompleteBoundedBaselineCopyInput, +): Promise { + assertExactKeys( + input, + ['category', 'planSha256', 'proofSha256', 'completedAt'], + 'complete bounded baseline Copy', + ); + if (!hash(input.planSha256) || !hash(input.proofSha256) || !positiveInteger(input.completedAt)) { + throw new Error('Invalid bounded baseline Copy completion'); + } + return storage.transaction(async (transaction) => { + const existing = await baselineCopyCheckpoint(transaction, input.category); + if ( + !existing || + !isBoundedBaselineCopy(existing) || + existing.plan.sha256 !== input.planSha256 || + existing.activeJob || + existing.completedJobs.length !== existing.plan.chunks.length + ) { + throw new Error('Bounded baseline Copy completion conflict'); + } + if (existing.complete) { + if ( + existing.completion?.proofSha256 !== input.proofSha256 || + existing.completion.completedAt !== input.completedAt + ) { + throw new Error('Bounded baseline Copy completion conflict'); + } + return existing; + } + const lastJobId = existing.completedJobs.at(-1)?.jobId; + if (!lastJobId) throw new Error('Bounded baseline Copy has no completed jobs'); + const checkpoint: BoundedBaselineCopyCheckpoint = { + ...existing, + complete: true, + completion: { proofSha256: input.proofSha256, completedAt: input.completedAt, lastJobId }, + }; + requireBoundedCheckpointSize(checkpoint); + await transaction.put(key(input.category), checkpoint); + return checkpoint; + }); +} + +async function requireCurrentPlan( + storage: Pick, + input: BaselineCopyChunkInput, +): Promise { + const checkpoint = await baselineCopyCheckpoint(storage, input.category); + if ( + !checkpoint || + !isBoundedBaselineCopy(checkpoint) || + checkpoint.plan.sha256 !== input.planSha256 + ) { + throw new Error('Bounded baseline Copy plan conflict'); + } + if (input.chunkIndex < 0 || input.chunkIndex >= checkpoint.plan.chunks.length) { + throw new Error('Bounded baseline Copy chunk index is out of range'); + } + return checkpoint; +} + +function validateBeginInput(input: BeginBoundedBaselineCopyInput): void { + assertExactKeys( + input, + [ + 'category', + 'startDay', + 'endDay', + 'startedAt', + 'plan', + ...('legacyFailure' in input ? ['legacyFailure'] : []), + ], + 'begin bounded baseline Copy', + ); + if (!positiveInteger(input.startedAt)) throw new Error('Invalid bounded baseline Copy start'); + const failure = input.legacyFailure; + if ( + failure && + (!positiveInteger(failure.expectedCopyAttempt) || + !positiveInteger(failure.observedAt) || + !hash(failure.providerErrorSha256) || + !hash(failure.journalSha256)) + ) { + throw new Error('Invalid bounded baseline Copy legacy failure'); + } + if (failure) requireJobId(failure.expectedJobId); + if (failure) { + assertExactKeys( + failure, + [ + 'expectedJobId', + 'expectedCopyAttempt', + 'observedAt', + 'providerErrorSha256', + 'journalSha256', + ], + 'bounded baseline Copy legacy failure', + ); + } +} + +function validateChunkInput(input: BaselineCopyChunkInput): void { + assertExactKeys( + input, + ['category', 'planSha256', 'chunkIndex', 'copyAttempt', ...('jobId' in input ? ['jobId'] : [])], + 'bounded baseline Copy chunk input', + ); + if ( + !hash(input.planSha256) || + !positiveInteger(input.copyAttempt) || + !Number.isSafeInteger(input.chunkIndex) + ) { + throw new Error('Invalid bounded baseline Copy chunk input'); + } +} + +function requireJobId(value: string): void { + if (typeof value !== 'string' || !/^[a-zA-Z0-9-]{1,128}$/.test(value)) { + throw new Error('Invalid bounded baseline Copy job receipt'); + } +} + +function positiveInteger(value: number): boolean { + return Number.isSafeInteger(value) && value > 0; +} + +function hash(value: string): boolean { + return /^[0-9a-f]{64}$/.test(value); +} diff --git a/apps/agent-consumer/src/index.ts b/apps/agent-consumer/src/index.ts index 1dd3f51b..d5475b75 100644 --- a/apps/agent-consumer/src/index.ts +++ b/apps/agent-consumer/src/index.ts @@ -1,8 +1,12 @@ import type { BaselineCopyCheckpoint, BaselineMigrationWindow, + BaselineCopyChunkInput, BeginBaselineCopyInput, + BeginBoundedBaselineCopyInput, + CompleteBoundedBaselineCopyInput, ConfirmBaselineCopyInput, + ConfirmBaselineCopyChunkInput, RetryBaselineCopyInput, } from './baseline-copy-migration'; /** @@ -220,28 +224,7 @@ export class AgentIngestion extends WorkerEntrypoint { } export class TraceRecovery extends WorkerEntrypoint { - beginBaselineMigrationWindow(_shardId: string, input: BaselineMigrationWindow) { - return this.env.AGENT_DELIVERY_COORDINATOR.getByName( - 'migration:bounded-agent-ingestion-v1', - ).beginBaselineMigrationWindow(input); - } - getBaselineCopy(orgId: string, input: { category: BaselineCopyCheckpoint['category'] }) { - return this.env.AGENT_DELIVERY_COORDINATOR.getByName( - `baseline:${normalizeAgentShardId(orgId)}`, - ).getBaselineCopy(input); - } - beginBaselineCopy(orgId: string, input: BeginBaselineCopyInput) { - return this.env.AGENT_DELIVERY_COORDINATOR.getByName( - `baseline:${normalizeAgentShardId(orgId)}`, - ).beginBaselineCopy(input); - } - confirmBaselineCopy(orgId: string, input: ConfirmBaselineCopyInput) { - return this.env.AGENT_DELIVERY_COORDINATOR.getByName( - `baseline:${normalizeAgentShardId(orgId)}`, - ).confirmBaselineCopy(input); - } - - async retryBaselineCopy(orgId: string, input: RetryBaselineCopyInput) { + private async requireBaselineMutation(orgId: string) { const normalized = normalizeAgentShardId(orgId); const organization = this.env.AGENT_DELIVERY_COORDINATOR.getByName(`org:${normalized}`); const baseline = this.env.AGENT_DELIVERY_COORDINATOR.getByName(`baseline:${normalized}`); @@ -251,7 +234,7 @@ export class TraceRecovery extends WorkerEntrypoint { getAgentBatcher(this.env, normalized).getIngestionMigrationState(), ]); if (migration !== null) - throw new Error('Baseline Copy retry is forbidden after migration seed'); + throw new Error('Baseline Copy mutation is forbidden after migration seed'); if ( stats.lastDeliverySequence !== 1 || stats.lastSnapshotGeneration !== 0 || @@ -265,16 +248,61 @@ export class TraceRecovery extends WorkerEntrypoint { stats.gateExpiresAtMs !== null || stats.erasureStarted ) { - throw new Error('Baseline Copy retry requires an empty organization coordinator'); + throw new Error('Baseline Copy mutation requires an empty organization coordinator'); } if ( legacy.migrationId !== 'bounded-agent-ingestion-v1' || legacy.queuedRows !== 0 || legacy.flushing !== false ) { - throw new Error('Baseline Copy retry requires frozen, drained legacy ingestion'); + throw new Error('Baseline Copy mutation requires frozen, drained legacy ingestion'); } - return baseline.retryBaselineCopy(input); + return baseline; + } + + beginBaselineMigrationWindow(_shardId: string, input: BaselineMigrationWindow) { + return this.env.AGENT_DELIVERY_COORDINATOR.getByName( + 'migration:bounded-agent-ingestion-v1', + ).beginBaselineMigrationWindow(input); + } + getBaselineCopy(orgId: string, input: { category: BaselineCopyCheckpoint['category'] }) { + return this.env.AGENT_DELIVERY_COORDINATOR.getByName( + `baseline:${normalizeAgentShardId(orgId)}`, + ).getBaselineCopy(input); + } + beginBaselineCopy(orgId: string, input: BeginBaselineCopyInput) { + return this.env.AGENT_DELIVERY_COORDINATOR.getByName( + `baseline:${normalizeAgentShardId(orgId)}`, + ).beginBaselineCopy(input); + } + confirmBaselineCopy(orgId: string, input: ConfirmBaselineCopyInput) { + return this.env.AGENT_DELIVERY_COORDINATOR.getByName( + `baseline:${normalizeAgentShardId(orgId)}`, + ).confirmBaselineCopy(input); + } + + async retryBaselineCopy(orgId: string, input: RetryBaselineCopyInput) { + return (await this.requireBaselineMutation(orgId)).retryBaselineCopy(input); + } + + async beginBoundedBaselineCopy(orgId: string, input: BeginBoundedBaselineCopyInput) { + return (await this.requireBaselineMutation(orgId)).beginBoundedBaselineCopy(input); + } + + async armBoundedBaselineCopyChunk(orgId: string, input: BaselineCopyChunkInput) { + return (await this.requireBaselineMutation(orgId)).armBoundedBaselineCopyChunk(input); + } + + async confirmBoundedBaselineCopyChunk(orgId: string, input: ConfirmBaselineCopyChunkInput) { + return (await this.requireBaselineMutation(orgId)).confirmBoundedBaselineCopyChunk(input); + } + + async completeBoundedBaselineCopyChunk(orgId: string, input: ConfirmBaselineCopyChunkInput) { + return (await this.requireBaselineMutation(orgId)).completeBoundedBaselineCopyChunk(input); + } + + async completeBoundedBaselineCopy(orgId: string, input: CompleteBoundedBaselineCopyInput) { + return (await this.requireBaselineMutation(orgId)).completeBoundedBaselineCopy(input); } inspectGlobalIngestionMigration() { diff --git a/scripts/ingest-recovery/agent-baseline-copy-plan.ts b/scripts/ingest-recovery/agent-baseline-copy-plan.ts new file mode 100644 index 00000000..ccfe5e5a --- /dev/null +++ b/scripts/ingest-recovery/agent-baseline-copy-plan.ts @@ -0,0 +1,62 @@ +import { createHash } from 'node:crypto'; +import type { BaselineCopyPlan } from '../../apps/agent-consumer/src/baseline-copy-contract'; +import { + baselineCopyPlanHashInput, + MAX_BASELINE_COPY_CHUNK_BYTES, + MAX_BASELINE_COPY_CHUNK_DAYS, + MAX_BASELINE_COPY_CHUNK_ROWS, + validateBaselineCopyPlan, +} from '../../apps/agent-consumer/src/baseline-copy-plan'; +import type { BaselineCategoryProof, MigrationWindow } from './agent-migration-proof'; + +export function buildBaselineCopyPlan( + proof: BaselineCategoryProof, + window: MigrationWindow, +): BaselineCopyPlan { + if (proof.dailyStats.length === 0) throw new Error('Cannot plan an empty baseline Copy'); + const chunks: BaselineCopyPlan['chunks'] = []; + let current: BaselineCopyPlan['chunks'][number] | undefined; + for (const stat of proof.dailyStats) { + if ( + stat.rows > MAX_BASELINE_COPY_CHUNK_ROWS || + stat.projectedBytes > MAX_BASELINE_COPY_CHUNK_BYTES + ) { + throw new Error(`Baseline Copy day ${stat.day} exceeds the bounded chunk limit`); + } + const canAppend = + current && + dayNumber(stat.day) - dayNumber(current.startDay) < MAX_BASELINE_COPY_CHUNK_DAYS && + current.rows + stat.rows <= MAX_BASELINE_COPY_CHUNK_ROWS && + current.projectedBytes + stat.projectedBytes <= MAX_BASELINE_COPY_CHUNK_BYTES; + if (!canAppend) { + if (current) chunks.push(current); + current = { + startDay: stat.day, + endDay: stat.day, + rows: stat.rows, + projectedBytes: stat.projectedBytes, + }; + } else { + current.endDay = stat.day; + current.rows += stat.rows; + current.projectedBytes += stat.projectedBytes; + } + } + if (current) chunks.push(current); + const plan: BaselineCopyPlan = { + sha256: '0'.repeat(64), + dailyStats: proof.dailyStats, + chunks, + totalRows: proof.rows, + totalProjectedBytes: proof.dailyStats.reduce((sum, row) => sum + row.projectedBytes, 0), + }; + plan.sha256 = createHash('sha256') + .update(baselineCopyPlanHashInput(proof.category, window.startDay, window.endDay, plan)) + .digest('hex'); + validateBaselineCopyPlan(proof.category, window.startDay, window.endDay, plan); + return plan; +} + +function dayNumber(day: string): number { + return Date.parse(`${day}T00:00:00.000Z`) / 86_400_000; +} diff --git a/scripts/ingest-recovery/agent-baseline-copy-retry-journal.ts b/scripts/ingest-recovery/agent-baseline-copy-retry-journal.ts index bdc2ba56..3beb5f30 100644 --- a/scripts/ingest-recovery/agent-baseline-copy-retry-journal.ts +++ b/scripts/ingest-recovery/agent-baseline-copy-retry-journal.ts @@ -9,12 +9,12 @@ import { writeFileSync, } from 'node:fs'; import { join, resolve } from 'node:path'; -import type { BaselineCopyCheckpoint } from '../../apps/agent-consumer/src/baseline-copy-contract'; +import type { LegacyBaselineCopyCheckpoint } from '../../apps/agent-consumer/src/baseline-copy-contract'; interface BaselineCopyFailureJournalEntry { version: 1; orgId: string; - checkpoint: BaselineCopyCheckpoint; + checkpoint: LegacyBaselineCopyCheckpoint; observedAt: number; providerJob: Record; } diff --git a/scripts/ingest-recovery/agent-baseline-copy.test.ts b/scripts/ingest-recovery/agent-baseline-copy.test.ts index 358b749a..1cdb8a74 100644 --- a/scripts/ingest-recovery/agent-baseline-copy.test.ts +++ b/scripts/ingest-recovery/agent-baseline-copy.test.ts @@ -8,7 +8,7 @@ import { type AgentRecoveryClient, type AgentTinybirdClient, } from './agent-transport'; -import type { BaselineCopyCheckpoint } from '../../apps/agent-consumer/src/baseline-copy-contract'; +import type { LegacyBaselineCopyCheckpoint } from '../../apps/agent-consumer/src/baseline-copy-contract'; const window = { startDay: '2026-09-01', endDay: '2026-09-13' }; const originalFetch = globalThis.fetch; @@ -62,7 +62,7 @@ function restoreEnvironment(name: string, value: string | undefined): void { const failedJobRow = JSON.parse( readFileSync(new URL('./fixtures/tinybird-copy-error-job.json', import.meta.url), 'utf8'), ) as Record; -function fixture(initial: BaselineCopyCheckpoint | null = null) { +function fixture(initial: LegacyBaselineCopyCheckpoint | null = null) { let checkpoint = initial; let loseCopyReceipt = false; let jobApiExpired = false; @@ -71,6 +71,7 @@ function fixture(initial: BaselineCopyCheckpoint | null = null) { let proofOverrides: Record = {}; const requests: string[] = []; const queries: string[] = []; + let boundedInput: Record | undefined; const jobDetails = new Map>([ ['job-one', { job_id: 'job-one', status: 'done' }], ['job-old', { job_id: 'job-old', status: 'done' }], @@ -91,8 +92,12 @@ function fixture(initial: BaselineCopyCheckpoint | null = null) { org: 'org-proof', async call(method: string, input: Record) { if (method === 'getBaselineCopy') return checkpoint; + if (method === 'beginBoundedBaselineCopy') { + boundedInput = input; + throw new Error('bounded transition armed'); + } if (method === 'beginBaselineCopy') { - checkpoint = { ...input, complete: false } as unknown as BaselineCopyCheckpoint; + checkpoint = { ...input, complete: false } as unknown as LegacyBaselineCopyCheckpoint; return { ...checkpoint, created: true }; } if (method === 'confirmBaselineCopy') { @@ -102,7 +107,7 @@ function fixture(initial: BaselineCopyCheckpoint | null = null) { ) { throw new Error('confirmation conflict'); } - checkpoint = { ...checkpoint, ...input } as BaselineCopyCheckpoint; + checkpoint = { ...checkpoint, ...input } as LegacyBaselineCopyCheckpoint; return checkpoint; } if (method === 'retryBaselineCopy') { @@ -171,6 +176,9 @@ function fixture(initial: BaselineCopyCheckpoint | null = null) { get checkpoint() { return checkpoint; }, + get boundedInput() { + return boundedInput; + }, loseReceipt() { loseCopyReceipt = true; }, @@ -194,6 +202,13 @@ function fixture(initial: BaselineCopyCheckpoint | null = null) { const retryJournal = () => mkdtempSync(join(tmpdir(), 'baseline-copy-retry-')); +const toolProof = () => ({ + category: 'tool_events' as const, + rows: 1, + days: [window.startDay], + dailyStats: [{ day: window.startDay, rows: 1, projectedBytes: 10 }], +}); + describe('baseline Copy restart safety', () => { test('reads only verified tinybird.jobs_log columns for failed-job proof', async () => { expect(Object.keys(failedJobRow).sort()).toEqual([ @@ -220,9 +235,12 @@ describe('baseline Copy restart safety', () => { jobId: 'job-failed', complete: false, }); - await runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { - retryJournalRoot: journal, - }); + await expect( + runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { + retryJournalRoot: journal, + sourceProof: toolProof(), + }), + ).rejects.toThrow('bounded transition armed'); const query = f.queries.find((candidate) => candidate.includes(' AS target_rows'))!; expect(query).toContain('job_metadata'); expect(query).not.toMatch(/^\s+pipe_name,/m); @@ -232,30 +250,40 @@ describe('baseline Copy restart safety', () => { }); test('a completed Copy is reused without writing again', async () => { - const f = fixture(); - expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toBe('job-one'); + const f = fixture({ + category: 'messages', + ...window, + startedAt: 1, + copyAttempt: 1, + jobId: 'job-one', + complete: true, + }); + expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toEqual(['job-one']); expect(f.checkpoint?.complete).toBe(true); - expect(f.requests[0]).toContain('copy_attempt='); - expect(f.requests[0]).not.toContain('on_demand_compute'); const count = f.requests.length; await runBaselineCopy(f.tb, f.recovery, 'messages', window); expect(f.requests).toHaveLength(count); }); - test('a lost submission response never causes a blind duplicate Copy', async () => { - const f = fixture(); - f.loseReceipt(); + test('an unresolved durable intent never causes a blind duplicate Copy', async () => { + const f = fixture({ + category: 'messages', + ...window, + startedAt: 1, + copyAttempt: 1, + complete: false, + }); await expect(runBaselineCopy(f.tb, f.recovery, 'messages', window)).rejects.toThrow( - 'Connection lost', + 'unresolved', ); expect(f.checkpoint?.jobId).toBeUndefined(); await expect(runBaselineCopy(f.tb, f.recovery, 'messages', window)).rejects.toThrow( 'unresolved', ); - expect(f.requests).toHaveLength(1); + expect(f.requests).toHaveLength(0); f.jobs([{ job_id: 'job-recovered', status: 'done' }]); - expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toBe('job-recovered'); + expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toEqual(['job-recovered']); expect(f.checkpoint?.complete).toBe(true); - expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(1); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); expect(f.queries[f.queries.length - 1]).toContain("'copy_attempt'"); expect(f.queries[f.queries.length - 1]).not.toContain('created_at >='); }); @@ -271,7 +299,7 @@ describe('baseline Copy restart safety', () => { f.expireJobApi(); f.jobs([{ job_id: 'job-old', status: 'done' }]); - expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toBe('job-old'); + expect(await runBaselineCopy(f.tb, f.recovery, 'messages', window)).toEqual(['job-old']); expect(f.queries[f.queries.length - 1]).toContain("job_id = 'job-old'"); expect(f.checkpoint?.complete).toBe(true); }); @@ -293,7 +321,7 @@ describe('baseline Copy restart safety', () => { expect(f.requests).toHaveLength(0); }); - test('retries one exact failed job on dedicated compute and preserves private evidence', async () => { + test('transitions a verified failed whole-window job to a bounded plan without another whole Copy', async () => { const journal = retryJournal(); try { const f = fixture({ @@ -304,35 +332,27 @@ describe('baseline Copy restart safety', () => { jobId: 'job-failed', complete: false, }); - - expect( - await runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { + const sourceProof = toolProof(); + await expect( + runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { retryJournalRoot: journal, + sourceProof, }), - ).toBe('job-retry'); - const copyRequests = f.requests.filter((path) => path.includes('/copy?')); - expect(copyRequests).toHaveLength(1); - expect(copyRequests[0]).toContain('on_demand_compute=true'); - expect(copyRequests[0]).toContain('start_day=2026-09-01'); - expect(copyRequests[0]).toContain('end_day=2026-09-13'); - expect(f.checkpoint).toMatchObject({ - jobId: 'job-retry', - complete: true, - failedAttempt: { - copyAttempt: 1, - jobId: 'job-failed', - status: 'error', - }, + ).rejects.toThrow('bounded transition armed'); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + expect(f.boundedInput).toMatchObject({ + category: 'tool_events', + startDay: window.startDay, + endDay: window.endDay, + legacyFailure: { expectedJobId: 'job-failed', expectedCopyAttempt: 1 }, }); - const entries = readdirSync(journal); - expect(entries).toEqual(['tool_events-job-failed.json']); - const preserved = readFileSync(join(journal, entries[0]!), 'utf8'); - expect(preserved).toContain('private provider SQL'); - expect(f.checkpoint?.failedAttempt?.journalSha256).toMatch(/^[0-9a-f]{64}$/); + expect(readdirSync(journal)).toEqual(['tool_events-job-failed.json']); + expect(readFileSync(join(journal, 'tool_events-job-failed.json'), 'utf8')).toContain( + 'private provider SQL', + ); expect(guardRequests.filter((url) => url.includes('collector.trace-flow.dev'))).toHaveLength( 2, ); - expect(guardRequests.filter((url) => url.includes('/queues?'))).toHaveLength(2); expect(f.queries.find((query) => query.includes(' AS target_rows'))).toContain( 'FROM agent_tool_event_fact_versions FINAL WHERE OrgId', ); @@ -341,44 +361,7 @@ describe('baseline Copy restart safety', () => { } }); - test('recovers a lost dedicated-compute receipt by the exact new attempt', async () => { - const journal = retryJournal(); - try { - const f = fixture({ - category: 'tool_events', - ...window, - startedAt: 1, - copyAttempt: 1, - jobId: 'job-failed', - complete: false, - }); - f.loseReceipt(); - await expect( - runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { - retryJournalRoot: journal, - }), - ).rejects.toThrow('Connection lost'); - const retryAttempt = f.checkpoint!.copyAttempt; - expect(f.checkpoint?.jobId).toBeUndefined(); - f.jobs([{ job_id: 'job-retry', status: 'done' }]); - - expect( - await runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { - retryJournalRoot: journal, - }), - ).toBe('job-retry'); - expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(1); - const recoveryQuery = [...f.queries] - .reverse() - .find((query) => query.includes('copy_attempt') && !query.includes(' AS target_rows')); - expect(recoveryQuery).toContain(`'${retryAttempt}'`); - expect(f.checkpoint?.failedAttempt?.jobId).toBe('job-failed'); - } finally { - rmSync(journal, { recursive: true }); - } - }); - - test('refuses terminal non-error jobs and nonempty category targets', async () => { + test('refuses terminal non-error jobs, nonempty targets, and mismatched failed metadata', async () => { const cancelled = fixture({ category: 'tool_events', ...window, @@ -391,7 +374,6 @@ describe('baseline Copy restart safety', () => { await expect( runBaselineCopy(cancelled.tb, cancelled.recovery, 'tool_events', window), ).rejects.toThrow('terminal status cancelled; retry refused'); - expect(cancelled.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); const journal = retryJournal(); try { @@ -407,17 +389,10 @@ describe('baseline Copy restart safety', () => { await expect( runBaselineCopy(populated.tb, populated.recovery, 'tool_events', window, { retryJournalRoot: journal, + sourceProof: toolProof(), }), ).rejects.toThrow('empty organization category target'); - expect(populated.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); - } finally { - rmSync(journal, { recursive: true }); - } - }); - test('requires exact failed-job metadata and never retries a second failure', async () => { - const journal = retryJournal(); - try { const mismatch = fixture({ category: 'tool_events', ...window, @@ -438,19 +413,28 @@ describe('baseline Copy restart safety', () => { await expect( runBaselineCopy(mismatch.tb, mismatch.recovery, 'tool_events', window, { retryJournalRoot: journal, + sourceProof: toolProof(), }), ).rejects.toThrow('does not match its durable intent'); + expect(populated.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); expect(mismatch.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + } finally { + rmSync(journal, { recursive: true }); + } + }); - const exhausted = fixture({ + test('preserves the original failed attempt when transitioning a second failed receipt', async () => { + const journal = retryJournal(); + try { + const f = fixture({ category: 'tool_events', ...window, startedAt: 1, copyAttempt: 2, - jobId: 'job-retry', + jobId: 'job-failed', failedAttempt: { copyAttempt: 1, - jobId: 'job-failed', + jobId: 'job-original', status: 'error', observedAt: 2, providerErrorSha256: 'a'.repeat(64), @@ -458,13 +442,26 @@ describe('baseline Copy restart safety', () => { }, complete: false, }); - exhausted.setJob('job-retry', { status: 'error', error: 'Retry timed out' }); + f.setProof({ + job_metadata: JSON.stringify({ + ...JSON.parse(failedJobRow.job_metadata as string), + parameters: { + ...JSON.parse(failedJobRow.job_metadata as string).parameters, + copy_attempt: '2', + }, + }), + }); await expect( - runBaselineCopy(exhausted.tb, exhausted.recovery, 'tool_events', window, { + runBaselineCopy(f.tb, f.recovery, 'tool_events', window, { retryJournalRoot: journal, + sourceProof: toolProof(), }), - ).rejects.toThrow('retry limit reached'); - expect(exhausted.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + ).rejects.toThrow('bounded transition armed'); + expect(f.checkpoint?.failedAttempt?.jobId).toBe('job-original'); + expect(f.boundedInput).toMatchObject({ + legacyFailure: { expectedJobId: 'job-failed', expectedCopyAttempt: 2 }, + }); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); } finally { rmSync(journal, { recursive: true }); } diff --git a/scripts/ingest-recovery/agent-baseline-copy.ts b/scripts/ingest-recovery/agent-baseline-copy.ts index 01d0b185..522f00df 100644 --- a/scripts/ingest-recovery/agent-baseline-copy.ts +++ b/scripts/ingest-recovery/agent-baseline-copy.ts @@ -1,21 +1,31 @@ -import type { BaselineCopyCheckpoint } from '../../apps/agent-consumer/src/baseline-copy-contract'; +import type { + BaselineCopyCheckpoint, + LegacyBaselineCopyCheckpoint, +} from '../../apps/agent-consumer/src/baseline-copy-contract'; +import { isBoundedBaselineCopy } from '../../apps/agent-consumer/src/baseline-copy-plan'; import { FACT_VERSION_DATASOURCES } from '../../apps/agent-consumer/src/delivery-write'; import { quote } from './agent-data'; -import { preserveBaselineCopyFailure } from './agent-baseline-copy-retry-journal'; import type { AgentRecoveryClient, AgentTinybirdClient } from './agent-transport'; +import { waitForMigrationCopy } from './agent-migration-runtime'; +import type { BaselineCategoryProof } from './agent-migration-proof'; import { - requireAgentProducerMaintenance, - requireDrainedAgentQueues, - waitForMigrationCopy, -} from './agent-migration-runtime'; + beginFreshBoundedBaselineCopy, + runBoundedBaselineCopy, + transitionFailedBaselineCopy, +} from './agent-bounded-baseline-copy'; + +interface BaselineCopyOptions { + retryJournalRoot?: string; + sourceProof?: BaselineCategoryProof; +} export async function runBaselineCopy( tb: AgentTinybirdClient, recovery: AgentRecoveryClient, category: BaselineCopyCheckpoint['category'], window: { startDay: string; endDay: string }, - options: { retryJournalRoot?: string } = {}, -): Promise { + options: BaselineCopyOptions = {}, +): Promise { const pipe = `repair_agent_${category}_versions_baseline`; let checkpoint = (await recovery.call('getBaselineCopy', { category, @@ -28,93 +38,61 @@ export async function runBaselineCopy( 'Baseline Copy retention window changed; existing intent must be reconciled before resuming', ); } + if (checkpoint && isBoundedBaselineCopy(checkpoint)) { + return runBoundedBaselineCopy(tb, recovery, checkpoint, options); + } if (!checkpoint) { - const startedAt = Date.now(); - const claimed = (await recovery.call('beginBaselineCopy', { - category, - ...window, - startedAt, - copyAttempt: startedAt, - })) as BaselineCopyCheckpoint & { created: boolean }; - checkpoint = claimed; - if (claimed.created) { - checkpoint = await startBaselineCopy(tb, recovery, pipe, claimed, false); - } + const proof = requireSourceProof(options.sourceProof, category); + const bounded = await beginFreshBoundedBaselineCopy(tb, recovery, proof, window); + return runBoundedBaselineCopy(tb, recovery, bounded, options); } + const existing = checkpoint as LegacyBaselineCopyCheckpoint; + if (existing.complete) { + if (!existing.jobId) throw new Error('Completed baseline Copy omitted its job receipt'); + return [existing.jobId]; + } + let legacy = existing; while (true) { - if (!checkpoint.jobId) - checkpoint = await recoverBaselineCopyReceipt(tb, recovery, pipe, checkpoint); - if (!checkpoint.jobId) throw new Error('Baseline Copy checkpoint omitted its job receipt'); - if (checkpoint.complete) return checkpoint.jobId; - const jobId = checkpoint.jobId; + if (!legacy.jobId) legacy = await recoverBaselineCopyReceipt(tb, recovery, pipe, legacy); + if (!legacy.jobId) throw new Error('Baseline Copy checkpoint omitted its job receipt'); + if (legacy.complete) return [legacy.jobId]; + const jobId = legacy.jobId; const terminal = await waitForMigrationCopy(tb, jobId); if (terminal.status === 'done') { - checkpoint = (await recovery.call('confirmBaselineCopy', { + legacy = (await recovery.call('confirmBaselineCopy', { category, - copyAttempt: checkpoint.copyAttempt, + copyAttempt: legacy.copyAttempt, jobId, complete: true, - })) as BaselineCopyCheckpoint; - return checkpoint.jobId!; + })) as LegacyBaselineCopyCheckpoint; + return [legacy.jobId!]; } if (terminal.status !== 'error') { throw new Error( `Baseline Copy job reached terminal status ${terminal.status}; retry refused`, ); } - if (checkpoint.failedAttempt) { - throw new Error('Baseline Copy dedicated-compute retry failed; retry limit reached'); - } - checkpoint = await armBaselineCopyRetry( + const proof = requireSourceProof(options.sourceProof, category); + const bounded = await transitionFailedBaselineCopy( tb, recovery, - pipe, - { ...checkpoint, jobId }, + { ...legacy, jobId }, terminal, - options.retryJournalRoot, + proof, + options, + () => readFailedBaselineCopy(tb, recovery, pipe, { ...legacy, jobId }), ); - checkpoint = await startBaselineCopy(tb, recovery, pipe, checkpoint, true); + return runBoundedBaselineCopy(tb, recovery, bounded, options); } } -async function startBaselineCopy( - tb: AgentTinybirdClient, - recovery: AgentRecoveryClient, - pipe: string, - checkpoint: BaselineCopyCheckpoint, - onDemandCompute: boolean, -): Promise { - const params = new URLSearchParams({ - org_id: recovery.org, - start_day: checkpoint.startDay, - end_day: checkpoint.endDay, - copy_attempt: String(checkpoint.copyAttempt), - _mode: 'append', - }); - if (onDemandCompute) params.set('on_demand_compute', 'true'); - const response: unknown = await tb.request(`/v0/pipes/${pipe}/copy?${params.toString()}`, ''); - const receipt = - response && typeof response === 'object' && 'job' in response ? response.job : null; - const jobId: unknown = - receipt && typeof receipt === 'object' && 'job_id' in receipt ? receipt.job_id : null; - if (typeof jobId !== 'string') { - throw new Error('Baseline Copy returned no job receipt; durable intent remains unresolved'); - } - return recovery.call('confirmBaselineCopy', { - category: checkpoint.category, - copyAttempt: checkpoint.copyAttempt, - jobId, - complete: false, - }) as Promise; -} - async function recoverBaselineCopyReceipt( tb: AgentTinybirdClient, recovery: AgentRecoveryClient, pipe: string, - checkpoint: BaselineCopyCheckpoint, -): Promise { + checkpoint: LegacyBaselineCopyCheckpoint, +): Promise { const result = await tb.sql(`SELECT job_id, status FROM tinybird.jobs_log WHERE job_type IN ('copy', 'copy_from_branch') AND JSONExtractString(job_metadata, 'pipe_name') = ${quote(pipe)} @@ -134,55 +112,14 @@ async function recoverBaselineCopyReceipt( copyAttempt: checkpoint.copyAttempt, jobId: job.job_id, complete: false, - }) as Promise; -} - -async function armBaselineCopyRetry( - tb: AgentTinybirdClient, - recovery: AgentRecoveryClient, - pipe: string, - checkpoint: BaselineCopyCheckpoint & { jobId: string }, - providerJob: Record, - retryJournalRoot: string | undefined, -): Promise { - if (!retryJournalRoot) { - throw new Error('Baseline Copy retry requires --retry-journal '); - } - await requireAgentProducerMaintenance(); - await requireDrainedAgentQueues(); - const job = await readFailedBaselineCopy(tb, recovery, pipe, checkpoint); - - const observedAt = Date.now(); - const evidence = preserveBaselineCopyFailure(retryJournalRoot, { - version: 1, - orgId: recovery.org, - checkpoint, - observedAt, - providerJob: { ...providerJob, ...job }, - }); - await requireAgentProducerMaintenance(); - await requireDrainedAgentQueues(); - const freshJob = await readFailedBaselineCopy(tb, recovery, pipe, checkpoint); - if (freshJob.error !== job.error) { - throw new Error('Baseline Copy provider error changed while arming its retry'); - } - const nextCopyAttempt = Math.max(Date.now(), checkpoint.copyAttempt + 1); - return recovery.call('retryBaselineCopy', { - category: checkpoint.category, - expectedJobId: checkpoint.jobId, - expectedCopyAttempt: checkpoint.copyAttempt, - nextCopyAttempt, - observedAt: evidence.observedAt, - providerErrorSha256: evidence.providerErrorSha256, - journalSha256: evidence.journalSha256, - }) as Promise; + }) as Promise; } -async function readFailedBaselineCopy( +export async function readFailedBaselineCopy( tb: AgentTinybirdClient, recovery: AgentRecoveryClient, pipe: string, - checkpoint: BaselineCopyCheckpoint & { jobId: string }, + checkpoint: LegacyBaselineCopyCheckpoint & { jobId: string }, ): Promise & { error: string }> { const target = FACT_VERSION_DATASOURCES[checkpoint.category]; const proof = ( @@ -223,6 +160,16 @@ async function readFailedBaselineCopy( return { ...job, metadata, error: job.error }; } +function requireSourceProof( + proof: BaselineCategoryProof | undefined, + category: BaselineCopyCheckpoint['category'], +): BaselineCategoryProof { + if (!proof || proof.category !== category || proof.rows <= 0) { + throw new Error('Bounded baseline Copy requires an exact source plan'); + } + return proof; +} + type BaselineCopyJobMetadata = Record & { parameters: Record; }; diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts new file mode 100644 index 00000000..421511c6 --- /dev/null +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts @@ -0,0 +1,328 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, test } from 'bun:test'; +import type { + BaselineCopyCheckpoint, + BoundedBaselineCopyCheckpoint, +} from '../../apps/agent-consumer/src/baseline-copy-contract'; +import { buildBaselineCopyPlan } from './agent-baseline-copy-plan'; +import { runBoundedBaselineCopy } from './agent-bounded-baseline-copy'; +import { verifyChunkSource, verifyCompletedChunkTarget } from './agent-bounded-baseline-proof'; +import { chunkAgentDayRange } from './agent-day-chunks'; +import { retainedMigrationWindow } from './agent-migration-proof'; +import type { AgentRecoveryClient, AgentTinybirdClient } from './agent-transport'; + +const originalFetch = globalThis.fetch; +const originalAccountId = process.env.CLOUDFLARE_ACCOUNT_ID; +const originalApiToken = process.env.CLOUDFLARE_API_TOKEN; +const guardRequests: string[] = []; + +beforeAll(() => { + process.env.CLOUDFLARE_ACCOUNT_ID = 'account-fixture'; + process.env.CLOUDFLARE_API_TOKEN = 'token-fixture'; + globalThis.fetch = Object.assign( + async (input: string | URL | Request) => { + const url = String(input); + guardRequests.push(url); + if (url === 'https://collector.trace-flow.dev/v1/ingest') { + return Response.json({ error: 'ingestion_maintenance' }, { status: 503 }); + } + if (url.includes('/queues?')) { + return Response.json({ + success: true, + result: [ + { queue_name: 'agent-ingest-prod', queue_id: 'ingest-id' }, + { queue_name: 'agent-ingest-dlq-prod', queue_id: 'dlq-id' }, + ], + }); + } + if (url.includes('/queues/')) { + return Response.json({ success: true, result: { backlog_count: 0 } }); + } + throw new Error(`Unexpected guard request ${url}`); + }, + { preconnect: originalFetch.preconnect }, + ); +}); + +beforeEach(() => { + guardRequests.length = 0; +}); + +afterAll(() => { + globalThis.fetch = originalFetch; + restore('CLOUDFLARE_ACCOUNT_ID', originalAccountId); + restore('CLOUDFLARE_API_TOKEN', originalApiToken); +}); + +function restore(name: string, value: string | undefined): void { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; +} + +function checkpoint(): BoundedBaselineCopyCheckpoint { + const window = retainedMigrationWindow(); + const day = window.endDay; + const proof = { + category: 'tool_events' as const, + rows: 2, + days: [day], + dailyStats: [{ day, rows: 2, projectedBytes: 20 }], + }; + return { + mode: 'bounded', + category: proof.category, + startDay: day, + endDay: day, + startedAt: 1, + plan: buildBaselineCopyPlan(proof, { startDay: day, endDay: day }), + completedJobs: [], + complete: false, + }; +} + +function fixture(initial = checkpoint()) { + let state: BaselineCopyCheckpoint = structuredClone(initial); + let jobs: Record[] = []; + let terminal: Record = { job_id: 'job-chunk', status: 'done' }; + let loseResponse = false; + const requests: string[] = []; + const queries: string[] = []; + const calls: string[] = []; + const verificationChunk = chunkAgentDayRange(initial)[0]!; + const recovery = { + org: 'org-proof', + async call(method: string, input: Record) { + calls.push(method); + if (method === 'armBoundedBaselineCopyChunk') { + const bounded = state as BoundedBaselineCopyCheckpoint; + state = { ...bounded, activeJob: { copyAttempt: input.copyAttempt as number } }; + return { ...state, created: true }; + } + if (method === 'confirmBoundedBaselineCopyChunk') { + const bounded = state as BoundedBaselineCopyCheckpoint; + state = { + ...bounded, + activeJob: { copyAttempt: input.copyAttempt as number, jobId: input.jobId as string }, + }; + return state; + } + if (method === 'completeBoundedBaselineCopyChunk') { + const bounded = state as BoundedBaselineCopyCheckpoint; + const { activeJob: _active, ...rest } = bounded; + state = { + ...rest, + completedJobs: [ + ...bounded.completedJobs, + { copyAttempt: input.copyAttempt as number, jobId: input.jobId as string }, + ], + }; + return state; + } + if (method === 'completeBoundedBaselineCopy') { + state = { ...state, complete: true } as BoundedBaselineCopyCheckpoint; + return state; + } + throw new Error(`Unexpected recovery call ${method}`); + }, + } as unknown as AgentRecoveryClient; + const tb = { + async request(path: string) { + requests.push(path); + if (path.startsWith('/v0/jobs/')) return terminal; + if (loseResponse) throw new Error('lost chunk response'); + return { job: { job_id: 'job-chunk' } }; + }, + async sql(query: string) { + queries.push(query); + if (query.includes('tinybird.jobs_log')) return { data: jobs, meta: [] }; + if (query.includes('projected_bytes')) { + return { data: [{ rows: 2, projected_bytes: 20 }], meta: [] }; + } + if (query.includes('invalid_rows')) { + return { data: [{ rows: 2, invalid_rows: 0 }], meta: [] }; + } + if (query.includes('invalid_metadata')) { + const rows = query.includes(`toDateTime('${verificationChunk.startDay}')`) ? 2 : 0; + return { + data: [ + { + source_rows: rows, + target_rows: rows, + invalid_metadata: 0, + missing_target: 0, + unexpected_target: 0, + }, + ], + meta: [], + }; + } + if (query.includes('source_index')) { + const rows = query.includes(`toDateTime('${verificationChunk.startDay}')`) ? 2 : 0; + return { + data: [{ source_rows: rows, target_rows: rows, missing_target: 0, unexpected_target: 0 }], + meta: [], + }; + } + return { data: [{ rows: 0 }], meta: [] }; + }, + } as unknown as AgentTinybirdClient; + return { + tb, + recovery, + requests, + queries, + calls, + state: () => state as BoundedBaselineCopyCheckpoint, + jobs: (value: Record[]) => (jobs = value), + loseResponse: () => (loseResponse = true), + terminal: (value: Record) => (terminal = value), + }; +} + +describe('bounded baseline Copy operator', () => { + test('uses the immutable global window plus outer chunk bounds on shared compute', async () => { + const f = fixture(); + const result = await runBoundedBaselineCopy(f.tb, f.recovery, f.state(), {}); + expect(result).toEqual(['job-chunk']); + const copy = f.requests.find((path) => path.includes('/copy?'))!; + expect(copy).toContain(`start_day=${f.state().startDay}`); + expect(copy).toContain(`end_day=${f.state().endDay}`); + expect(copy).toContain(`chunk_start_day=${f.state().plan.chunks[0]!.startDay}`); + expect(copy).toContain(`chunk_end_day=${f.state().plan.chunks[0]!.endDay}`); + expect(copy).not.toContain('on_demand_compute'); + expect(f.state()).toMatchObject({ complete: true, completedJobs: [{ jobId: 'job-chunk' }] }); + }); + + test('recovers exactly one lost receipt and refuses missing or ambiguous matches', async () => { + for (const matches of [[], [{ job_id: 'one' }, { job_id: 'two' }]]) { + const active = checkpoint(); + active.activeJob = { copyAttempt: 123 }; + const f = fixture(active); + f.jobs(matches); + await expect(runBoundedBaselineCopy(f.tb, f.recovery, f.state(), {})).rejects.toThrow( + 'duplicate submission refused', + ); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + expect(f.state().activeJob).toEqual({ copyAttempt: 123 }); + const query = f.queries.find((value) => value.includes('tinybird.jobs_log'))!; + expect(query).toContain("job_type IN ('copy','copy_from_branch')"); + expect(query).toContain("'chunk_start_day'"); + expect(query).toContain("'chunk_end_day'"); + expect(query).toContain("'_mode')='append'"); + } + }); + + test('leaves terminal partial failures active and never submits another chunk', async () => { + const active = checkpoint(); + active.activeJob = { copyAttempt: 123, jobId: 'job-error' }; + const f = fixture(active); + f.terminal({ job_id: 'job-error', status: 'error', error: 'private provider error' }); + await expect(runBoundedBaselineCopy(f.tb, f.recovery, f.state(), {})).rejects.toThrow( + 'terminal status error', + ); + expect(f.state().activeJob).toEqual({ copyAttempt: 123, jobId: 'job-error' }); + expect(f.calls).not.toContain('completeBoundedBaselineCopyChunk'); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + }); + + test('clips retained source proof by UTC day while keeping the original Copy window', async () => { + const retained = retainedMigrationWindow(); + const previous = new Date(`${retained.startDay}T00:00:00.000Z`); + previous.setUTCDate(previous.getUTCDate() - 1); + const previousDay = previous.toISOString().slice(0, 10); + const full = { startDay: previousDay, endDay: retained.startDay }; + const proof = { + category: 'tool_events' as const, + rows: 3, + days: [previousDay, retained.startDay], + dailyStats: [ + { day: previousDay, rows: 1, projectedBytes: 10 }, + { day: retained.startDay, rows: 2, projectedBytes: 20 }, + ], + }; + const bounded: BoundedBaselineCopyCheckpoint = { + mode: 'bounded', + category: proof.category, + ...full, + startedAt: 1, + plan: buildBaselineCopyPlan(proof, full), + completedJobs: [], + complete: false, + }; + const queries: string[] = []; + const tb = { + async sql(query: string) { + queries.push(query); + return { data: [{ rows: 2, projected_bytes: 20 }], meta: [] }; + }, + } as unknown as AgentTinybirdClient; + const recovery = { org: 'org-proof' } as AgentRecoveryClient; + await expect(verifyChunkSource(tb, recovery, bounded, bounded.plan.chunks[0]!)).resolves.toBe( + 2, + ); + expect(queries[0]).toContain(`toDateTime('${bounded.startDay}')`); + expect(queries[0]).toContain(`toDateTime('${retained.startDay}')`); + }); + + test('ignores physically lingering TTL rows for a fully expired chunk', async () => { + const full = { startDay: '2020-01-01', endDay: '2020-01-01' }; + const proof = { + category: 'tool_events' as const, + rows: 1, + days: [full.startDay], + dailyStats: [{ day: full.startDay, rows: 1, projectedBytes: 10 }], + }; + const bounded: BoundedBaselineCopyCheckpoint = { + mode: 'bounded', + category: proof.category, + ...full, + startedAt: 1, + plan: buildBaselineCopyPlan(proof, full), + completedJobs: [], + complete: false, + }; + const tb = { + async sql() { + throw new Error('expired physical rows must not be queried'); + }, + } as unknown as AgentTinybirdClient; + const recovery = { org: 'org-proof' } as AgentRecoveryClient; + await expect(verifyChunkSource(tb, recovery, bounded, bounded.plan.chunks[0]!)).resolves.toBe( + 0, + ); + await expect( + verifyCompletedChunkTarget(tb, recovery, bounded, bounded.plan.chunks[0]!), + ).resolves.toBeUndefined(); + }); + + test('rejects malformed Tinybird aggregates instead of coercing them to zero', async () => { + const bounded = checkpoint(); + for (const value of [null, '', false, undefined]) { + const tb = { + async sql() { + return { data: [{ rows: value, projected_bytes: value }], meta: [] }; + }, + } as unknown as AgentTinybirdClient; + await expect( + verifyChunkSource( + tb, + { org: 'org-proof' } as AgentRecoveryClient, + bounded, + bounded.plan.chunks[0]!, + ), + ).rejects.toThrow('Invalid bounded baseline retained source rows'); + } + const tb = { + async sql() { + return { data: [{ rows: 0, projected_bytes: null }], meta: [] }; + }, + } as unknown as AgentTinybirdClient; + await expect( + verifyChunkSource( + tb, + { org: 'org-proof' } as AgentRecoveryClient, + bounded, + bounded.plan.chunks[0]!, + ), + ).rejects.toThrow('Invalid bounded baseline retained source bytes'); + }); +}); diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts new file mode 100644 index 00000000..62fd2ced --- /dev/null +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts @@ -0,0 +1,269 @@ +import { createHash } from 'node:crypto'; +import type { + BaselineCopyCheckpoint, + BoundedBaselineCopyCheckpoint, + LegacyBaselineCopyCheckpoint, +} from '../../apps/agent-consumer/src/baseline-copy-contract'; +import { isBoundedBaselineCopy } from '../../apps/agent-consumer/src/baseline-copy-plan'; +import { quote } from './agent-data'; +import { preserveBaselineCopyFailure } from './agent-baseline-copy-retry-journal'; +import { buildBaselineCopyPlan } from './agent-baseline-copy-plan'; +import { + intersectMigrationWindows, + retainedMigrationWindow, + verifyBaseline, + type BaselineCategoryProof, + type MigrationWindow, +} from './agent-migration-proof'; +import { + requireAgentProducerMaintenance, + requireDrainedAgentQueues, + waitForMigrationCopy, +} from './agent-migration-runtime'; +import type { AgentRecoveryClient, AgentTinybirdClient } from './agent-transport'; +import { + preserveChunkFailure, + requireEmptyCategoryTarget, + requireEmptyChunkTarget, + verifyChunkSource, + verifyCompletedChunkTarget, +} from './agent-bounded-baseline-proof'; + +interface BoundedCopyOptions { + retryJournalRoot?: string; +} + +export async function beginFreshBoundedBaselineCopy( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + proof: BaselineCategoryProof, + window: MigrationWindow, +): Promise { + const plan = buildBaselineCopyPlan(proof, window); + await freshExternalGuards(); + await requireEmptyCategoryTarget(tb, recovery, proof.category); + await freshExternalGuards(); + await requireEmptyCategoryTarget(tb, recovery, proof.category); + return recovery.call('beginBoundedBaselineCopy', { + category: proof.category, + ...window, + startedAt: Date.now(), + plan, + }) as Promise; +} + +export async function transitionFailedBaselineCopy( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + checkpoint: LegacyBaselineCopyCheckpoint & { jobId: string }, + providerJob: Record, + proof: BaselineCategoryProof, + options: BoundedCopyOptions, + readFailedJob: () => Promise & { error: string }>, +): Promise { + if (!options.retryJournalRoot) { + throw new Error('Bounded baseline Copy recovery requires --retry-journal '); + } + const plan = buildBaselineCopyPlan(proof, checkpoint); + await freshExternalGuards(); + const job = await readFailedJob(); + const evidence = preserveBaselineCopyFailure(options.retryJournalRoot, { + version: 1, + orgId: recovery.org, + checkpoint, + observedAt: Date.now(), + providerJob: { ...providerJob, ...job }, + }); + await freshExternalGuards(); + const freshJob = await readFailedJob(); + if (freshJob.error !== job.error) { + throw new Error('Baseline Copy provider error changed while arming bounded recovery'); + } + return recovery.call('beginBoundedBaselineCopy', { + category: checkpoint.category, + startDay: checkpoint.startDay, + endDay: checkpoint.endDay, + startedAt: checkpoint.startedAt, + plan, + legacyFailure: { + expectedJobId: checkpoint.jobId, + expectedCopyAttempt: checkpoint.copyAttempt, + observedAt: evidence.observedAt, + providerErrorSha256: evidence.providerErrorSha256, + journalSha256: evidence.journalSha256, + }, + }) as Promise; +} + +export async function runBoundedBaselineCopy( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + options: BoundedCopyOptions, +): Promise { + if (checkpoint.complete) return checkpoint.completedJobs.map((job) => job.jobId); + const pipe = `repair_agent_${checkpoint.category}_versions_baseline`; + while (checkpoint.completedJobs.length < checkpoint.plan.chunks.length) { + const chunkIndex = checkpoint.completedJobs.length; + const chunk = checkpoint.plan.chunks[chunkIndex]!; + if (!checkpoint.activeJob) { + await freshExternalGuards(); + await verifyChunkSource(tb, recovery, checkpoint, chunk); + await requireEmptyChunkTarget(tb, recovery, checkpoint, chunk); + await freshExternalGuards(); + const copyAttempt = Date.now(); + const armed = (await recovery.call('armBoundedBaselineCopyChunk', { + category: checkpoint.category, + planSha256: checkpoint.plan.sha256, + chunkIndex, + copyAttempt, + })) as BoundedBaselineCopyCheckpoint & { created: boolean }; + checkpoint = armed; + if (armed.created) { + checkpoint = await startChunk(tb, recovery, pipe, checkpoint, chunkIndex); + } + } + if (!checkpoint.activeJob) throw new Error('Bounded baseline Copy omitted its active intent'); + if (!checkpoint.activeJob.jobId) { + checkpoint = await recoverChunkReceipt(tb, recovery, pipe, checkpoint, chunkIndex); + } + const active = checkpoint.activeJob; + if (!active?.jobId) throw new Error('Bounded baseline Copy omitted its job receipt'); + const received = { copyAttempt: active.copyAttempt, jobId: active.jobId }; + const terminal = await waitForMigrationCopy(tb, active.jobId); + if (terminal.status !== 'done') { + if (terminal.status === 'error' && options.retryJournalRoot) { + preserveChunkFailure(options.retryJournalRoot, recovery, checkpoint, received, terminal); + } + throw new Error(`Bounded baseline Copy chunk reached terminal status ${terminal.status}`); + } + await freshExternalGuards(); + await verifyChunkSource(tb, recovery, checkpoint, chunk); + await verifyCompletedChunkTarget(tb, recovery, checkpoint, chunk); + await freshExternalGuards(); + checkpoint = (await recovery.call('completeBoundedBaselineCopyChunk', { + category: checkpoint.category, + planSha256: checkpoint.plan.sha256, + chunkIndex, + copyAttempt: active.copyAttempt, + jobId: active.jobId, + })) as BoundedBaselineCopyCheckpoint; + } + + const retained = intersectMigrationWindows(checkpoint, retainedMigrationWindow()); + const retainedStats = checkpoint.plan.dailyStats.filter( + (stat) => stat.day >= retained.startDay && stat.day <= retained.endDay, + ); + const proof = { + category: checkpoint.category, + rows: retainedStats.reduce((sum, stat) => sum + stat.rows, 0), + days: retainedStats.map((stat) => stat.day), + dailyStats: retainedStats, + }; + await verifyBaseline(tb, recovery.org, retained, proof, checkpoint); + const proofSha256 = createHash('sha256') + .update( + JSON.stringify({ + category: checkpoint.category, + retained, + proof, + plan: checkpoint.plan.sha256, + }), + ) + .digest('hex'); + await freshExternalGuards(); + checkpoint = (await recovery.call('completeBoundedBaselineCopy', { + category: checkpoint.category, + planSha256: checkpoint.plan.sha256, + proofSha256, + completedAt: Date.now(), + })) as BoundedBaselineCopyCheckpoint; + return checkpoint.completedJobs.map((job) => job.jobId); +} + +async function startChunk( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + pipe: string, + checkpoint: BoundedBaselineCopyCheckpoint, + chunkIndex: number, +): Promise { + const active = checkpoint.activeJob; + const chunk = checkpoint.plan.chunks[chunkIndex]; + if (!active || !chunk) throw new Error('Bounded baseline Copy start state is invalid'); + const params = chunkParams(recovery, checkpoint, chunkIndex, active.copyAttempt); + const response: unknown = await tb.request(`/v0/pipes/${pipe}/copy?${params.toString()}`, ''); + const receipt = + response && typeof response === 'object' && 'job' in response ? response.job : null; + const jobId = + receipt && typeof receipt === 'object' && 'job_id' in receipt ? receipt.job_id : null; + if (typeof jobId !== 'string') { + throw new Error('Bounded baseline Copy returned no job receipt; intent remains unresolved'); + } + return recovery.call('confirmBoundedBaselineCopyChunk', { + category: checkpoint.category, + planSha256: checkpoint.plan.sha256, + chunkIndex, + copyAttempt: active.copyAttempt, + jobId, + }) as Promise; +} + +async function recoverChunkReceipt( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + pipe: string, + checkpoint: BoundedBaselineCopyCheckpoint, + chunkIndex: number, +): Promise { + const active = checkpoint.activeJob; + const chunk = checkpoint.plan.chunks[chunkIndex]; + if (!active || !chunk) throw new Error('Bounded baseline Copy receipt state is invalid'); + const result = await tb.sql(`SELECT job_id,status FROM tinybird.jobs_log + WHERE job_type IN ('copy','copy_from_branch') + AND JSONExtractString(job_metadata,'pipe_name')=${quote(pipe)} + AND JSONExtractString(job_metadata,'parameters','org_id')=${quote(recovery.org)} + AND JSONExtractString(job_metadata,'parameters','start_day')=${quote(checkpoint.startDay)} + AND JSONExtractString(job_metadata,'parameters','end_day')=${quote(checkpoint.endDay)} + AND JSONExtractString(job_metadata,'parameters','chunk_start_day')=${quote(chunk.startDay)} + AND JSONExtractString(job_metadata,'parameters','chunk_end_day')=${quote(chunk.endDay)} + AND JSONExtractString(job_metadata,'parameters','copy_attempt')=${quote(String(active.copyAttempt))} + AND JSONExtractString(job_metadata,'parameters','_mode')='append' + ORDER BY created_at DESC,job_id DESC LIMIT 2`); + const job = result.data[0]; + if (result.data.length !== 1 || typeof job?.job_id !== 'string') { + throw new Error( + 'Bounded baseline Copy start outcome is unresolved; duplicate submission refused', + ); + } + return recovery.call('confirmBoundedBaselineCopyChunk', { + category: checkpoint.category, + planSha256: checkpoint.plan.sha256, + chunkIndex, + copyAttempt: active.copyAttempt, + jobId: job.job_id, + }) as Promise; +} + +function chunkParams( + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + chunkIndex: number, + copyAttempt: number, +): URLSearchParams { + const chunk = checkpoint.plan.chunks[chunkIndex]!; + return new URLSearchParams({ + org_id: recovery.org, + start_day: checkpoint.startDay, + end_day: checkpoint.endDay, + chunk_start_day: chunk.startDay, + chunk_end_day: chunk.endDay, + copy_attempt: String(copyAttempt), + _mode: 'append', + }); +} + +async function freshExternalGuards(): Promise { + await requireAgentProducerMaintenance(); + await requireDrainedAgentQueues(); +} diff --git a/scripts/ingest-recovery/agent-bounded-baseline-proof.ts b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts new file mode 100644 index 00000000..802785dc --- /dev/null +++ b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts @@ -0,0 +1,158 @@ +import type { + BaselineCopyChunk, + BoundedBaselineCopyCheckpoint, +} from '../../apps/agent-consumer/src/baseline-copy-contract'; +import { FACT_VERSION_DATASOURCES } from '../../apps/agent-consumer/src/delivery-write'; +import { quote } from './agent-data'; +import { preserveBaselineCopyFailure } from './agent-baseline-copy-retry-journal'; +import { + baselineProjection, + latestBaselineRows, + migrationScope, + retainedMigrationWindow, + type MigrationWindow, +} from './agent-migration-proof'; +import type { AgentRecoveryClient, AgentTinybirdClient } from './agent-transport'; + +export async function requireEmptyCategoryTarget( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + category: BoundedBaselineCopyCheckpoint['category'], +): Promise { + const result = await tb.sql( + `SELECT count() AS rows FROM ${FACT_VERSION_DATASOURCES[category]} FINAL WHERE OrgId=${quote(recovery.org)}`, + ); + if (count(result.data[0]?.rows, 'bounded baseline target rows') !== 0) { + throw new Error('Bounded baseline Copy requires an empty organization category target'); + } +} + +export async function requireEmptyChunkTarget( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + chunk: BaselineCopyChunk, +): Promise { + const result = await tb.sql( + `SELECT count() AS rows FROM ${FACT_VERSION_DATASOURCES[checkpoint.category]} FINAL WHERE ${migrationScope(checkpoint.category, recovery.org, chunk)}`, + ); + if (count(result.data[0]?.rows, 'bounded baseline chunk target rows') !== 0) { + throw new Error('Bounded baseline Copy chunk target is not empty'); + } +} + +export async function verifyChunkSource( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + chunk: BaselineCopyChunk, +): Promise { + const retained = retainedSlice(chunk, retainedMigrationWindow()); + const expected = expectedRetained(checkpoint, chunk, retained); + if (!retained) return 0; + const projection = baselineProjection(checkpoint.category); + const source = latestBaselineRows( + checkpoint.category, + recovery.org, + checkpoint, + retained, + projection, + ); + const result = await tb.sql(`SELECT count() AS rows, + sum(length(toJSONString(tuple(${projection})))) AS projected_bytes + FROM (${source})`); + const rows = count(result.data[0]?.rows, 'bounded baseline retained source rows'); + const projectedBytes = count( + result.data[0]?.projected_bytes, + 'bounded baseline retained source bytes', + ); + if (rows !== expected.rows || projectedBytes !== expected.projectedBytes) { + throw new Error('Bounded baseline Copy retained source changed after planning'); + } + return rows; +} + +export async function verifyCompletedChunkTarget( + tb: AgentTinybirdClient, + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + chunk: BaselineCopyChunk, +): Promise { + const retained = retainedSlice(chunk, retainedMigrationWindow()); + const expected = expectedRetained(checkpoint, chunk, retained); + if (!retained) return; + const scope = migrationScope(checkpoint.category, recovery.org, retained); + const result = await tb.sql(`SELECT count() AS rows, + countIf(DeliverySequence != 1 OR IsDeleted != 0) AS invalid_rows + FROM ${FACT_VERSION_DATASOURCES[checkpoint.category]} FINAL WHERE ${scope}`); + if ( + count(result.data[0]?.rows, 'bounded baseline completed target rows') !== expected.rows || + count(result.data[0]?.invalid_rows, 'bounded baseline completed invalid rows') !== 0 + ) { + throw new Error('Bounded baseline Copy completed target does not match its retained plan'); + } +} + +export function preserveChunkFailure( + root: string, + recovery: AgentRecoveryClient, + checkpoint: BoundedBaselineCopyCheckpoint, + active: { copyAttempt: number; jobId: string }, + providerJob: Record, +): void { + preserveBaselineCopyFailure(root, { + version: 1, + orgId: recovery.org, + checkpoint: { + category: checkpoint.category, + startDay: checkpoint.startDay, + endDay: checkpoint.endDay, + startedAt: checkpoint.startedAt, + copyAttempt: active.copyAttempt, + jobId: active.jobId, + complete: false, + }, + observedAt: Date.now(), + providerJob, + }); +} + +function expectedRetained( + checkpoint: BoundedBaselineCopyCheckpoint, + chunk: BaselineCopyChunk, + retained: MigrationWindow | null, +): { rows: number; projectedBytes: number } { + if (!retained) return { rows: 0, projectedBytes: 0 }; + return checkpoint.plan.dailyStats + .filter((stat) => stat.day >= retained.startDay && stat.day <= retained.endDay) + .reduce( + (total, stat) => ({ + rows: total.rows + stat.rows, + projectedBytes: total.projectedBytes + stat.projectedBytes, + }), + { rows: 0, projectedBytes: 0 }, + ); +} + +function retainedSlice( + chunk: BaselineCopyChunk, + retained: MigrationWindow, +): MigrationWindow | null { + const startDay = chunk.startDay > retained.startDay ? chunk.startDay : retained.startDay; + const endDay = chunk.endDay < retained.endDay ? chunk.endDay : retained.endDay; + return startDay <= endDay ? { startDay, endDay } : null; +} + +function count(value: unknown, label: string): number { + if ( + !( + (typeof value === 'number' && Number.isSafeInteger(value)) || + (typeof value === 'string' && /^\d+$/.test(value)) + ) + ) { + throw new Error(`Invalid ${label}`); + } + const number = Number(value); + if (!Number.isSafeInteger(number) || number < 0) throw new Error(`Invalid ${label}`); + return number; +} diff --git a/scripts/ingest-recovery/agent-migration-proof.test.ts b/scripts/ingest-recovery/agent-migration-proof.test.ts index 3b8233b3..bbda9ead 100644 --- a/scripts/ingest-recovery/agent-migration-proof.test.ts +++ b/scripts/ingest-recovery/agent-migration-proof.test.ts @@ -45,6 +45,7 @@ const messageProof = (rows: number): BaselineCategoryProof => ({ category: 'messages', rows, days: [], + dailyStats: [], }); describe('migrationOrganizations', () => { @@ -180,7 +181,13 @@ describe('latest baseline selection', () => { queries.push(query); return query.includes('HAVING uniqExact') ? { data: [], meta: [] } - : { data: [{ rows: '2', days: ['2026-09-12', '2026-09-13'] }], meta: [] }; + : { + data: [ + { day: '2026-09-12', rows: '1', projected_bytes: '100' }, + { day: '2026-09-13', rows: '1', projected_bytes: '110' }, + ], + meta: [], + }; }, } as unknown as AgentTinybirdClient; @@ -195,6 +202,10 @@ describe('latest baseline selection', () => { category, rows: 2, days: ['2026-09-12', '2026-09-13'], + dailyStats: [ + { day: '2026-09-12', rows: 1, projectedBytes: 100 }, + { day: '2026-09-13', rows: 1, projectedBytes: 110 }, + ], })), ); expect(queries).toHaveLength(CATEGORIES.length * 2); @@ -225,6 +236,23 @@ describe('latest baseline selection', () => { expect(queries[0]).toContain('> 1'); expect(queries[0]).toContain('LIMIT 1'); }); + + test('rejects malformed daily aggregate values without numeric coercion', async () => { + const client = { + async sql(query: string) { + return query.includes('HAVING uniqExact') + ? { data: [], meta: [] } + : { + data: [{ day: '2026-09-13', rows: true, projected_bytes: '10' }], + meta: [], + }; + }, + } as unknown as AgentTinybirdClient; + const window = { startDay: '2026-09-13', endDay: '2026-09-13' }; + await expect(inspectBaseline(client, 'org-proof', window, window)).rejects.toThrow( + 'Invalid messages daily rows', + ); + }); }); describe('verifyBaseline', () => { @@ -236,7 +264,7 @@ describe('verifyBaseline', () => { async sql(query: string) { inspections.push(query); return { - data: query.includes('HAVING uniqExact') ? [] : [{ rows: 0, days: [] }], + data: [], meta: [], }; }, diff --git a/scripts/ingest-recovery/agent-migration-proof.ts b/scripts/ingest-recovery/agent-migration-proof.ts index 759648a9..b64dec11 100644 --- a/scripts/ingest-recovery/agent-migration-proof.ts +++ b/scripts/ingest-recovery/agent-migration-proof.ts @@ -10,6 +10,7 @@ export interface BaselineCategoryProof { category: Category; rows: number; days: string[]; + dailyStats: { day: string; rows: number; projectedBytes: number }[]; } export interface MigrationWindow { startDay: string; @@ -126,26 +127,45 @@ export async function inspectBaseline( ); } const rows = ( - await tb.sql(`SELECT count() AS rows, - arraySort(groupUniqArray(toString(toDate(latest_time)))) AS days + await tb.sql(`SELECT toString(toDate(latest_time)) AS day, + count() AS rows, + sum(latest_projected_bytes) AS projected_bytes FROM ( - SELECT argMax(${time},IngestedAt) AS latest_time + SELECT argMax(${time},IngestedAt) AS latest_time, + argMax(length(toJSONString(tuple(${projection}))),IngestedAt) AS latest_projected_bytes FROM ${DATASOURCES[category]} WHERE ${migrationScope(category, org, copyWindow)} GROUP BY ${identity} ) WHERE latest_time >= toDateTime(${quote(window.startDay)}) - AND latest_time < toDateTime(${quote(window.endDay)}) + INTERVAL 1 DAY`) + AND latest_time < toDateTime(${quote(window.endDay)}) + INTERVAL 1 DAY + GROUP BY day ORDER BY day`) ).data; - const row = rows[0]; + if (!Array.isArray(rows) || rows.length > 367) { + throw new Error(`Source ${category} is not uniquely defined; baseline requires repair`); + } + const dailyStats = rows.map((row) => ({ + day: String(row.day), + rows: safeCount(row.rows, `${category} daily rows`), + projectedBytes: safeCount(row.projected_bytes, `${category} daily projected bytes`), + })); if ( - !row || - !Number.isSafeInteger(Number(row.rows)) || - !Array.isArray(row.days) || - row.days.length > 367 + dailyStats.some( + (row) => + !/^\d{4}-\d{2}-\d{2}$/.test(row.day) || + !Number.isSafeInteger(row.rows) || + row.rows <= 0 || + !Number.isSafeInteger(row.projectedBytes) || + row.projectedBytes <= 0, + ) ) { throw new Error(`Source ${category} is not uniquely defined; baseline requires repair`); } - proofs.push({ category, rows: Number(row.rows), days: row.days as string[] }); + proofs.push({ + category, + rows: dailyStats.reduce((sum, row) => sum + row.rows, 0), + days: dailyStats.map((row) => row.day), + dailyStats, + }); } return proofs; } @@ -270,12 +290,20 @@ export async function verifyBaseline( } function safeCount(value: unknown, label: string): number { + if ( + !( + (typeof value === 'number' && Number.isSafeInteger(value)) || + (typeof value === 'string' && /^\d+$/.test(value)) + ) + ) { + throw new Error(`Invalid ${label}`); + } const count = Number(value); if (!Number.isSafeInteger(count) || count < 0) throw new Error(`Invalid ${label}`); return count; } -function baselineProjection(category: Category): string { +export function baselineProjection(category: Category): string { const source = DATASOURCES[category]; const columns = [ ...readFileSync(`datasources/${source}.datasource`, 'utf8').matchAll(/^\s+`([^`]+)`\s/gm), diff --git a/scripts/ingest-recovery/migrate-agent-ingestion.ts b/scripts/ingest-recovery/migrate-agent-ingestion.ts index f5f656a0..150a4bad 100644 --- a/scripts/ingest-recovery/migrate-agent-ingestion.ts +++ b/scripts/ingest-recovery/migrate-agent-ingestion.ts @@ -90,9 +90,10 @@ try { for (const category of categories) { if (category.rows === 0) continue; jobs.push( - await runBaselineCopy(runtime.tb, recovery, category.category, copyWindow, { + ...(await runBaselineCopy(runtime.tb, recovery, category.category, copyWindow, { retryJournalRoot: values['retry-journal'], - }), + sourceProof: category, + })), ); } } diff --git a/scripts/ingest-recovery/worker.mjs b/scripts/ingest-recovery/worker.mjs index 28d7dbe9..dbad8c1a 100644 --- a/scripts/ingest-recovery/worker.mjs +++ b/scripts/ingest-recovery/worker.mjs @@ -16,6 +16,11 @@ const AGENT_METHODS = new Set([ 'beginBaselineCopy', 'confirmBaselineCopy', 'retryBaselineCopy', + 'beginBoundedBaselineCopy', + 'armBoundedBaselineCopyChunk', + 'confirmBoundedBaselineCopyChunk', + 'completeBoundedBaselineCopyChunk', + 'completeBoundedBaselineCopy', 'completeIngestionMigration', 'completeGlobalIngestionMigration', 'beginFactRebuild', From 3b69a81427dd32f626e104ee3a21c951906fc6e4 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 21:38:33 -0700 Subject: [PATCH 5/9] docs(agent-consumer): record bounded copy handoff --- docs/adr/0024-bounded-agent-ingestion.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/docs/adr/0024-bounded-agent-ingestion.md b/docs/adr/0024-bounded-agent-ingestion.md index 57a0f722..125b59e6 100644 --- a/docs/adr/0024-bounded-agent-ingestion.md +++ b/docs/adr/0024-bounded-agent-ingestion.md @@ -67,6 +67,16 @@ CI first expands the Tinybird schema while preserving the exact previously deplo deploys the compatible consumer, pauses producer acceptance, drains and freezes the old batchers, and copies retained baseline facts at revision 1. New deliveries start at revision 2. A durable baseline Copy checkpoint prevents a restart from launching another job after an unknown outcome. +Each category's source proof becomes an immutable plan of at most seven calendar days, 50,000 +global-winner rows, and 64 MiB of projected tuple JSON per Copy. The checkpoint records one active +intent and ordered job receipts, so a lost response is recovered from one exact `jobs_log` match and +a terminal partial failure cannot submit another chunk. + +A terminal failed whole-window checkpoint transitions to this bounded plan only through the local +operator with `--retry-journal` pointing at a private filesystem directory. The journal is fsynced +before the coordinator archives the failed receipt and its evidence hashes. The deployment Action +does not supply this option: it stops at a failed legacy checkpoint until the operator preserves the +provider record and arms the bounded plan, after which an ordinary rerun resumes its chunks. Legacy MergeTree tables can contain multiple versions of one identity. The baseline selects the greatest `IngestedAt` per natural identity across the retained window, matching the existing recovery From 4e189d0c3f105721b4bde6a9497428cc313c4375 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Mon, 14 Sep 2026 09:46:14 -0700 Subject: [PATCH 6/9] chore(agent-consumer): keep baseline plan caps module-private Co-Authored-By: Claude Fable 5.1 --- apps/agent-consumer/src/baseline-copy-plan.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/agent-consumer/src/baseline-copy-plan.ts b/apps/agent-consumer/src/baseline-copy-plan.ts index 7f2b368d..4466cc7c 100644 --- a/apps/agent-consumer/src/baseline-copy-plan.ts +++ b/apps/agent-consumer/src/baseline-copy-plan.ts @@ -6,11 +6,11 @@ import type { import { CATEGORIES, type Category } from './facts'; import { assertExactKeys } from './agent-delivery-coordinator-validation'; -export const MAX_BASELINE_COPY_DAYS = 367; +const MAX_BASELINE_COPY_DAYS = 367; export const MAX_BASELINE_COPY_CHUNK_DAYS = 7; export const MAX_BASELINE_COPY_CHUNK_ROWS = 50_000; export const MAX_BASELINE_COPY_CHUNK_BYTES = 64 * 1024 * 1024; -export const MAX_BASELINE_COPY_CHECKPOINT_BYTES = 128 * 1024; +const MAX_BASELINE_COPY_CHECKPOINT_BYTES = 128 * 1024; export function isBoundedBaselineCopy( checkpoint: BaselineCopyCheckpoint, From fef2163e0ce000a97814e36e084e13b0c28c117f Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Mon, 14 Sep 2026 09:56:17 -0700 Subject: [PATCH 7/9] fix(ingest): clip bounded Copy chunks to retention and drop legacy Copy RPC Retention rolls forward at UTC midnight during a long bounded run. The chunk proofs only cover the retained slice, so the Copy now posts that slice and refuses a fully expired chunk instead of appending unverified days. Receipt recovery matches the clipped bounds within the plan chunk. Remove the unguarded whole-window beginBaselineCopy RPC and bridge method, which no longer has a caller and would strand a category with an unrecoverable legacy checkpoint. Make the baseline mutation guard a real private member so the coordinator stub cannot leave the entrypoint over RPC. Update the Tinybird Local fixture harness to the daily-stats inspection shape that the bounded plan introduced; hosted Tinybird Schema Check failed on the old stub. Co-Authored-By: Claude Fable 5.1 --- apps/agent-consumer/src/index.ts | 20 ++---- .../ci/tinybird_baseline_version_fixtures.py | 11 ++- .../agent-bounded-baseline-copy.test.ts | 70 ++++++++++++++++++- .../agent-bounded-baseline-copy.ts | 17 +++-- .../agent-bounded-baseline-proof.ts | 2 +- scripts/ingest-recovery/worker.mjs | 1 - 6 files changed, 97 insertions(+), 24 deletions(-) diff --git a/apps/agent-consumer/src/index.ts b/apps/agent-consumer/src/index.ts index d5475b75..c28a7a49 100644 --- a/apps/agent-consumer/src/index.ts +++ b/apps/agent-consumer/src/index.ts @@ -2,7 +2,6 @@ import type { BaselineCopyCheckpoint, BaselineMigrationWindow, BaselineCopyChunkInput, - BeginBaselineCopyInput, BeginBoundedBaselineCopyInput, CompleteBoundedBaselineCopyInput, ConfirmBaselineCopyInput, @@ -224,7 +223,7 @@ export class AgentIngestion extends WorkerEntrypoint { } export class TraceRecovery extends WorkerEntrypoint { - private async requireBaselineMutation(orgId: string) { + async #requireBaselineMutation(orgId: string) { const normalized = normalizeAgentShardId(orgId); const organization = this.env.AGENT_DELIVERY_COORDINATOR.getByName(`org:${normalized}`); const baseline = this.env.AGENT_DELIVERY_COORDINATOR.getByName(`baseline:${normalized}`); @@ -270,11 +269,6 @@ export class TraceRecovery extends WorkerEntrypoint { `baseline:${normalizeAgentShardId(orgId)}`, ).getBaselineCopy(input); } - beginBaselineCopy(orgId: string, input: BeginBaselineCopyInput) { - return this.env.AGENT_DELIVERY_COORDINATOR.getByName( - `baseline:${normalizeAgentShardId(orgId)}`, - ).beginBaselineCopy(input); - } confirmBaselineCopy(orgId: string, input: ConfirmBaselineCopyInput) { return this.env.AGENT_DELIVERY_COORDINATOR.getByName( `baseline:${normalizeAgentShardId(orgId)}`, @@ -282,27 +276,27 @@ export class TraceRecovery extends WorkerEntrypoint { } async retryBaselineCopy(orgId: string, input: RetryBaselineCopyInput) { - return (await this.requireBaselineMutation(orgId)).retryBaselineCopy(input); + return (await this.#requireBaselineMutation(orgId)).retryBaselineCopy(input); } async beginBoundedBaselineCopy(orgId: string, input: BeginBoundedBaselineCopyInput) { - return (await this.requireBaselineMutation(orgId)).beginBoundedBaselineCopy(input); + return (await this.#requireBaselineMutation(orgId)).beginBoundedBaselineCopy(input); } async armBoundedBaselineCopyChunk(orgId: string, input: BaselineCopyChunkInput) { - return (await this.requireBaselineMutation(orgId)).armBoundedBaselineCopyChunk(input); + return (await this.#requireBaselineMutation(orgId)).armBoundedBaselineCopyChunk(input); } async confirmBoundedBaselineCopyChunk(orgId: string, input: ConfirmBaselineCopyChunkInput) { - return (await this.requireBaselineMutation(orgId)).confirmBoundedBaselineCopyChunk(input); + return (await this.#requireBaselineMutation(orgId)).confirmBoundedBaselineCopyChunk(input); } async completeBoundedBaselineCopyChunk(orgId: string, input: ConfirmBaselineCopyChunkInput) { - return (await this.requireBaselineMutation(orgId)).completeBoundedBaselineCopyChunk(input); + return (await this.#requireBaselineMutation(orgId)).completeBoundedBaselineCopyChunk(input); } async completeBoundedBaselineCopy(orgId: string, input: CompleteBoundedBaselineCopyInput) { - return (await this.requireBaselineMutation(orgId)).completeBoundedBaselineCopy(input); + return (await this.#requireBaselineMutation(orgId)).completeBoundedBaselineCopy(input); } inspectGlobalIngestionMigration() { diff --git a/scripts/ci/tinybird_baseline_version_fixtures.py b/scripts/ci/tinybird_baseline_version_fixtures.py index 9f2551b1..bfcf7732 100644 --- a/scripts/ci/tinybird_baseline_version_fixtures.py +++ b/scripts/ci/tinybird_baseline_version_fixtures.py @@ -189,14 +189,14 @@ def verify_resumed_proof(client, org, previous, today, query_rows) -> None: const capture={sql:async query=>{ const kind=query.includes('HAVING uniqExact')?'conflict':query.includes('argMax(')?'inspection':'parity'; queries.push({kind,query,category:currentCategory}); - return {data:kind==='conflict'?[]:kind==='inspection'?[{rows:0,days:[]}]:[ + return {data:kind==='conflict'?[]:kind==='inspection'?[]:[ {source_rows:0,target_rows:0,invalid_metadata:0,missing_target:0,unexpected_target:0} ],meta:[]}; }}; await inspectBaseline(capture,org,retainedWindow,copyWindow); for(const category of CATEGORIES) { currentCategory=category; - await verifyBaseline(capture,org,retainedWindow,{category,rows:0,days:[]},copyWindow); + await verifyBaseline(capture,org,retainedWindow,{category,rows:0,days:[],dailyStats:[]},copyWindow); } console.log(JSON.stringify(queries)); """ @@ -215,7 +215,12 @@ def verify_resumed_proof(client, org, previous, today, query_rows) -> None: if entry["kind"] == "conflict": valid = rows == [] elif entry["kind"] == "inspection": - valid = len(rows) == 1 and int(rows[0]["rows"]) == 1 and rows[0]["days"] == [today.strftime("%Y-%m-%d")] + valid = ( + len(rows) == 1 + and int(rows[0]["rows"]) == 1 + and rows[0]["day"] == today.strftime("%Y-%m-%d") + and int(rows[0]["projected_bytes"]) > 0 + ) else: valid = len(rows) == 1 and int(rows[0]["source_rows"]) == int(rows[0]["target_rows"]) valid = valid and not any(int(rows[0].get(field, 0)) for field in ["invalid_metadata", "missing_target", "unexpected_target"]) diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts index 421511c6..d6b34596 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts @@ -53,6 +53,12 @@ afterAll(() => { restore('CLOUDFLARE_API_TOKEN', originalApiToken); }); +function shiftDay(day: string, days: number): string { + return new Date(Date.parse(`${day}T00:00:00.000Z`) + days * 86_400_000) + .toISOString() + .slice(0, 10); +} + function restore(name: string, value: string | undefined): void { if (value === undefined) delete process.env[name]; else process.env[name] = value; @@ -79,7 +85,7 @@ function checkpoint(): BoundedBaselineCopyCheckpoint { }; } -function fixture(initial = checkpoint()) { +function fixture(initial = checkpoint(), verificationStartDay?: string) { let state: BaselineCopyCheckpoint = structuredClone(initial); let jobs: Record[] = []; let terminal: Record = { job_id: 'job-chunk', status: 'done' }; @@ -87,7 +93,9 @@ function fixture(initial = checkpoint()) { const requests: string[] = []; const queries: string[] = []; const calls: string[] = []; - const verificationChunk = chunkAgentDayRange(initial)[0]!; + const verificationChunk = { + startDay: verificationStartDay ?? chunkAgentDayRange(initial)[0]!.startDay, + }; const recovery = { org: 'org-proof', async call(method: string, input: Record) { @@ -192,6 +200,64 @@ describe('bounded baseline Copy operator', () => { expect(f.state()).toMatchObject({ complete: true, completedJobs: [{ jobId: 'job-chunk' }] }); }); + test('clips the Copy to the retained slice when retention rolled past a chunk day', async () => { + const window = retainedMigrationWindow(); + const expired = shiftDay(window.startDay, -1); + const proof = { + category: 'tool_events' as const, + rows: 3, + days: [expired, window.startDay], + dailyStats: [ + { day: expired, rows: 1, projectedBytes: 10 }, + { day: window.startDay, rows: 2, projectedBytes: 20 }, + ], + }; + const initial: BoundedBaselineCopyCheckpoint = { + mode: 'bounded', + category: proof.category, + startDay: expired, + endDay: window.endDay, + startedAt: 1, + plan: buildBaselineCopyPlan(proof, { startDay: expired, endDay: window.endDay }), + completedJobs: [], + complete: false, + }; + expect(initial.plan.chunks[0]).toMatchObject({ startDay: expired, endDay: window.startDay }); + const f = fixture(initial, window.startDay); + await expect(runBoundedBaselineCopy(f.tb, f.recovery, f.state(), {})).resolves.toEqual([ + 'job-chunk', + ]); + const copy = f.requests.find((path) => path.includes('/copy?'))!; + expect(copy).toContain(`chunk_start_day=${window.startDay}`); + expect(copy).toContain(`chunk_end_day=${window.startDay}`); + expect(copy).not.toContain(`chunk_start_day=${expired}`); + }); + + test('refuses to Copy a chunk that is fully outside analytics retention', async () => { + const day = '2020-01-01'; + const proof = { + category: 'tool_events' as const, + rows: 1, + days: [day], + dailyStats: [{ day, rows: 1, projectedBytes: 10 }], + }; + const initial: BoundedBaselineCopyCheckpoint = { + mode: 'bounded', + category: proof.category, + startDay: day, + endDay: day, + startedAt: 1, + plan: buildBaselineCopyPlan(proof, { startDay: day, endDay: day }), + completedJobs: [], + complete: false, + }; + const f = fixture(initial); + await expect(runBoundedBaselineCopy(f.tb, f.recovery, f.state(), {})).rejects.toThrow( + 'fully outside analytics retention', + ); + expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + }); + test('recovers exactly one lost receipt and refuses missing or ambiguous matches', async () => { for (const matches of [[], [{ job_id: 'one' }, { job_id: 'two' }]]) { const active = checkpoint(); diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts index 62fd2ced..e8048fef 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts @@ -25,6 +25,7 @@ import { preserveChunkFailure, requireEmptyCategoryTarget, requireEmptyChunkTarget, + retainedSlice, verifyChunkSource, verifyCompletedChunkTarget, } from './agent-bounded-baseline-proof'; @@ -225,8 +226,8 @@ async function recoverChunkReceipt( AND JSONExtractString(job_metadata,'parameters','org_id')=${quote(recovery.org)} AND JSONExtractString(job_metadata,'parameters','start_day')=${quote(checkpoint.startDay)} AND JSONExtractString(job_metadata,'parameters','end_day')=${quote(checkpoint.endDay)} - AND JSONExtractString(job_metadata,'parameters','chunk_start_day')=${quote(chunk.startDay)} - AND JSONExtractString(job_metadata,'parameters','chunk_end_day')=${quote(chunk.endDay)} + AND JSONExtractString(job_metadata,'parameters','chunk_start_day')>=${quote(chunk.startDay)} + AND JSONExtractString(job_metadata,'parameters','chunk_end_day')<=${quote(chunk.endDay)} AND JSONExtractString(job_metadata,'parameters','copy_attempt')=${quote(String(active.copyAttempt))} AND JSONExtractString(job_metadata,'parameters','_mode')='append' ORDER BY created_at DESC,job_id DESC LIMIT 2`); @@ -252,12 +253,20 @@ function chunkParams( copyAttempt: number, ): URLSearchParams { const chunk = checkpoint.plan.chunks[chunkIndex]!; + // Retention rolls forward at UTC midnight while a long run is in flight. The proofs only cover + // the retained slice of a chunk, so the Copy must never append days those proofs no longer see. + const retained = retainedSlice(chunk, retainedMigrationWindow()); + if (!retained) { + throw new Error( + 'Bounded baseline Copy chunk is fully outside analytics retention; replan before resuming', + ); + } return new URLSearchParams({ org_id: recovery.org, start_day: checkpoint.startDay, end_day: checkpoint.endDay, - chunk_start_day: chunk.startDay, - chunk_end_day: chunk.endDay, + chunk_start_day: retained.startDay, + chunk_end_day: retained.endDay, copy_attempt: String(copyAttempt), _mode: 'append', }); diff --git a/scripts/ingest-recovery/agent-bounded-baseline-proof.ts b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts index 802785dc..171fd21f 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-proof.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts @@ -134,7 +134,7 @@ function expectedRetained( ); } -function retainedSlice( +export function retainedSlice( chunk: BaselineCopyChunk, retained: MigrationWindow, ): MigrationWindow | null { diff --git a/scripts/ingest-recovery/worker.mjs b/scripts/ingest-recovery/worker.mjs index dbad8c1a..754131b4 100644 --- a/scripts/ingest-recovery/worker.mjs +++ b/scripts/ingest-recovery/worker.mjs @@ -13,7 +13,6 @@ const AGENT_METHODS = new Set([ 'freezeIngestionMigration', 'seedIngestionMigration', 'beginBaselineMigrationWindow', - 'beginBaselineCopy', 'confirmBaselineCopy', 'retryBaselineCopy', 'beginBoundedBaselineCopy', From a0b7883785aa7ae76ae8c58bef8eb250a02e4a93 Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:03:34 -0700 Subject: [PATCH 8/9] fix(ingest): refuse expired bounded chunks before arming their intent The retention check ran inside startChunk, after armBoundedBaselineCopyChunk had durably committed the intent, so a fully expired chunk left an active job with no receipt that no later run could clear. Decide the retained slice once per chunk before arming and post exactly that slice. Co-Authored-By: Claude Fable 5.1 --- .../agent-bounded-baseline-copy.test.ts | 2 ++ .../agent-bounded-baseline-copy.ts | 25 ++++++++++--------- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts index d6b34596..88d6431e 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.test.ts @@ -256,6 +256,8 @@ describe('bounded baseline Copy operator', () => { 'fully outside analytics retention', ); expect(f.requests.filter((path) => path.includes('/copy?'))).toHaveLength(0); + expect(f.calls).not.toContain('armBoundedBaselineCopyChunk'); + expect(f.state().activeJob).toBeUndefined(); }); test('recovers exactly one lost receipt and refuses missing or ambiguous matches', async () => { diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts index e8048fef..fc8cc23c 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts @@ -108,6 +108,15 @@ export async function runBoundedBaselineCopy( const chunkIndex = checkpoint.completedJobs.length; const chunk = checkpoint.plan.chunks[chunkIndex]!; if (!checkpoint.activeJob) { + // Retention rolls forward at UTC midnight while a long run is in flight. The proofs only + // cover the retained slice of a chunk, so decide the slice once, before the intent is + // durably armed, and post exactly that slice. A fully expired chunk must never be armed. + const retained = retainedSlice(chunk, retainedMigrationWindow()); + if (!retained) { + throw new Error( + 'Bounded baseline Copy chunk is fully outside analytics retention; replan before resuming', + ); + } await freshExternalGuards(); await verifyChunkSource(tb, recovery, checkpoint, chunk); await requireEmptyChunkTarget(tb, recovery, checkpoint, chunk); @@ -121,7 +130,7 @@ export async function runBoundedBaselineCopy( })) as BoundedBaselineCopyCheckpoint & { created: boolean }; checkpoint = armed; if (armed.created) { - checkpoint = await startChunk(tb, recovery, pipe, checkpoint, chunkIndex); + checkpoint = await startChunk(tb, recovery, pipe, checkpoint, chunkIndex, retained); } } if (!checkpoint.activeJob) throw new Error('Bounded baseline Copy omitted its active intent'); @@ -188,11 +197,12 @@ async function startChunk( pipe: string, checkpoint: BoundedBaselineCopyCheckpoint, chunkIndex: number, + retained: MigrationWindow, ): Promise { const active = checkpoint.activeJob; const chunk = checkpoint.plan.chunks[chunkIndex]; if (!active || !chunk) throw new Error('Bounded baseline Copy start state is invalid'); - const params = chunkParams(recovery, checkpoint, chunkIndex, active.copyAttempt); + const params = chunkParams(recovery, checkpoint, retained, active.copyAttempt); const response: unknown = await tb.request(`/v0/pipes/${pipe}/copy?${params.toString()}`, ''); const receipt = response && typeof response === 'object' && 'job' in response ? response.job : null; @@ -249,18 +259,9 @@ async function recoverChunkReceipt( function chunkParams( recovery: AgentRecoveryClient, checkpoint: BoundedBaselineCopyCheckpoint, - chunkIndex: number, + retained: MigrationWindow, copyAttempt: number, ): URLSearchParams { - const chunk = checkpoint.plan.chunks[chunkIndex]!; - // Retention rolls forward at UTC midnight while a long run is in flight. The proofs only cover - // the retained slice of a chunk, so the Copy must never append days those proofs no longer see. - const retained = retainedSlice(chunk, retainedMigrationWindow()); - if (!retained) { - throw new Error( - 'Bounded baseline Copy chunk is fully outside analytics retention; replan before resuming', - ); - } return new URLSearchParams({ org_id: recovery.org, start_day: checkpoint.startDay, From a280ce35590cc718fe961c2a413420ac74e90e0b Mon Sep 17 00:00:00 2001 From: "useotto-dev[bot]" <252773270+useotto-dev[bot]@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:18:33 -0700 Subject: [PATCH 9/9] fix(ingest): prove and post one retained slice per bounded chunk The source proof, the arm, the Copy POST, and the target proof now share the slice decided at the top of the chunk iteration, so a UTC midnight between them cannot make the proofs cover different days than the Copy appended. Co-Authored-By: Claude Fable 5.1 --- .../ingest-recovery/agent-bounded-baseline-copy.ts | 12 ++++++------ .../ingest-recovery/agent-bounded-baseline-proof.ts | 4 ++-- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts index fc8cc23c..fdbe16a9 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-copy.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-copy.ts @@ -107,18 +107,18 @@ export async function runBoundedBaselineCopy( while (checkpoint.completedJobs.length < checkpoint.plan.chunks.length) { const chunkIndex = checkpoint.completedJobs.length; const chunk = checkpoint.plan.chunks[chunkIndex]!; + // Retention rolls forward at UTC midnight while a long run is in flight. Decide the retained + // slice once per chunk so the source proof, the arm, the Copy POST, and the target proof all + // cover exactly the same days. A fully expired chunk must never be armed. + const retained = retainedSlice(chunk, retainedMigrationWindow()); if (!checkpoint.activeJob) { - // Retention rolls forward at UTC midnight while a long run is in flight. The proofs only - // cover the retained slice of a chunk, so decide the slice once, before the intent is - // durably armed, and post exactly that slice. A fully expired chunk must never be armed. - const retained = retainedSlice(chunk, retainedMigrationWindow()); if (!retained) { throw new Error( 'Bounded baseline Copy chunk is fully outside analytics retention; replan before resuming', ); } await freshExternalGuards(); - await verifyChunkSource(tb, recovery, checkpoint, chunk); + await verifyChunkSource(tb, recovery, checkpoint, chunk, retained); await requireEmptyChunkTarget(tb, recovery, checkpoint, chunk); await freshExternalGuards(); const copyAttempt = Date.now(); @@ -149,7 +149,7 @@ export async function runBoundedBaselineCopy( } await freshExternalGuards(); await verifyChunkSource(tb, recovery, checkpoint, chunk); - await verifyCompletedChunkTarget(tb, recovery, checkpoint, chunk); + await verifyCompletedChunkTarget(tb, recovery, checkpoint, chunk, retained); await freshExternalGuards(); checkpoint = (await recovery.call('completeBoundedBaselineCopyChunk', { category: checkpoint.category, diff --git a/scripts/ingest-recovery/agent-bounded-baseline-proof.ts b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts index 171fd21f..c8f0540f 100644 --- a/scripts/ingest-recovery/agent-bounded-baseline-proof.ts +++ b/scripts/ingest-recovery/agent-bounded-baseline-proof.ts @@ -46,8 +46,8 @@ export async function verifyChunkSource( recovery: AgentRecoveryClient, checkpoint: BoundedBaselineCopyCheckpoint, chunk: BaselineCopyChunk, + retained: MigrationWindow | null = retainedSlice(chunk, retainedMigrationWindow()), ): Promise { - const retained = retainedSlice(chunk, retainedMigrationWindow()); const expected = expectedRetained(checkpoint, chunk, retained); if (!retained) return 0; const projection = baselineProjection(checkpoint.category); @@ -77,8 +77,8 @@ export async function verifyCompletedChunkTarget( recovery: AgentRecoveryClient, checkpoint: BoundedBaselineCopyCheckpoint, chunk: BaselineCopyChunk, + retained: MigrationWindow | null = retainedSlice(chunk, retainedMigrationWindow()), ): Promise { - const retained = retainedSlice(chunk, retainedMigrationWindow()); const expected = expectedRetained(checkpoint, chunk, retained); if (!retained) return; const scope = migrationScope(checkpoint.category, recovery.org, retained);