diff --git a/docs/contributing/architecture/data-storage.md b/docs/contributing/architecture/data-storage.md index 7aacd18798..f9692274c1 100644 --- a/docs/contributing/architecture/data-storage.md +++ b/docs/contributing/architecture/data-storage.md @@ -709,11 +709,13 @@ on write unless a migration backfills existing rows. (`0028-published-bundle-artifacts-and-archived-jobs.sql`) stores package dependency pointers queried with SQLite JSON functions in `packages/worker/src/repo/published-bundle-artifacts-repo.ts`. -- `package_invocations.package_ids_json`, - `package_invocations.package_kody_ids_json`, - `package_invocations.export_names_json`, `package_invocations.sources_json`, - and `package_invocations.response_json` (`0029-package-invocations.sql`) store - invocation routing and cached response projections. +- `package_invocation_tokens.package_ids_json`, + `package_invocation_tokens.package_kody_ids_json`, + `package_invocation_tokens.export_names_json`, and + `package_invocation_tokens.sources_json` (`0029-package-invocations.sql`) + store invocation-token scope projections. The sibling `package_invocations` + table from that migration was dropped (`0112-drop-package-invocations.sql`); + its replay cache lives in the RunLog Durable Object ledger now. - `email_messages.*_addresses_json`, `email_messages.references_json`, `email_messages.headers_json`, and `email_delivery_events.detail_json` (`0030-email-primitives.sql`, `0031-unified-email-receipt.sql`, @@ -923,14 +925,6 @@ documented exemption. Current retention policies: -- `package_invocations`: **legacy dual-read window.** Keyed invocations now - claim and store replay responses in the per-user `RunLog` Durable Object - ledger (`package_invocation_ledger` table inside the DO), which self-enforces - the same 90-day terminal-row window; the keyed path only reads this D1 table - as a fallback for pre-migration keys and never writes it. The sweep keeps - draining pre-migration terminal rows (rows with `status = 'in_progress'` are - never pruned) until a follow-up drops the table, the fallback read, and this - policy entry. - `mcp_memory_conversation_suppressions`: keep active suppressions and prune expired rows only after they have not been seen for 90 days. The existing request-time memory prune may remove expired rows sooner. diff --git a/docs/contributing/architecture/invocation-overhead-guardrails.md b/docs/contributing/architecture/invocation-overhead-guardrails.md index c45b4c29b6..85df07410a 100644 --- a/docs/contributing/architecture/invocation-overhead-guardrails.md +++ b/docs/contributing/architecture/invocation-overhead-guardrails.md @@ -27,9 +27,8 @@ call, not about what user code does inside the call. lives in the per-user `RunLog` Durable Object (see [Run records](./run-records.md)), so the durability cost is **one awaited DO call for claim + run-record begin and one for terminal response + run-record - finish** — not D1 round trips. During the dual-read window the keyed path also - pays one awaited D1 **read** for pre-migration keys; a follow-up removes it - with the legacy `package_invocations` table. + finish** — no D1 round trips at all (the legacy `package_invocations` table + and its dual-read fallback are gone). ## Watch the percentiles diff --git a/docs/contributing/architecture/run-records.md b/docs/contributing/architecture/run-records.md index 1ceed7f8e9..506411d274 100644 --- a/docs/contributing/architecture/run-records.md +++ b/docs/contributing/architecture/run-records.md @@ -221,12 +221,13 @@ means a replayed delivery for that key re-executes instead of replaying. overwritten. - **Ledger retention is DO-local**: terminal rows keep replay responses for 90 days (`packageInvocationLedgerRetentionDays`), pruned by the same retention - passes and alarm as run rows; `in_progress` rows are never pruned. The old D1 - retention sweep remains only to drain pre-migration rows. -- **Dual-read window**: the keyed path reads the DO first and falls back to a - read-only D1 `package_invocations` lookup for keys claimed before the - migration. Writes go only to the DO. A follow-up removes the fallback, the D1 - table, its sweep, and the legacy row shapes. + passes and alarm as run rows; `in_progress` rows are never pruned. There is no + D1 sweep — the legacy table is gone. +- **The DO is the only store**: the legacy D1 `package_invocations` table was + dropped (`0112-drop-package-invocations.sql`) after the dual-read window was + deliberately waived, and the keyed path performs no D1 ledger reads or writes. + Keys claimed before the DO migration no longer replay; a redelivery for such a + key executes fresh, exactly like a new key. - Account export pages ledger rows through the same `run_records` section cursor (runs first, then ledger rows); account deletion purges them with `clearAll`. Disaster recovery deliberately does not stage the DO — losing it risks diff --git a/docs/contributing/disaster-recovery.md b/docs/contributing/disaster-recovery.md index 0de64a8ff0..b43116e7ae 100644 --- a/docs/contributing/disaster-recovery.md +++ b/docs/contributing/disaster-recovery.md @@ -193,17 +193,17 @@ Restore rebuilds these; do not treat them as recovery media: **Known risk — keyed invocation idempotency ledger:** the same `RunLog` DO also holds the keyed package-invocation idempotency ledger (claims + 90-day - terminal replay responses; migrated off the D1 `package_invocations` table). - Losing the DO therefore loses idempotency/replay state, not just - observability: webhook providers and other keyed callers that retry deliveries - would **re-execute** invocations whose keys were already terminal, and - in-flight replays would return fresh executions instead of stored responses. - This is accepted with eyes open: lost rows are gone — later traffic - re-establishes claims only for keys used after the loss, it does not restore - protection for keys used before it. The blast radius is duplicate side effects - (a loss of correctness state) bounded by the 90-day retention window; no - stored user content is lost. Restores from a D1 backup restore the legacy - table only; the DO ledger is not part of DR media. + terminal replay responses; the legacy D1 `package_invocations` table was + dropped, so the DO is the only store). Losing the DO therefore loses + idempotency/replay state, not just observability: webhook providers and other + keyed callers that retry deliveries would **re-execute** invocations whose + keys were already terminal, and in-flight replays would return fresh + executions instead of stored responses. This is accepted with eyes open: lost + rows are gone — later traffic re-establishes claims only for keys used after + the loss, it does not restore protection for keys used before it. The blast + radius is duplicate side effects (a loss of correctness state) bounded by the + 90-day retention window; no stored user content is lost. The DO ledger is not + part of DR media, and D1 backups no longer carry any invocation ledger at all. ### Honest gaps diff --git a/packages/worker/migrations/0112-drop-package-invocations.sql b/packages/worker/migrations/0112-drop-package-invocations.sql new file mode 100644 index 0000000000..3180590831 --- /dev/null +++ b/packages/worker/migrations/0112-drop-package-invocations.sql @@ -0,0 +1,8 @@ +-- Retire the legacy keyed package-invocation idempotency ledger. The ledger +-- moved into the per-user RunLog Durable Object (see +-- docs/contributing/architecture/run-records.md); writes stopped with that +-- migration and the dual-read fallback was removed together with this drop. +-- Pre-migration keys stop replaying and simply execute fresh on redelivery +-- (explicitly accepted when the dual-read window was waived). +-- package_invocation_tokens is unrelated auth state and stays. +DROP TABLE IF EXISTS package_invocations; diff --git a/packages/worker/src/account/data-targets.ts b/packages/worker/src/account/data-targets.ts index fd5ba0e525..7ef5cb6480 100644 --- a/packages/worker/src/account/data-targets.ts +++ b/packages/worker/src/account/data-targets.ts @@ -124,7 +124,6 @@ export function getAccountExportExcludedD1Surfaces(): Array<{ * tables) are not represented. */ export const accountUserDataTargets: ReadonlyArray = [ - { kind: 'user_id', table: 'package_invocations' }, { kind: 'user_id', table: 'package_invocation_tokens' }, { kind: 'user_id', table: 'workflow_runs' }, { kind: 'user_id', table: 'package_service_states' }, diff --git a/packages/worker/src/app/account-deletion.node.test.ts b/packages/worker/src/app/account-deletion.node.test.ts index 84a47030c3..413b2c013d 100644 --- a/packages/worker/src/app/account-deletion.node.test.ts +++ b/packages/worker/src/app/account-deletion.node.test.ts @@ -801,7 +801,6 @@ test('deleteUserAccount cascades user-scoped rows for the requested user', async ], mcp_user_server_instructions: [{ user_id: userAaa }], package_invocation_tokens: [{ id: 'pit-1', user_id: userAaa }], - package_invocations: [{ id: 'pi-1', user_id: userAaa }], agent_package_conversation_uses: [ { user_id: userAaa, diff --git a/packages/worker/src/app/account-retention-dispositions.ts b/packages/worker/src/app/account-retention-dispositions.ts index 1ca547ea30..7775700deb 100644 --- a/packages/worker/src/app/account-retention-dispositions.ts +++ b/packages/worker/src/app/account-retention-dispositions.ts @@ -5,7 +5,6 @@ export type AccountRetentionDisposition = export const accountRetentionDispositions: ReadonlyArray = [ - { table: 'package_invocations', kind: 'scheduled_policy' }, { table: 'mcp_memory_conversation_suppressions', kind: 'scheduled_policy', diff --git a/packages/worker/src/app/retention.node.test.ts b/packages/worker/src/app/retention.node.test.ts index dc89f5171f..56740977b2 100644 --- a/packages/worker/src/app/retention.node.test.ts +++ b/packages/worker/src/app/retention.node.test.ts @@ -9,13 +9,11 @@ import { entitlementDailyCounterRetentionDays, getRetentionPolicyCoverage, memorySuppressionRetentionDays, - packageInvocationRetentionDays, platformFeedbackRetentionDays, pruneAuditEventsForRetention, pruneEmailDeliveryEventsForRetention, pruneEntitlementDailyCountersForRetention, pruneMemorySuppressionsForRetention, - prunePackageInvocationsForRetention, prunePlatformFeedbackForRetention, prunePublishedBundleArtifactsForRetention, pruneRetention, @@ -99,22 +97,6 @@ function createD1FromSqlite( function createRetentionDb() { const sqlite = new DatabaseSync(':memory:') sqlite.exec(` - CREATE TABLE package_invocations ( - id TEXT PRIMARY KEY, - user_id TEXT NOT NULL, - token_id TEXT NOT NULL, - package_id TEXT NOT NULL, - package_kody_id TEXT NOT NULL, - export_name TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - request_hash TEXT NOT NULL, - source TEXT, - topic TEXT, - status TEXT NOT NULL, - response_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - ); CREATE TABLE workflow_runs ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, @@ -315,21 +297,23 @@ function insertEmailThread( ) } -function insertPackageInvocation( +function insertWorkflowRun( db: DatabaseSync, - input: { id: string; status?: string; createdAt: string }, + input: { id: string; status?: string | null; completedAt: string }, ) { db.prepare( - `INSERT INTO package_invocations ( - id, user_id, token_id, package_id, package_kody_id, export_name, - idempotency_key, request_hash, status, created_at, updated_at - ) VALUES (?, 'user-1', 'token', 'pkg-1', 'pkg-1', '.', ?, 'hash', ?, ?, ?)`, + `INSERT INTO workflow_runs ( + id, user_id, source_type, workflow_name, idempotency_key, run_at, + status, created_at, updated_at, completed_at + ) VALUES (?, 'user-1', 'inline', 'wf', ?, ?, ?, ?, ?, ?)`, ).run( input.id, `key-${input.id}`, - input.status ?? 'completed', - input.createdAt, - input.createdAt, + input.completedAt, + input.status === undefined ? 'complete' : input.status, + input.completedAt, + input.completedAt, + input.completedAt, ) } @@ -350,31 +334,8 @@ test('retention cron runs only on the hourly gate', () => { ) }) -test('package invocation and workflow retention keeps boundary and active idempotency rows', async () => { +test('workflow retention keeps boundary and non-terminal rows', async () => { const { sqlite, db } = createRetentionDb() - for (const [id, status, createdAt] of [ - [ - 'inv-old-completed', - 'completed', - daysAgo(packageInvocationRetentionDays + 1), - ], - ['inv-old-failed', 'failed', daysAgo(packageInvocationRetentionDays + 1)], - ['inv-boundary', 'completed', daysAgo(packageInvocationRetentionDays)], - [ - 'inv-in-progress', - 'in_progress', - daysAgo(packageInvocationRetentionDays + 1), - ], - ]) { - sqlite - .prepare( - `INSERT INTO package_invocations ( - id, user_id, token_id, package_id, package_kody_id, export_name, - idempotency_key, request_hash, status, response_json, created_at, updated_at - ) VALUES (?, 'user-1', 'token', 'pkg-1', 'pkg-1', '.', ?, 'hash', ?, '{}', ?, ?)`, - ) - .run(id, id, status, createdAt, createdAt) - } for (const [id, status, completedAt] of [ [ 'workflow-old-complete', @@ -386,29 +347,14 @@ test('package invocation and workflow retention keeps boundary and active idempo ['workflow-running', 'running', daysAgo(workflowRunRetentionDays + 1)], ['workflow-unknown', null, daysAgo(workflowRunRetentionDays + 1)], ] as const) { - sqlite - .prepare( - `INSERT INTO workflow_runs ( - id, user_id, source_type, workflow_name, idempotency_key, run_at, status, - created_at, updated_at, completed_at - ) VALUES (?, 'user-1', 'inline', 'wf', ?, ?, ?, ?, ?, ?)`, - ) - .run(id, id, completedAt, status, completedAt, completedAt, completedAt) + insertWorkflowRun(sqlite, { id, status, completedAt }) } - expect(await prunePackageInvocationsForRetention({ db, now })).toEqual({ - selected: 2, - deleted: 2, - }) expect(await pruneWorkflowRunsForRetention({ db, now })).toEqual({ selected: 2, deleted: 2, }) - expect(idsForTable(sqlite, 'package_invocations')).toEqual([ - 'inv-boundary', - 'inv-in-progress', - ]) expect(idsForTable(sqlite, 'workflow_runs')).toEqual([ 'workflow-boundary', 'workflow-running', @@ -865,14 +811,14 @@ test('retention prune reports selected separately from deleted when rows vanish const dbWithVanishingRow = { prepare(query: string) { const prepared = baseDb.prepare(query) - if (!query.includes('DELETE FROM package_invocations')) return prepared + if (!query.includes('DELETE FROM workflow_runs')) return prepared return { bind(...params: Array) { const bound = prepared.bind(...params) return { async run() { sqlite - .prepare(`DELETE FROM package_invocations WHERE id = 'inv-0'`) + .prepare(`DELETE FROM workflow_runs WHERE id = 'wf-0'`) .run() return bound.run() }, @@ -884,14 +830,14 @@ test('retention prune reports selected separately from deleted when rows vanish }, } as unknown as D1Database for (let index = 0; index < 2; index += 1) { - insertPackageInvocation(sqlite, { - id: `inv-${index}`, - createdAt: daysAgo(120), + insertWorkflowRun(sqlite, { + id: `wf-${index}`, + completedAt: daysAgo(120), }) } expect( - await prunePackageInvocationsForRetention({ + await pruneWorkflowRunsForRetention({ db: dbWithVanishingRow, now, batchSize: 2, @@ -1137,53 +1083,44 @@ test('published bundle artifact retention rechecks staleness before deleting sel test('retention pruning deletes only one configured batch per table invocation', async () => { const { sqlite, db } = createRetentionDb() for (let index = 0; index < 3; index += 1) { - sqlite - .prepare( - `INSERT INTO package_invocations ( - id, user_id, token_id, package_id, package_kody_id, export_name, - idempotency_key, request_hash, status, created_at, updated_at - ) VALUES (?, 'user-1', 'token', 'pkg-1', 'pkg-1', '.', ?, 'hash', 'completed', ?, ?)`, - ) - .run(`inv-${index}`, `key-${index}`, daysAgo(120), daysAgo(120)) + insertWorkflowRun(sqlite, { + id: `wf-${index}`, + completedAt: daysAgo(120), + }) } expect( - await prunePackageInvocationsForRetention({ db, now, batchSize: 2 }), + await pruneWorkflowRunsForRetention({ db, now, batchSize: 2 }), ).toEqual({ selected: 2, deleted: 2 }) - expect(idsForTable(sqlite, 'package_invocations')).toEqual(['inv-2']) + expect(idsForTable(sqlite, 'workflow_runs')).toEqual(['wf-2']) expect( - await prunePackageInvocationsForRetention({ db, now, batchSize: 2 }), + await pruneWorkflowRunsForRetention({ db, now, batchSize: 2 }), ).toEqual({ selected: 1, deleted: 1 }) - expect(idsForTable(sqlite, 'package_invocations')).toEqual([]) + expect(idsForTable(sqlite, 'workflow_runs')).toEqual([]) }) test('retention row deletes chunk ids to stay within the D1 binding limit', async () => { const { sqlite } = createRetentionDb() const db = createD1FromSqlite(sqlite, { maxBindings: 100 }) for (let index = 0; index < 101; index += 1) { - sqlite - .prepare( - `INSERT INTO package_invocations ( - id, user_id, token_id, package_id, package_kody_id, export_name, - idempotency_key, request_hash, status, created_at, updated_at - ) VALUES (?, 'user-1', 'token', 'pkg-1', 'pkg-1', '.', ?, 'hash', - 'completed', ?, ?)`, - ) - .run(`inv-${index}`, `key-${index}`, daysAgo(120), daysAgo(120)) + insertWorkflowRun(sqlite, { + id: `wf-${String(index).padStart(3, '0')}`, + completedAt: daysAgo(120), + }) } expect( - await prunePackageInvocationsForRetention({ db, now, batchSize: 101 }), + await pruneWorkflowRunsForRetention({ db, now, batchSize: 101 }), ).toEqual({ selected: 101, deleted: 101 }) - expect(idsForTable(sqlite, 'package_invocations')).toEqual([]) + expect(idsForTable(sqlite, 'workflow_runs')).toEqual([]) }) test('retention run loops batches per table until backlogs are drained', async () => { const { sqlite, db } = createRetentionDb() for (let index = 0; index < 501; index += 1) { - insertPackageInvocation(sqlite, { - id: `inv-${String(index).padStart(3, '0')}`, - createdAt: daysAgo(120), + insertWorkflowRun(sqlite, { + id: `wf-${String(index).padStart(3, '0')}`, + completedAt: daysAgo(120), }) } for (let index = 0; index < 300; index += 1) { @@ -1203,20 +1140,20 @@ test('retention run loops batches per table until backlogs are drained', async ( const result = await pruneRetention({ env, now }) - expect(result.packageInvocations).toBe(501) + expect(result.workflowRuns).toBe(501) expect(result.auditEvents).toBe(300) - expect(result.batchesPerTable['package_invocations']).toBe(3) + expect(result.batchesPerTable['workflow_runs']).toBe(3) expect(result.batchesPerTable['audit_events']).toBe(2) expect(result.timeBudgetExhausted).toBe(false) - expect(idsForTable(sqlite, 'package_invocations')).toEqual([]) + expect(idsForTable(sqlite, 'workflow_runs')).toEqual([]) }) test('retention run gives every table one batch and stops when the budget is exhausted', async () => { const { sqlite, db } = createRetentionDb() for (let index = 0; index < 300; index += 1) { - insertPackageInvocation(sqlite, { - id: `inv-${String(index).padStart(3, '0')}`, - createdAt: daysAgo(120), + insertWorkflowRun(sqlite, { + id: `wf-${String(index).padStart(3, '0')}`, + completedAt: daysAgo(120), }) } sqlite @@ -1236,11 +1173,11 @@ test('retention run gives every table one batch and stops when the budget is exh // The first round-robin pass always completes so a hot table cannot starve // the others, then the exhausted budget stops further passes. - expect(result.packageInvocations).toBe(250) + expect(result.workflowRuns).toBe(250) expect(result.auditEvents).toBe(1) - expect(result.batchesPerTable['package_invocations']).toBe(1) + expect(result.batchesPerTable['workflow_runs']).toBe(1) expect(result.timeBudgetExhausted).toBe(true) - expect(idsForTable(sqlite, 'package_invocations')).toHaveLength(50) + expect(idsForTable(sqlite, 'workflow_runs')).toHaveLength(50) }) test('retention coverage includes every live growth-pattern table or documented exemption', () => { diff --git a/packages/worker/src/app/retention.ts b/packages/worker/src/app/retention.ts index 77c5c60431..31bd5a09d0 100644 --- a/packages/worker/src/app/retention.ts +++ b/packages/worker/src/app/retention.ts @@ -43,7 +43,6 @@ export const publishedBundleArtifactRetentionBatchSize = 100 */ export const retentionRunTimeBudgetMs = 20_000 -export const packageInvocationRetentionDays = 90 export const memorySuppressionRetentionDays = 90 export const workflowRunRetentionDays = 90 export const platformFeedbackRetentionDays = 365 @@ -71,14 +70,6 @@ const terminalWorkflowStatusList = terminalWorkflowStatusValues * in `retention.node.test.ts` remains a second net for discovering new tables. */ export const retentionPolicies: ReadonlyArray = [ - { - table: 'package_invocations', - scope: 'per-user', - retentionDays: packageInvocationRetentionDays, - batchSize: retentionDefaultBatchSize, - description: - 'LEGACY (dual-read window): keyed invocations now claim and store responses in the per-user RunLog Durable Object ledger, which self-enforces the same 90-day terminal-row window. This sweep only drains pre-migration terminal rows; in-progress rows are never pruned. Remove with the follow-up that drops the table and the D1 fallback read.', - }, { table: 'mcp_memory_conversation_suppressions', scope: 'per-user', @@ -192,7 +183,6 @@ export const retentionPolicyExemptions: ReadonlyArray })) export type RetentionPruneResult = { - packageInvocations: number memorySuppressions: number workflowRuns: number platformFeedback: number @@ -352,29 +342,6 @@ async function deletePublishedBundleArtifactRowIfStillStale(input: { return result.meta.changes ?? 0 } -export async function prunePackageInvocationsForRetention(input: { - db: D1Database - now?: Date - batchSize?: number -}) { - const cutoff = cutoffIso( - input.now ?? new Date(), - packageInvocationRetentionDays, - ) - return selectAndDeleteByIds({ - db: input.db, - bindings: [cutoff, input.batchSize ?? retentionDefaultBatchSize], - sql: `SELECT id - FROM package_invocations - WHERE created_at < ? - AND status != 'in_progress' - ORDER BY created_at ASC, id ASC - LIMIT ?`, - table: 'package_invocations', - idColumn: 'id', - }) -} - export async function pruneMemorySuppressionsForRetention(input: { db: D1Database now?: Date @@ -899,7 +866,6 @@ export async function pruneRetention(input: { // for blob reasons so a blocked head cannot wedge later batches. let emailMessagesCursor: EmailMessageRetentionCursor | null = null const result: RetentionPruneResult = { - packageInvocations: 0, memorySuppressions: 0, workflowRuns: 0, platformFeedback: 0, @@ -945,13 +911,6 @@ export async function pruneRetention(input: { done: boolean run: () => Promise }> = [ - countTask( - 'package_invocations', - () => prunePackageInvocationsForRetention({ db, now }), - (count) => { - result.packageInvocations += count - }, - ), countTask( 'mcp_memory_conversation_suppressions', () => pruneMemorySuppressionsForRetention({ db, now }), diff --git a/packages/worker/src/email/inbound.workers.test.ts b/packages/worker/src/email/inbound.workers.test.ts index c5c4a29cae..2dd0b6060d 100644 --- a/packages/worker/src/email/inbound.workers.test.ts +++ b/packages/worker/src/email/inbound.workers.test.ts @@ -3119,32 +3119,6 @@ test( ON published_bundle_artifacts(user_id, source_id, artifact_kind, COALESCE(artifact_name, ''), entry_point)`, ) .run() - await db - .prepare( - `CREATE TABLE IF NOT EXISTS package_invocations ( - id TEXT PRIMARY KEY, - user_id TEXT NOT NULL, - token_id TEXT NOT NULL, - package_id TEXT NOT NULL, - package_kody_id TEXT NOT NULL, - export_name TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - request_hash TEXT NOT NULL, - source TEXT, - topic TEXT, - status TEXT NOT NULL, - response_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - )`, - ) - .run() - await db - .prepare( - `CREATE UNIQUE INDEX IF NOT EXISTS idx_package_invocations_key - ON package_invocations(user_id, token_id, package_id, export_name, idempotency_key)`, - ) - .run() const now = new Date().toISOString() await db diff --git a/packages/worker/src/email/system-email-subscriptions.workers.test.ts b/packages/worker/src/email/system-email-subscriptions.workers.test.ts index 6aa00b6df3..9249802f2a 100644 --- a/packages/worker/src/email/system-email-subscriptions.workers.test.ts +++ b/packages/worker/src/email/system-email-subscriptions.workers.test.ts @@ -64,24 +64,6 @@ async function ensurePackageSubscriptionTestSchema(db: D1Database) { )`, `CREATE UNIQUE INDEX IF NOT EXISTS idx_published_bundle_artifacts_identity ON published_bundle_artifacts(user_id, source_id, artifact_kind, COALESCE(artifact_name, ''), entry_point)`, - `CREATE TABLE IF NOT EXISTS package_invocations ( - id TEXT PRIMARY KEY, - user_id TEXT NOT NULL, - token_id TEXT NOT NULL, - package_id TEXT NOT NULL, - package_kody_id TEXT NOT NULL, - export_name TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - request_hash TEXT NOT NULL, - source TEXT, - topic TEXT, - status TEXT NOT NULL, - response_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - )`, - `CREATE UNIQUE INDEX IF NOT EXISTS idx_package_invocations_key - ON package_invocations(user_id, token_id, package_id, export_name, idempotency_key)`, ] for (const statement of statements) { await db.prepare(statement).run() diff --git a/packages/worker/src/entitlements/entitlements.node.test.ts b/packages/worker/src/entitlements/entitlements.node.test.ts index 5897a5c4b1..809bae902e 100644 --- a/packages/worker/src/entitlements/entitlements.node.test.ts +++ b/packages/worker/src/entitlements/entitlements.node.test.ts @@ -60,7 +60,6 @@ function createEntitlementsTestDb( | 'saved_packages' | 'jobs' | 'repo_sessions' - | 'package_invocations' | 'published_bundle_artifacts' | 'secret_entries' | 'value_entries' @@ -92,7 +91,6 @@ function createEntitlementsTestDb( 'entity_sources', 'jobs', 'repo_sessions', - 'package_invocations', 'published_bundle_artifacts', 'workflow_runs', ] as const diff --git a/packages/worker/src/entitlements/service.ts b/packages/worker/src/entitlements/service.ts index e1bb2bc1ba..d358a7596d 100644 --- a/packages/worker/src/entitlements/service.ts +++ b/packages/worker/src/entitlements/service.ts @@ -485,28 +485,10 @@ export async function readUserD1StorageBytes(input: { WHERE user_id = ?`, [userId], ), - sumStorageBytes( - db, - `SELECT COALESCE(SUM( - ${textBytesExpression([ - 'token_id', - 'package_id', - 'package_kody_id', - 'export_name', - 'idempotency_key', - 'request_hash', - 'source', - 'topic', - 'response_json', - ])} - ), 0) AS count - FROM package_invocations - WHERE user_id = ?`, - [userId], - ), - // Run records are observability history, not user content, and now live - // in a per-user RunLog Durable Object. They are intentionally excluded - // from the storage quota. + // Run records and the keyed package-invocation idempotency ledger live + // in the per-user RunLog Durable Object (the D1 package_invocations + // table was dropped). Both are self-expiring operational state, not + // user content, and are intentionally excluded from the storage quota. sumStorageBytes( db, `SELECT COALESCE(SUM( diff --git a/packages/worker/src/package-invocations/idempotency.ts b/packages/worker/src/package-invocations/idempotency.ts index f5f3f10464..85b877ff6e 100644 --- a/packages/worker/src/package-invocations/idempotency.ts +++ b/packages/worker/src/package-invocations/idempotency.ts @@ -4,8 +4,8 @@ import { buildJsonErrorResponse } from './responses.ts' import { type PackageInvocationStoredResponse } from './repo.ts' /** - * Store-agnostic view of an idempotency row: the RunLog DO ledger record and - * the legacy D1 `package_invocations` row (dual-read window) both map into it. + * The fields `resolveExistingInvocation` needs from a RunLog DO ledger + * record (mapped from `PackageInvocationLedgerRecord` at the call site). */ export type ResolvableInvocationRecord = { requestHash: string diff --git a/packages/worker/src/package-invocations/idempotent-module-invocation.ts b/packages/worker/src/package-invocations/idempotent-module-invocation.ts index 8b98d7f186..d34ef28c2c 100644 --- a/packages/worker/src/package-invocations/idempotent-module-invocation.ts +++ b/packages/worker/src/package-invocations/idempotent-module-invocation.ts @@ -27,11 +27,7 @@ import { import { type ensureModuleArtifact } from './module-artifacts.ts' import { runSavedPackageModuleOnce } from './module-execution.ts' import { buildJsonErrorResponse } from './responses.ts' -import { - boundedResponseJson, - getPackageInvocationByKey, - parseStoredResponse, -} from './repo.ts' +import { boundedResponseJson, parseStoredResponse } from './repo.ts' export const packageInvocationStaleAfterMs = 15 * 60 * 1000 const packageInvocationPollIntervalMs = 100 @@ -51,16 +47,6 @@ function toResolvableLedgerRecord( } } -function toResolvableLegacyRecord( - record: NonNullable>>, -): ResolvableInvocationRecord { - return { - requestHash: record.request_hash, - status: record.status, - storedResponse: record.storedResponse, - } -} - type ClaimedInvocation = { invocationId: string claimUpdatedAt: string @@ -75,10 +61,10 @@ type ClaimedInvocation = { * `runSavedPackageModuleEphemeral` in module-execution.ts instead — the * execution itself is shared via `runSavedPackageModuleOnce`. * - * During the dual-read window this path still READS the legacy D1 - * `package_invocations` table for keys claimed before the migration, but it - * never writes it; a follow-up removes the fallback together with the table - * and its retention sweep. + * The legacy D1 `package_invocations` table is retired: this path performs + * no D1 ledger reads or writes at all. Keys claimed before the DO migration + * no longer replay — a redelivery for such a key simply executes fresh + * (accepted when the dual-read window was waived). * * This path deliberately takes no `withAccountWriteLease`: the lease guards * D1 writes against concurrent account deletion, and no D1 writes remain @@ -206,107 +192,6 @@ export async function invokeSavedPackageModule(input: { userId: input.actor.userId, key: ledgerKey, }) - const lookupLegacy = async () => - await getPackageInvocationByKey({ - db: input.env.APP_DB, - userId: input.actor.userId, - tokenId: input.actor.tokenId, - packageId: input.savedPackage.id, - exportName: input.invocationName, - idempotencyKey: input.idempotencyKey, - }) - const releaseClaimBestEffort = async (claimed: ClaimedInvocation) => { - try { - await releasePackageInvocationRecord({ - env: input.env, - userId: input.actor.userId, - invocationId: claimed.invocationId, - claimUpdatedAt: claimed.claimUpdatedAt, - handle: claimed.handle, - }) - } catch (error) { - // Worst case the key stays claimed until the 15-minute stale - // reclaim — the same failure mode as an isolate death mid-run. - console.warn( - 'package invocation claim release failed', - getErrorMessage(error), - ) - } - } - - /** - * Dual-read fallback for keys claimed before the ledger moved into the - * RunLog DO. Runs after every claim (fresh or stale reclaim): a - * pre-migration key can sit terminal in D1 while a dead attempt's stale - * DO claim still exists — e.g. a takeover attempt died and a zombie - * pre-migration isolate later finished the legacy row — and replaying it - * beats re-executing. DO replays never reach this (terminal DO rows - * resolve before claiming). Returns the response to serve from the - * legacy row, or `null` when this attempt should proceed to execute. - * Awaited D1 reads only — never writes. - */ - const resolveLegacyFallback = async (claimed: ClaimedInvocation) => { - let legacy: Awaited> - try { - legacy = await lookupLegacy() - } catch (error) { - console.error('package invocation idempotency lookup failed', error) - await releaseClaimBestEffort(claimed) - return buildLookupFailedResponse() - } - if (!legacy) return null - const resolveLegacy = ( - record: NonNullable>>, - ) => - resolveExistingInvocation({ - record: toResolvableLegacyRecord(record), - requestHash, - idempotencyKey: input.idempotencyKey, - }) - if ( - legacy.request_hash !== requestHash || - legacy.status !== 'in_progress' - ) { - await releaseClaimBestEffort(claimed) - return resolveLegacy(legacy) - } - // A legacy in-progress row may still be finished by an old-code isolate - // during the deploy window, so poll it like the pre-migration path did. - const pollDeadline = Date.now() + packageInvocationPollBudgetMs - let current: Awaited> = legacy - while ( - current && - current.status === 'in_progress' && - !isStaleInvocation(current.updated_at, new Date()) && - Date.now() < pollDeadline - ) { - await new Promise((resolve) => - setTimeout(resolve, packageInvocationPollIntervalMs), - ) - try { - current = await lookupLegacy() - } catch (error) { - console.error('package invocation idempotency lookup failed', error) - await releaseClaimBestEffort(claimed) - return buildLookupFailedResponse() - } - } - if (!current) return null - if (current.status !== 'in_progress') { - await releaseClaimBestEffort(claimed) - return resolveLegacy(current) - } - if (!isStaleInvocation(current.updated_at, new Date())) { - await releaseClaimBestEffort(claimed) - return resolveLegacy(current) - } - // Stale legacy in-progress row: nothing writes D1 anymore, so it can - // never finish. Keep the DO claim and execute — the equivalent of the - // old D1 stale reclaim, recorded in the DO instead. The abandoned row - // is dropped with the table in the follow-up. - return null - } - let claim: Awaited> try { claim = await attemptClaim() @@ -342,22 +227,9 @@ export async function invokeSavedPackageModule(input: { } } if (!record) { - // The owner released its claim (transient artifact failure, or a - // dual-read fallback hit whose response lives in the legacy table). - let legacy: Awaited> - try { - legacy = await lookupLegacy() - } catch (error) { - console.error('package invocation idempotency lookup failed', error) - return buildLookupFailedResponse() - } - if (legacy) { - return resolveExistingInvocation({ - record: toResolvableLegacyRecord(legacy), - requestHash, - idempotencyKey: input.idempotencyKey, - }) - } + // The owner released its claim (transient artifact failure) and the + // key is free again; the caller retries rather than us re-claiming + // mid-poll. return buildJsonErrorResponse({ status: 500, code: 'idempotency_conflict_unresolved', @@ -384,9 +256,6 @@ export async function invokeSavedPackageModule(input: { claimUpdatedAt: claim.claimUpdatedAt, handle: claim.handle, } - const legacyResponse = await resolveLegacyFallback(claimed) - if (legacyResponse) return legacyResponse - const outcome = await runSavedPackageModuleOnce({ env: input.env, baseUrl: input.baseUrl, diff --git a/packages/worker/src/package-invocations/repo.ts b/packages/worker/src/package-invocations/repo.ts index 9b48dd0741..7e09360baa 100644 --- a/packages/worker/src/package-invocations/repo.ts +++ b/packages/worker/src/package-invocations/repo.ts @@ -12,18 +12,19 @@ import { import { toHex } from '@kody-internal/shared/hex.ts' /** - * Replay-cache ceiling for `response_json`. Rows above this would serialize - * into D1 export statements that D1's import path rejects (SQLITE_TOOBIG), - * making the whole backup un-restorable. Oversized terminal responses are - * stored as NULL; idempotent duplicates of those invocations get the - * existing `idempotency_response_unavailable` outcome instead of a replay. + * Replay-cache ceiling for the ledger's `response_json`. The ledger lives in + * the per-user RunLog Durable Object now, but the bound is kept at the + * restore-safe D1 column ceiling it inherited: replay payloads above it were + * never stored, so the DO rows stay small and export sections stay pageable. + * Oversized terminal responses are stored as NULL; idempotent duplicates of + * those invocations get the existing `idempotency_response_unavailable` + * outcome instead of a replay. */ export const maxStoredInvocationResponseJsonBytes = maxRestorableTextColumnBytes /** - * Serialize a terminal response for the replay cache, dropping it when - * oversized. Shared by the legacy D1 write path and the RunLog DO ledger so - * both stores enforce the same restore-safe bound. + * Serialize a terminal response for the RunLog DO ledger's replay cache, + * dropping it when oversized. */ export function boundedResponseJson( response: PackageInvocationStoredResponse, @@ -38,23 +39,6 @@ export function boundedResponseJson( return responseJson } -const packageInvocationRowSchema = object({ - id: string(), - user_id: string(), - token_id: string(), - package_id: string(), - package_kody_id: string(), - export_name: string(), - idempotency_key: string(), - request_hash: string(), - source: nullable(string()), - topic: nullable(string()), - status: string(), - response_json: nullable(string()), - created_at: string(), - updated_at: string(), -}) - const packageInvocationTokenRowSchema = object({ id: string(), user_id: string(), @@ -77,12 +61,6 @@ export type PackageInvocationStoredResponse = { body: Record } -export type PackageInvocationRecord = InferOutput< - typeof packageInvocationRowSchema -> & { - storedResponse: PackageInvocationStoredResponse | null -} - export type PackageInvocationTokenRecord = InferOutput< typeof packageInvocationTokenRowSchema > & { @@ -183,18 +161,6 @@ function mapTokenRow( } } -function mapRow(row: Record): PackageInvocationRecord { - const parsed = parseSafe(packageInvocationRowSchema, row) - if (!parsed.success) { - const message = parsed.issues.map((issue) => issue.message).join(', ') - throw new Error(`Invalid package invocation record: ${message}`) - } - return { - ...parsed.value, - storedResponse: parseStoredResponse(parsed.value.response_json), - } -} - export async function getActivePackageInvocationTokenByHash(input: { db: D1Database tokenHash: string @@ -403,38 +369,3 @@ export async function deletePackageInvocationToken(input: { .run() return (result.meta.changes ?? 0) > 0 } - -/** - * Dual-read fallback lookup for keys claimed before the idempotency ledger - * moved into the per-user RunLog Durable Object. The D1 table is read-only - * now — the write helpers were removed with the migration — and a follow-up - * removes this lookup together with the table and its retention sweep. - */ -export async function getPackageInvocationByKey(input: { - db: D1Database - userId: string - tokenId: string - packageId: string - exportName: string - idempotencyKey: string -}) { - const row = await input.db - .prepare( - `SELECT * - FROM package_invocations - WHERE user_id = ? - AND token_id = ? - AND package_id = ? - AND export_name = ? - AND idempotency_key = ?`, - ) - .bind( - input.userId, - input.tokenId, - input.packageId, - input.exportName, - input.idempotencyKey, - ) - .first>() - return row ? mapRow(row) : null -} diff --git a/packages/worker/src/package-invocations/service.node.test.ts b/packages/worker/src/package-invocations/service.node.test.ts index 1d3db8239f..731d515fa7 100644 --- a/packages/worker/src/package-invocations/service.node.test.ts +++ b/packages/worker/src/package-invocations/service.node.test.ts @@ -8,7 +8,6 @@ import { invokePackageExport, invokePackageSubscription, } from './service.ts' -import { createRequestHash } from './idempotency.ts' import { maxStoredInvocationResponseJsonBytes } from './repo.ts' const repoMockModule = vi.hoisted(() => ({ @@ -296,36 +295,25 @@ function createFakeRunLog(options: { failClaim?: boolean } = {}) { } /** - * Fake D1 with a READ-ONLY legacy `package_invocations` table for the - * dual-read window. Any attempted D1 write throws, which is the test-level - * proof that no awaited D1 write remains on the keyed hot path. + * Fake D1 that rejects EVERY `package_invocations` statement (the table is + * dropped; the ledger lives in the RunLog DO) and every write. Keyed tests + * passing against this is the proof that no D1 ledger read or write remains + * anywhere on the invoke path. */ function createDatabase(options: { failClaim?: boolean } = {}) { - const legacyRows: Array> = [] const runLog = createFakeRunLog(options) - const clone = (value: T): T => structuredClone(value) const db = { prepare(query: string) { + if (query.includes('package_invocations')) { + throw new Error( + `Unexpected D1 access to the dropped package_invocations table: ${query}`, + ) + } return { - bind(...params: Array) { + bind() { return { async first>() { - if ( - query.includes('FROM package_invocations') && - query.includes('idempotency_key = ?') - ) { - return clone( - legacyRows.find( - (row) => - row['user_id'] === params[0] && - row['token_id'] === params[1] && - row['package_id'] === params[2] && - row['export_name'] === params[3] && - row['idempotency_key'] === params[4], - ) ?? null, - ) as T | null - } - return null + return null as T | null }, async run() { throw new Error( @@ -337,70 +325,8 @@ function createDatabase(options: { failClaim?: boolean } = {}) { } }, runLog, - seedLegacyInvocation(row: { - tokenId: string - packageId: string - packageKodyId: string - exportName: string - idempotencyKey: string - requestHash: string - source?: string | null - topic?: string | null - status: 'in_progress' | 'completed' | 'failed' - responseJson?: string | null - updatedAt?: string - }) { - const updatedAt = row.updatedAt ?? new Date().toISOString() - const legacy = { - id: crypto.randomUUID(), - user_id: 'user-123', - token_id: row.tokenId, - package_id: row.packageId, - package_kody_id: row.packageKodyId, - export_name: row.exportName, - idempotency_key: row.idempotencyKey, - request_hash: row.requestHash, - source: row.source ?? null, - topic: row.topic ?? null, - status: row.status, - response_json: row.responseJson ?? null, - created_at: updatedAt, - updated_at: updatedAt, - } - legacyRows.push(legacy) - return clone(legacy) - }, - completeLegacyInvocation(input: { - idempotencyKey: string - responseJson: string - }) { - const row = legacyRows.find( - (candidate) => candidate['idempotency_key'] === input.idempotencyKey, - ) - if (!row) throw new Error('Expected legacy invocation row.') - row['status'] = 'completed' - row['response_json'] = input.responseJson - row['updated_at'] = new Date().toISOString() - }, } as unknown as D1Database & { runLog: ReturnType - seedLegacyInvocation(row: { - tokenId: string - packageId: string - packageKodyId: string - exportName: string - idempotencyKey: string - requestHash: string - source?: string | null - topic?: string | null - status: 'in_progress' | 'completed' | 'failed' - responseJson?: string | null - updatedAt?: string - }): Record - completeLegacyInvocation(input: { - idempotencyKey: string - responseJson: string - }): void } return db } @@ -1926,290 +1852,48 @@ test('oversized terminal responses are not stored so backups stay restorable', a expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) }) -async function dispatchMessageCreatedRequestHash( - params: Record, -) { - return await createRequestHash({ - packageId: 'pkg-1', - exportName: './dispatch-message-created', - params, - source: 'discord-gateway', - topic: 'discord.message.created', - }) -} - -function dispatchMessageCreatedRequest(idempotencyKey: string) { - return { - packageIdOrKodyId: 'discord-gateway', - exportName: 'dispatch-message-created', - params: { content: 'hi' }, - idempotencyKey, - source: 'discord-gateway', - topic: 'discord.message.created', - } -} - -test('dual-read window: a pre-migration terminal D1 row replays without executing', async () => { +test('a pre-migration-style key misses cleanly: it executes fresh instead of erroring', async () => { + // Keys claimed before the ledger moved into the RunLog DO no longer have + // any store to replay from (the D1 table is dropped). A redelivery for + // such a key is indistinguishable from a brand-new key: it claims in the + // DO and executes, without touching D1 at all (the fake D1 above throws + // on any package_invocations statement). const db = createDatabase() seedPackageResolution() repoMockModule.runBundledModuleWithRegistry.mockClear() - db.seedLegacyInvocation({ - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-pre-migration', - requestHash: await dispatchMessageCreatedRequestHash({ content: 'hi' }), - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'completed', - responseJson: JSON.stringify({ - status: 200, - body: { - ok: true, - result: { reply: 'stored before the migration' }, - idempotency: { key: 'evt-pre-migration', replayed: false }, - }, - }), - }) - - const replayed = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-pre-migration'), - }) - - expect(replayed.status).toBe(200) - expect(replayed.body).toMatchObject({ - ok: true, - result: { reply: 'stored before the migration' }, - idempotency: { key: 'evt-pre-migration', replayed: true }, + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + result: { reply: 'executed fresh' }, + logs: [], }) - expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() - // The DO claim taken ahead of the fallback lookup was released again, so - // nothing lingers in the ledger for the legacy-owned key. - expect(db.runLog.ledgerRows).toHaveLength(0) - // A different request against the same pre-migration key still mismatches. - const mismatch = await invokePackageExport({ + const response = await invokePackageExport({ env: createEnv(db), baseUrl: 'https://kody.dev', token: createToken(), request: { - ...dispatchMessageCreatedRequest('evt-pre-migration'), - params: { content: 'different' }, + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-from-before-the-migration', + source: 'discord-gateway', + topic: 'discord.message.created', }, }) - expect(mismatch.status).toBe(409) - expect(mismatch.body).toMatchObject({ - ok: false, - error: { code: 'idempotency_mismatch' }, - }) - expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() - expect(db.runLog.ledgerRows).toHaveLength(0) -}) -test('dual-read window: fresh legacy in-progress rows are polled, stale ones are taken over', async () => { - const db = createDatabase() - seedPackageResolution() - repoMockModule.runBundledModuleWithRegistry.mockClear() - repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ - result: { reply: 'executed after takeover' }, - logs: [], - }) - const requestHash = await dispatchMessageCreatedRequestHash({ - content: 'hi', - }) - // Fresh in-progress row finished by an old-code isolate mid-poll. - db.seedLegacyInvocation({ - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-legacy-open', - requestHash, - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'in_progress', - }) - const completionTimer = setTimeout(() => { - db.completeLegacyInvocation({ - idempotencyKey: 'evt-legacy-open', - responseJson: JSON.stringify({ - status: 200, - body: { - ok: true, - result: { reply: 'finished by an old isolate' }, - idempotency: { key: 'evt-legacy-open', replayed: false }, - }, - }), - }) - }, 150) - try { - const polled = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-legacy-open'), - }) - expect(polled.status).toBe(200) - expect(polled.body).toMatchObject({ - ok: true, - result: { reply: 'finished by an old isolate' }, - idempotency: { key: 'evt-legacy-open', replayed: true }, - }) - } finally { - clearTimeout(completionTimer) - } - expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() - expect(db.runLog.ledgerRows).toHaveLength(0) - - // A stale legacy in-progress row can never finish (nothing writes D1 any - // more), so the DO claim proceeds to execute — the old reclaim, recorded - // in the DO instead of D1. - db.seedLegacyInvocation({ - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-legacy-stale', - requestHash, - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'in_progress', - updatedAt: '2026-01-01T00:00:00.000Z', - }) - const takeover = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-legacy-stale'), - }) - expect(takeover.status).toBe(200) - expect(takeover.body).toMatchObject({ + expect(response.status).toBe(200) + expect(response.body).toMatchObject({ ok: true, - result: { reply: 'executed after takeover' }, - idempotency: { key: 'evt-legacy-stale', replayed: false }, + result: { reply: 'executed fresh' }, + idempotency: { key: 'evt-from-before-the-migration', replayed: false }, }) expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) expect( db.runLog.ledgerRows.find( - (row) => row.idempotencyKey === 'evt-legacy-stale', + (row) => row.idempotencyKey === 'evt-from-before-the-migration', ), ).toMatchObject({ status: 'completed' }) }) -test('dual-read window: a stale DO claim defers to a terminal pre-migration row instead of re-executing', async () => { - const db = createDatabase() - seedPackageResolution() - repoMockModule.runBundledModuleWithRegistry.mockClear() - const requestHash = await dispatchMessageCreatedRequestHash({ - content: 'hi', - }) - // A takeover attempt claimed the key in the DO and died (claim now - // stale), while a zombie pre-migration isolate finished the legacy row. - db.runLog.ledgerRows.push({ - id: crypto.randomUUID(), - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-zombie-finish', - requestHash, - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'in_progress', - responseJson: null, - createdAt: '2026-01-01T00:00:00.000Z', - updatedAt: '2026-01-01T00:00:00.000Z', - }) - db.seedLegacyInvocation({ - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-zombie-finish', - requestHash, - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'completed', - responseJson: JSON.stringify({ - status: 200, - body: { - ok: true, - result: { reply: 'finished by a zombie isolate' }, - idempotency: { key: 'evt-zombie-finish', replayed: false }, - }, - }), - }) - - const replayed = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-zombie-finish'), - }) - - expect(replayed.status).toBe(200) - expect(replayed.body).toMatchObject({ - ok: true, - result: { reply: 'finished by a zombie isolate' }, - idempotency: { key: 'evt-zombie-finish', replayed: true }, - }) - expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() - // The reclaimed DO claim was released; the legacy row owns the key again. - expect(db.runLog.ledgerRows).toHaveLength(0) -}) - -test('the RunLog DO ledger is read first; legacy D1 rows are only a fallback for unknown keys', async () => { - const db = createDatabase() - seedPackageResolution() - repoMockModule.runBundledModuleWithRegistry.mockClear() - repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ - result: { reply: 'do-backed execution' }, - logs: [], - }) - const first = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-do-first'), - }) - expect(first.status).toBe(200) - - // A conflicting legacy row appearing afterwards (e.g. a restored backup) - // must lose to the DO row: replay serves the DO-stored response. - db.seedLegacyInvocation({ - tokenId: 'discord-gateway', - packageId: 'pkg-1', - packageKodyId: 'discord-gateway', - exportName: './dispatch-message-created', - idempotencyKey: 'evt-do-first', - requestHash: await dispatchMessageCreatedRequestHash({ content: 'hi' }), - source: 'discord-gateway', - topic: 'discord.message.created', - status: 'completed', - responseJson: JSON.stringify({ - status: 200, - body: { ok: true, result: { reply: 'stale legacy copy' } }, - }), - }) - const replay = await invokePackageExport({ - env: createEnv(db), - baseUrl: 'https://kody.dev', - token: createToken(), - request: dispatchMessageCreatedRequest('evt-do-first'), - }) - expect(replay.status).toBe(200) - expect(replay.body).toMatchObject({ - ok: true, - result: { reply: 'do-backed execution' }, - idempotency: { key: 'evt-do-first', replayed: true }, - }) - expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) -}) - test('invokePackageExport enforces source scopes for wildcard tokens', async () => { const db = createDatabase() seedPackageResolution() diff --git a/packages/worker/src/package-registry/service.node.test.ts b/packages/worker/src/package-registry/service.node.test.ts index 8dd7a22df1..d8283bfff1 100644 --- a/packages/worker/src/package-registry/service.node.test.ts +++ b/packages/worker/src/package-registry/service.node.test.ts @@ -1113,7 +1113,6 @@ function createEntitlementsDatabase(input: { 'entity_sources', 'jobs', 'repo_sessions', - 'package_invocations', 'published_bundle_artifacts', ] if ( diff --git a/packages/worker/src/test-support/workers-seed.ts b/packages/worker/src/test-support/workers-seed.ts index 0f37ab57ac..3792049a8f 100644 --- a/packages/worker/src/test-support/workers-seed.ts +++ b/packages/worker/src/test-support/workers-seed.ts @@ -77,24 +77,6 @@ export async function ensurePackageSubscriptionTestSchema(db: D1Database) { )`, `CREATE UNIQUE INDEX IF NOT EXISTS idx_published_bundle_artifacts_identity ON published_bundle_artifacts(user_id, source_id, artifact_kind, COALESCE(artifact_name, ''), entry_point)`, - `CREATE TABLE IF NOT EXISTS package_invocations ( - id TEXT PRIMARY KEY, - user_id TEXT NOT NULL, - token_id TEXT NOT NULL, - package_id TEXT NOT NULL, - package_kody_id TEXT NOT NULL, - export_name TEXT NOT NULL, - idempotency_key TEXT NOT NULL, - request_hash TEXT NOT NULL, - source TEXT, - topic TEXT, - status TEXT NOT NULL, - response_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - )`, - `CREATE UNIQUE INDEX IF NOT EXISTS idx_package_invocations_key - ON package_invocations(user_id, token_id, package_id, export_name, idempotency_key)`, ] for (const statement of statements) { await db.prepare(statement).run() diff --git a/tools/migration-ledger.json b/tools/migration-ledger.json index 67b8d79fcb..f12cb70eab 100644 --- a/tools/migration-ledger.json +++ b/tools/migration-ledger.json @@ -467,6 +467,10 @@ { "filename": "0111-package-codemod-ledger.sql", "sha256": "d42350b7bfa88400d8133a8d8436dd6fa2e25e3ab7650e047c447e7f4704863e" + }, + { + "filename": "0112-drop-package-invocations.sql", + "sha256": "1b3c57a2b6d07f1400939e769a4f024c29180026e78ad77d066a9be31597f505" } ] }