Skip to content

fix(plugin-state): transact writer mutations before rollout #4101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TorichanCapgo merged 1 commit into from
Nov 13, 2026
Merged

fix(plugin-state): transact writer mutations before rollout #4101

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions cloudflare_workers/api/wrangler.jsonc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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"
}
]
},
Expand Down
36 changes: 18 additions & 18 deletions cloudflare_workers/plugin/wrangler.jsonc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -112,7 +112,7 @@
},
{
"binding": "PLUGIN_STATE_POINTERS",
"bucket_name": "capgo-plugin-state-pointers-prod"
"bucket_name": "capgo-plugin-state-pointers-prod-20261113"
}
],
"queues": {
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"
}
]
},
Expand Down
8 changes: 8 additions & 0 deletions docs/plugin-state-kv.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
38 changes: 23 additions & 15 deletions supabase/functions/_backend/triggers/plugin_state_sync.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,16 @@ const FORWARD_TIMEOUT_MS = 25_000
const FORWARDED_HEADER = 'x-capgo-plugin-state-forwarded'

type Db = ReturnType<typeof getDrizzleClient>
type Pg = Awaited<ReturnType<typeof getPgClient>>
type SyncContext = Context<MiddlewareKeyVariables>

// 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<T>(pg: Pg, run: (db: Db) => Promise<T>): Promise<T> {
return withPgTransaction(pg, client => run(getDrizzleClient(client, { logger: false })))
}

interface ClaimedApp {
app_id: string
jobs: { kind: 'app' | 'full' | 'device' | 'version' | 'manifest', ref: string }[]
Expand Down Expand Up @@ -834,8 +842,8 @@ async function syncApp(source: SourceReader, store: SyncStore, item: ClaimedApp,
}
}

async function claim(db: Db): Promise<ClaimResult> {
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<ClaimResult> {
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' }
}

Expand Down Expand Up @@ -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 {
Expand All @@ -929,10 +938,9 @@ export async function drainPluginStateSync(c: SyncContext, options: { budgetMs?:
let pg: Awaited<ReturnType<typeof getPgClient>> | 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
Expand Down Expand Up @@ -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[],
Expand All @@ -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

Expand All @@ -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++
Expand Down Expand Up @@ -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)
Expand Down
Loading
Loading