From 0dc8fb21d69c609958ef8898d79c06f931f31c11 Mon Sep 17 00:00:00 2001 From: Martin Donadieu Date: Fri, 13 Nov 2026 02:41:07 +0000 Subject: [PATCH] fix(plugin-state): transact writer mutations before rollout --- cloudflare_workers/api/wrangler.jsonc | 4 +- cloudflare_workers/plugin/wrangler.jsonc | 36 ++-- docs/plugin-state-kv.md | 8 + .../_backend/triggers/plugin_state_sync.ts | 38 ++-- tests/plugin-state-source.unit.test.ts | 184 +++++++++++++++++- 5 files changed, 226 insertions(+), 44 deletions(-) diff --git a/cloudflare_workers/api/wrangler.jsonc b/cloudflare_workers/api/wrangler.jsonc index a3c47affb9..d2b7a324be 100644 --- a/cloudflare_workers/api/wrangler.jsonc +++ b/cloudflare_workers/api/wrangler.jsonc @@ -138,7 +138,7 @@ } ] }, - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_DIRECT_EU", @@ -148,7 +148,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, diff --git a/cloudflare_workers/plugin/wrangler.jsonc b/cloudflare_workers/plugin/wrangler.jsonc index 5b932e1d65..24054ed1b4 100644 --- a/cloudflare_workers/plugin/wrangler.jsonc +++ b/cloudflare_workers/plugin/wrangler.jsonc @@ -101,7 +101,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], // Stats of the requests the snippet answers itself (snippetEdgeReplay.ts): // Logpush -> R2 -> files queue -> replay queue, consumed by this worker only. // Create them first: bun run snippet-edge-stats:setup @@ -112,7 +112,7 @@ }, { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ], "queues": { @@ -231,7 +231,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -273,7 +273,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -329,7 +329,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -371,7 +371,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -427,7 +427,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -469,7 +469,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -525,7 +525,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -567,7 +567,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -623,7 +623,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -665,7 +665,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -721,7 +721,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -763,7 +763,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -819,7 +819,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -861,7 +861,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, @@ -917,7 +917,7 @@ "database_id": "81236a0c-db6e-454d-87da-944fa9bc100c" } ], - "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "c4524ce7f3aa4d43a649659070dcb65a" }], + "kv_namespaces": [{ "binding": "CHANNEL_SELF_STORE", "id": "2860ca0f6b1e416fb8651792895cbffa" }, { "binding": "PLUGIN_NOTIFICATION_QUEUE", "id": "655e307469ba496f9131556709b5a765" }, { "binding": "PLUGIN_STATE", "id": "90986b85f1234bbb84df43578d138260" }], "hyperdrive": [ { "binding": "HYPERDRIVE_CAPGO_READ_NA", @@ -947,7 +947,7 @@ "r2_buckets": [ { "binding": "PLUGIN_STATE_POINTERS", - "bucket_name": "capgo-plugin-state-pointers-prod" + "bucket_name": "capgo-plugin-state-pointers-prod-20261113" } ] }, diff --git a/docs/plugin-state-kv.md b/docs/plugin-state-kv.md index a22575375d..a9fa471985 100644 --- a/docs/plugin-state-kv.md +++ b/docs/plugin-state-kv.md @@ -166,6 +166,14 @@ fits one immutable KV document. Tests cover that size and a separate local checks, not a production throughput benchmark. Source reads use at most four concurrent primary transactions per drain. +Queue claims, acknowledgements, garbage-collection progress and explicit repair +enqueue calls also use separate short read-write transactions. These functions +mutate queue state even though they are invoked with `SELECT`; allowing +Hyperdrive to cache a custom function's result can replay a settled lease rather +than claim new work. Each mutation commits before KV/R2 operations start. The +writer does not hold a database transaction across provider I/O or disable +shared Hyperdrive caching. + Each bounded app snapshot, manifest, or source page uses `REPEATABLE READ, READ ONLY` through `withPgTransaction`. Its data is materialized, the transaction commits, and the checked-out client is released diff --git a/supabase/functions/_backend/triggers/plugin_state_sync.ts b/supabase/functions/_backend/triggers/plugin_state_sync.ts index 9e1d00b030..9e99d05501 100644 --- a/supabase/functions/_backend/triggers/plugin_state_sync.ts +++ b/supabase/functions/_backend/triggers/plugin_state_sync.ts @@ -69,8 +69,16 @@ const FORWARD_TIMEOUT_MS = 25_000 const FORWARDED_HEADER = 'x-capgo-plugin-state-forwarded' type Db = ReturnType +type Pg = Awaited> type SyncContext = Context +// These RPCs acquire leases or commit queue progress. A bare SELECT can be +// cached by Hyperdrive even though its custom function mutates the database. +// Commit each mutation before doing any KV/R2 work; keep shared read caching. +function mutate(pg: Pg, run: (db: Db) => Promise): Promise { + return withPgTransaction(pg, client => run(getDrizzleClient(client, { logger: false }))) +} + interface ClaimedApp { app_id: string jobs: { kind: 'app' | 'full' | 'device' | 'version' | 'manifest', ref: string }[] @@ -834,8 +842,8 @@ async function syncApp(source: SourceReader, store: SyncStore, item: ClaimedApp, } } -async function claim(db: Db): Promise { - const [row] = await rows<{ result: ClaimResult }>(db, sql`SELECT public.claim_plugin_state_sync(p_limit => ${CLAIM_LIMIT}::integer) AS result`) +async function claim(pg: Pg): Promise { + const [row] = await mutate(pg, db => rows<{ result: ClaimResult }>(db, sql`SELECT public.claim_plugin_state_sync(p_limit => ${CLAIM_LIMIT}::integer) AS result`)) return row?.result ?? { status: 'empty' } } @@ -896,14 +904,15 @@ export async function collectPluginStateGarbage(namespace: KvStore, pointers: Po } } -async function drainPluginStateGarbage(db: Db, namespace: KvStore, pointers: PointerStore, budget: PublicationBudget, deadline: number) { - const [claimed] = await rows<{ result: { lease_token: string, cursor: string | null } | null }>(db, sql`SELECT public.claim_plugin_state_gc() AS result`) - if (!claimed?.result) +async function drainPluginStateGarbage(pg: Pg, namespace: KvStore, pointers: PointerStore, budget: PublicationBudget, deadline: number) { + const [claimed] = await mutate(pg, db => rows<{ result: { lease_token: string, cursor: string | null } | null }>(db, sql`SELECT public.claim_plugin_state_gc() AS result`)) + const lease = claimed?.result + if (!lease) return - const progress = await collectPluginStateGarbage(namespace, pointers, { budget, deadline, cursor: claimed.result.cursor }) - await db.execute(sql`SELECT public.ack_plugin_state_gc( - ${claimed.result.lease_token}::uuid, ${progress.cursor}::text, ${progress.complete}::boolean - )`) + const progress = await collectPluginStateGarbage(namespace, pointers, { budget, deadline, cursor: lease.cursor }) + await mutate(pg, db => db.execute(sql`SELECT public.ack_plugin_state_gc( + ${lease.lease_token}::uuid, ${progress.cursor}::text, ${progress.complete}::boolean + )`)) } export interface PluginStateDrainResult { @@ -929,10 +938,9 @@ export async function drainPluginStateSync(c: SyncContext, options: { budgetMs?: let pg: Awaited> | null = null try { pg = await getPgClient(c) - const db = getDrizzleClient(pg, { logger: false }) let busyRetries = 0 while (Date.now() < deadline && budget.remaining > 0) { - const claimed = await claim(db) + const claimed = await claim(pg) if (claimed.status === 'busy') { if (++busyRetries > MAX_BUSY_RETRIES) break @@ -970,7 +978,7 @@ export async function drainPluginStateSync(c: SyncContext, options: { budgetMs?: // ACK validates the lease and persists the next cursor atomically. A // stale drain must never enqueue an older continuation after replacement. const continuations = Object.fromEntries(done.filter(outcome => outcome.continuation !== undefined).map(outcome => [outcome.appId, outcome.continuation])) - await db.execute(sql`SELECT public.ack_plugin_state_sync( + await mutate(pg, db => db.execute(sql`SELECT public.ack_plugin_state_sync( p_lease_token => ${claimed.lease_token}::uuid, p_done => ${sql.param(done.map(outcome => outcome.appId))}::text[], p_failed => ${sql.param(failed.map(outcome => outcome.appId))}::text[], @@ -982,7 +990,7 @@ export async function drainPluginStateSync(c: SyncContext, options: { budgetMs?: p_changed_apps => ${sql.param(outcomes.filter(outcome => outcome.changed).map(outcome => outcome.appId))}::text[], p_continuations => ${JSON.stringify(continuations)}::jsonb, p_app_published => ${sql.param(done.filter(outcome => outcome.appPublished).map(outcome => outcome.appId))}::text[] - )`) + )`)) result.apps += done.length result.failed += failed.length @@ -994,7 +1002,7 @@ export async function drainPluginStateSync(c: SyncContext, options: { budgetMs?: break } if (Date.now() < deadline && budget.remaining >= 3) - await drainPluginStateGarbage(db, namespace, pointers, budget, deadline) + await drainPluginStateGarbage(pg, namespace, pointers, budget, deadline) } catch (error) { result.failed++ @@ -1046,7 +1054,7 @@ app.post('/', middlewareAPISecret, async (c) => { if (apps.length > 0) { const pg = await getPgClient(c) try { - await getDrizzleClient(pg, { logger: false }).execute(sql`SELECT public.enqueue_plugin_state_full_sync(${sql.param(apps)}::text[])`) + await mutate(pg, db => db.execute(sql`SELECT public.enqueue_plugin_state_full_sync(${sql.param(apps)}::text[])`)) } finally { closeClient(c, pg) diff --git a/tests/plugin-state-source.unit.test.ts b/tests/plugin-state-source.unit.test.ts index 825ccb5be4..80ea742450 100644 --- a/tests/plugin-state-source.unit.test.ts +++ b/tests/plugin-state-source.unit.test.ts @@ -3,9 +3,9 @@ import type { Context } from 'hono' import type { MiddlewareKeyVariables } from '../supabase/functions/_backend/utils/hono.ts' import { PgDialect } from 'drizzle-orm/pg-core' import { Hono } from 'hono/tiny' -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { pluginStateVersionKey } from '../supabase/functions/_backend/plugin_runtime/utils/pluginStateFormat.ts' -import { buildManifestDoc, createPluginStateSourceReader, drainPluginStateSync } from '../supabase/functions/_backend/triggers/plugin_state_sync.ts' +import { app as syncApp, buildManifestDoc, createPluginStateSourceReader, drainPluginStateSync } from '../supabase/functions/_backend/triggers/plugin_state_sync.ts' import { publicationProviders } from './plugin-state-publication-fixture.ts' interface Connection { id: number, inTransaction: boolean } @@ -41,9 +41,10 @@ beforeEach(() => { peakTransactions = 0 nextConnection = 0 provider.connect.mockImplementation(async () => ({ id: nextConnection++, inTransaction: false })) - provider.transaction.mockImplementation(async (client: Connection, run: (client: Connection) => Promise, options: { appendSql?: string }) => { + provider.transaction.mockImplementation(async (client: Connection, run: (client: Connection) => Promise, options?: { appendSql?: string }) => { expect(client.inTransaction).toBe(false) - expect(options.appendSql).toBe('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY') + if (options) + expect(options.appendSql).toBe('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY') client.inTransaction = true activeTransactions++ peakTransactions = Math.max(peakTransactions, activeTransactions) @@ -60,6 +61,10 @@ beforeEach(() => { }) }) +afterEach(() => { + vi.unstubAllEnvs() +}) + async function inContext(run: (context: Context) => Promise) { const app = new Hono() app.get('/', async (c) => { @@ -72,13 +77,13 @@ async function inContext(run: (context: Context) => Prom } describe('fresh plugin snapshot source transactions', () => { - it('publishes the fresh primary row and commits before provider I/O or ACK', async () => { + it('commits the source snapshot before provider I/O and acknowledges in its own transaction', async () => { const storage = publicationProviders() let claims = 0 provider.execute.mockImplementation(async (client: Connection, query: SQL) => { const statement = dialect.sqlToQuery(query) if (statement.sql.includes('public.claim_plugin_state_sync(')) { - expect(client.inTransaction).toBe(false) + expect(client.inTransaction).toBe(true) claims++ return { rows: [{ result: { status: 'ok', lease_token: 'fixture-lease', generation: '1', has_more: false, apps: [{ app_id: 'com.test.freshsnapshot', jobs: [{ kind: 'version', ref: '1.0.0' }], attempts: 1 }], @@ -88,8 +93,8 @@ describe('fresh plugin snapshot source transactions', () => { // Plain queries emulate a previously warmed Hyperdrive result. return { rows: [{ id: client.inTransaction ? 202 : 101, owner_org: 'fixture-org', name: '1.0.0', deleted: false }] } } - expect(client.inTransaction).toBe(false) - expect(activeTransactions).toBe(0) + expect(client.inTransaction).toBe(true) + expect(activeTransactions).toBe(1) return { rows: [] } }) const originalPut = storage.namespace.put.getMockImplementation()! @@ -108,7 +113,7 @@ describe('fresh plugin snapshot source transactions', () => { expect(await response.json()).toEqual({ claims: 1, apps: 1, failed: 0 }) expect(claims).toBe(1) expect(storage.publishedValue(await pluginStateVersionKey('com.test.freshsnapshot', '1.0.0'))).toMatchObject({ id: 202 }) - expect(provider.transaction).toHaveBeenCalledTimes(1) + expect(provider.transaction).toHaveBeenCalledTimes(4) expect(provider.close).toHaveBeenCalledTimes(2) }) @@ -190,6 +195,167 @@ describe('fresh plugin snapshot source transactions', () => { }) }) +describe('plugin state queue mutation transactions', () => { + it('acquires new leases and publishes the next work item when plain RPC results would be cached', async () => { + const storage = publicationProviders() + const appId = 'com.test.mutationqueue' + const names = ['1.0.0', '2.0.0'] + const acknowledged: string[] = [] + let completed = 0 + let cachedClaims = 0 + const claimResult = (index: number) => ({ + status: 'ok', + lease_token: `fixture-lease-${index + 1}`, + generation: String(index + 1), + has_more: true, + apps: [{ app_id: appId, jobs: [{ kind: 'version', ref: names[index] }], attempts: 1 }], + }) + const cachedFirstClaim = claimResult(0) + provider.execute.mockImplementation(async (client: Connection, query: SQL) => { + const statement = dialect.sqlToQuery(query) + if (statement.sql.includes('public.claim_plugin_state_sync(')) { + if (!client.inTransaction) { + // Stop a broken drain quickly after demonstrating replay of an old lease. + if (++cachedClaims > 2) + throw new Error('cached mutation claim repeated after its ACK') + return { rows: [{ result: cachedFirstClaim }] } + } + return { rows: [{ result: completed < names.length ? claimResult(completed) : { status: 'empty' } }] } + } + if (statement.sql.includes('public.ack_plugin_state_sync(')) { + const lease = statement.params[0] + if (client.inTransaction && lease === `fixture-lease-${completed + 1}`) { + acknowledged.push(lease) + completed++ + } + return { rows: [] } + } + if (statement.sql.includes('public.claim_plugin_state_gc(')) { + expect(client.inTransaction).toBe(true) + return { rows: [{ result: null }] } + } + if (statement.sql.includes('FROM public.app_versions av')) { + expect(client.inTransaction).toBe(true) + const name = names.find(value => statement.params.some(param => Array.isArray(param) && param.includes(value))) + if (!name) + throw new Error('expected a queued version name') + return { rows: [{ id: name === names[0] ? 101 : 202, owner_org: 'fixture-org', name, deleted: false }] } + } + throw new Error(`unexpected queue query: ${statement.sql}`) + }) + const originalGet = storage.pointers.get.getMockImplementation()! + storage.pointers.get.mockImplementation(async (...args) => { + expect(activeTransactions).toBe(0) + return originalGet(...args) + }) + const originalPut = storage.namespace.put.getMockImplementation()! + storage.namespace.put.mockImplementation(async (...args) => { + expect(activeTransactions).toBe(0) + return originalPut(...args) + }) + const originalPointerPut = storage.pointers.put.getMockImplementation()! + storage.pointers.put.mockImplementation(async (...args) => { + expect(activeTransactions).toBe(0) + return originalPointerPut(...args) + }) + const app = new Hono() + app.post('/', async c => c.json(await drainPluginStateSync(c))) + const response = await app.request('/', { method: 'POST' }, { PLUGIN_STATE: storage.namespace, PLUGIN_STATE_POINTERS: storage.pointers }) + expect(await response.json()).toEqual({ claims: 2, apps: 2, failed: 0 }) + expect(acknowledged).toEqual(['fixture-lease-1', 'fixture-lease-2']) + expect(cachedClaims).toBe(0) + expect(storage.publishedValue(await pluginStateVersionKey(appId, names[0]))).toMatchObject({ id: 101 }) + expect(storage.publishedValue(await pluginStateVersionKey(appId, names[1]))).toMatchObject({ id: 202 }) + expect(activeTransactions).toBe(0) + }) + + it('commits each garbage-collection lease and cursor before resuming the next page', async () => { + const storage = publicationProviders() + let cursor: string | null = null + let complete = false + let lease = 0 + const acknowledgements: unknown[][] = [] + provider.execute.mockImplementation(async (client: Connection, query: SQL) => { + const statement = dialect.sqlToQuery(query) + expect(client.inTransaction).toBe(true) + if (statement.sql.includes('public.claim_plugin_state_sync(')) + return { rows: [{ result: { status: 'empty' } }] } + if (statement.sql.includes('public.claim_plugin_state_gc(')) { + lease++ + return { rows: [{ result: complete ? null : { lease_token: `fixture-gc-${lease}`, cursor } }] } + } + if (statement.sql.includes('public.ack_plugin_state_gc(')) { + expect(statement.params[0]).toBe(`fixture-gc-${lease}`) + const nextCursor = statement.params[1] + if (nextCursor !== null && typeof nextCursor !== 'string') + throw new Error('invalid GC cursor') + cursor = nextCursor + complete = statement.params[2] === true + acknowledgements.push(statement.params) + return { rows: [] } + } + throw new Error(`unexpected garbage-collection query: ${statement.sql}`) + }) + const listedCursors: (string | undefined)[] = [] + storage.namespace.list.mockImplementation(async (options) => { + expect(activeTransactions).toBe(0) + listedCursors.push(options.cursor) + return options.cursor + ? { keys: [], list_complete: true, cursor: undefined } + : { keys: [], list_complete: false, cursor: 'fixture-gc-page-2' } + }) + const app = new Hono() + app.post('/', async c => c.json(await drainPluginStateSync(c))) + for (let pass = 0; pass < 2; pass++) { + const response = await app.request('/', { method: 'POST' }, { PLUGIN_STATE: storage.namespace, PLUGIN_STATE_POINTERS: storage.pointers }) + expect(await response.json()).toEqual({ claims: 0, apps: 0, failed: 0 }) + } + expect(listedCursors).toEqual([undefined, 'fixture-gc-page-2']) + expect(acknowledgements).toEqual([ + ['fixture-gc-1', 'fixture-gc-page-2', false], + ['fixture-gc-2', null, true], + ]) + expect(complete).toBe(true) + expect(activeTransactions).toBe(0) + }) + + it('commits an explicit repair enqueue before the request begins draining', async () => { + vi.stubEnv('API_SECRET', 'fixture-api-secret') + const storage = publicationProviders() + const operations: string[] = [] + const appId = 'com.test.explicitmutationrepair' + provider.execute.mockImplementation(async (client: Connection, query: SQL) => { + const statement = dialect.sqlToQuery(query) + expect(client.inTransaction).toBe(true) + expect(activeTransactions).toBe(1) + if (statement.sql.includes('public.enqueue_plugin_state_full_sync(')) { + operations.push('enqueue') + expect(statement.params).toEqual([[appId]]) + return { rows: [] } + } + if (statement.sql.includes('public.claim_plugin_state_sync(')) { + operations.push('sync-claim') + return { rows: [{ result: { status: 'empty' } }] } + } + if (statement.sql.includes('public.claim_plugin_state_gc(')) { + operations.push('gc-claim') + return { rows: [{ result: null }] } + } + throw new Error(`unexpected repair query: ${statement.sql}`) + }) + const response = await syncApp.request('/', { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'apisecret': 'fixture-api-secret' }, + body: JSON.stringify({ apps: [appId], drain: true }), + }, { API_SECRET: 'fixture-api-secret', PLUGIN_STATE: storage.namespace, PLUGIN_STATE_POINTERS: storage.pointers }) + expect(response.status, await response.clone().text()).toBe(200) + expect(await response.json()).toEqual({ status: 'ok', claims: 0, apps: 0, failed: 0 }) + expect(operations).toEqual(['enqueue', 'sync-claim', 'gc-claim']) + expect(provider.close).toHaveBeenCalledTimes(2) + expect(activeTransactions).toBe(0) + }) +}) + describe('competing full backfills', () => { it('advances all four app cursors with one shared 800-operation provider budget', async () => { const storage = publicationProviders()