From 418f9ad395ae632450a23541736adf883582c377 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:29:21 -0500 Subject: [PATCH] fix(github): preserve terminal receipts across competing failure Guard the failure writer against another receiver's committed success. Prove the rollback race red/green and bounded HTTP body cancellation with unrelated-company progress. Co-Authored-By: Paperclip --- .../2026-10-30-github-connection-sleep.md | 12 ++++ .../github-connection-events.test.ts | 57 +++++++++++++++-- .../__tests__/github-connection-work.test.ts | 63 ++++++++++++++++--- .../src/services/github-connection-events.ts | 3 +- 4 files changed, 121 insertions(+), 14 deletions(-) diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md index 40f83708a71..6d381373b19 100644 --- a/doc/plans/2026-10-30-github-connection-sleep.md +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -47,6 +47,18 @@ grant. Binding-only retries use the existing credential refresh lease and curren secret resolver without repeating account/repository discovery. The initial OAuth callback uses the same saved debt and settlement path after connection activation. +Cloud binding calls retain the grant, its connection, and a company share lock. +That admission fence prevents a concurrent disable, revoke, or archive from +finishing before an already-admitted registration call. The API has no remote +revision fence that could safely replace those locks with two short transactions. +Each HTTP request and response body has a physical 15-second abort deadline. +Four calls run in an awaited batch; a failed batch leaves the saved debt and +releases the transaction after all four outcomes settle. Other companies can +continue through separate pool connections. Slow successful multi-batch passes +can delay writes to the same grant/connection and company status changes; the +schema permits up to 100 active and 100 removed installation IDs. This is not an +instance-wide lock or an idle hold. + Credential expiry remains enforced by the existing use path. Scheduled continuity checks run one hour before recorded access expiry or after the existing 30-day continuity window. A successful renewal that leaves the credential short-lived diff --git a/server/src/__tests__/github-connection-events.test.ts b/server/src/__tests__/github-connection-events.test.ts index aa6b1545078..68900aaa0db 100644 --- a/server/src/__tests__/github-connection-events.test.ts +++ b/server/src/__tests__/github-connection-events.test.ts @@ -257,7 +257,8 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { expect(connector.acknowledgeEvents).toHaveBeenCalledTimes(2); }); - it.each([false, true])("applies installation events once without discarding newer verified access (refreshed: %s)", async (refreshed) => { + it.each([false, true, "rollback-race"])("applies installation events once without discarding newer verified access (mode: %s)", async (mode) => { + const refreshed = mode === true; const companyId = randomUUID(); const applicationId = randomUUID(); const connectionId = randomUUID(); @@ -356,10 +357,54 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { }); try { const competing = githubConnectionEventService(db, { connector, now: () => currentTime }); - const outcomes = await Promise.all([service.pollOnce(), competing.pollOnce()]); - expect(outcomes.reduce((sum, result) => sum + result.processed, 0)).toBe(1); - expect(outcomes.reduce((sum, result) => sum + result.duplicate, 0)).toBe(1); - expect(outcomes.every(result => result.failed === 0)).toBe(true); + if (mode === "rollback-race") { + let enter!: () => void, release!: () => void, releaseFailure!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const mayRollback = new Promise(resolve => { release = resolve; }); + const mayMarkFailure = new Promise(resolve => { releaseFailure = resolve; }); + const transaction = db.transaction.bind(db), update = db.update.bind(db); + let first = true; + vi.spyOn(db, "transaction").mockImplementation(((callback: (tx: unknown) => Promise) => { + const rejectAfterApply = first; first = false; + return transaction(async tx => { + const value = await callback(tx); + if (rejectAfterApply) { enter(); await mayRollback; throw new Error("fixture rollback before receipt commit"); } + return value; + }); + }) as typeof db.transaction); + vi.spyOn(db, "update").mockImplementation(((table: unknown) => { + const builder = update(table as typeof connectionEventDeliveries); + if (table === connectionEventDeliveries) { + const set = builder.set.bind(builder); + builder.set = ((values: { status?: string }) => { + const statement = set(values); + if (values.status === "failed") { + const where = statement.where.bind(statement); + statement.where = (async (...args: Parameters) => { await mayMarkFailure; return where(...args); }) as typeof statement.where; + } + return statement; + }) as typeof builder.set; + } + return builder; + }) as typeof db.update); + const rejected = service.pollOnce(); + const firstResult = Promise.allSettled([rejected]); + try { + await entered; + const succeeding = competing.pollOnce(); + release(); + expect(await succeeding).toMatchObject({ processed: 1, failed: 0 }); + expect((await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, leasedEvent.id)))[0]!.status).toBe("processed"); + releaseFailure(); + expect(await rejected).toMatchObject({ failed: 1 }); + expect((await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, leasedEvent.id)))[0]!.status).toBe("processed"); + } finally { release(); releaseFailure(); await firstResult; vi.restoreAllMocks(); } + } else { + const outcomes = await Promise.all([service.pollOnce(), competing.pollOnce()]); + expect(outcomes.reduce((sum, result) => sum + result.processed, 0)).toBe(1); + expect(outcomes.reduce((sum, result) => sum + result.duplicate, 0)).toBe(1); + expect(outcomes.every(result => result.failed === 0)).toBe(true); + } } finally { unsubscribe(); } await health.expectDeadline(currentTime.getTime() + 900_000 - (refreshed ? 60_000 : 0)); await health.expectQuiet(); @@ -380,6 +425,6 @@ describeEmbeddedPostgres("GitHub connection event delivery", () => { connectionEventDeliveries.providerDeliveryId, leasedEvent.id, )); - expect(receipt).toMatchObject({ status: "processed", attempts: 1 }); + expect(receipt).toMatchObject({ status: "processed", attempts: mode === "rollback-race" ? 2 : 1 }); }); }); diff --git a/server/src/__tests__/github-connection-work.test.ts b/server/src/__tests__/github-connection-work.test.ts index a079c130f02..1a268d57605 100644 --- a/server/src/__tests__/github-connection-work.test.ts +++ b/server/src/__tests__/github-connection-work.test.ts @@ -31,10 +31,13 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( let db: ReturnType; let temporaryDatabase: Awaited> | undefined; let server: Server, origin: string; + let leaseTimes: number[] = []; let bindings: Array>, leases: number, failBindings: boolean, githubRequests: number; const signing = generateKeyPairSync("ed25519"), sealing = generateKeyPairSync("x25519"); let releaseRefresh: (() => void) | undefined, refreshArrived: (() => void) | undefined; let refreshGate: Promise | undefined; + let heldBindingCompany: string | undefined, bindingGate: Promise | undefined; + let bindingArrived: (() => void) | undefined, bindingClosed: (() => void) | undefined; const realFetch = globalThis.fetch; let companyIds: string[] = []; beforeAll(async () => { @@ -70,11 +73,16 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( res.setHeader("content-type", "application/json"); res.end(JSON.stringify({ sealed: { v: 1, alg: algorithm, purpose, provider: "github", profile: "github.code", epk: epk.toString("base64url"), iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); return; } if (req.url!.endsWith("/events/lease")) { - leases++; res.setHeader("content-type", "application/json"); + leases++; leaseTimes.push(Date.now()); res.setHeader("content-type", "application/json"); res.end(JSON.stringify({ events: [], ...(claims.ewv === 1 ? { eventWakeVersion: 1 } : {}) })); return; } expect(claims.sh).toBe(createHash("sha256").update(body.binding).digest("base64url")); bindings.push(JSON.parse(body.binding)); + if (claims.cid === heldBindingCompany) { + res.writeHead(200, { "content-type": "application/json" }); res.write("{"); + res.once("close", () => bindingClosed?.()); bindingArrived?.(); await bindingGate; + res.end("}"); return; + } res.writeHead(failBindings ? 503 : 200, { "content-type": "application/json" }).end("{}"); return; } githubRequests++; @@ -94,7 +102,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( afterAll(async () => { server.close(); await once(server, "close"); await db.$client.end({ timeout: 1 }); await temporaryDatabase?.cleanup(); }); async function fixture(ids: string[] = []) { - bindings = []; leases = 0; failBindings = false; githubRequests = 0; + bindings = []; leases = 0; leaseTimes = []; failBindings = false; githubRequests = 0; const companyId = randomUUID(); companyIds.push(companyId); const [company] = await db.insert(companies).values({ id: companyId, name: "GitHub work test", issuePrefix: `GW${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}` }).returning(); const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_server", status: "active" }).returning(); @@ -219,6 +227,39 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( } finally { releaseRefresh?.(); await result; refreshGate = undefined; refreshArrived = undefined; vi.unstubAllEnvs(); } }); + it("bounds a held Cloud body without releasing revocation admission, while another company progresses and restart retains removal debt", async () => { + const f = await fixture([]), other = await fixture([]); + for (const row of [f, other]) await db.update(connectionGrants).set({ providerTenant: { + ...row.grant.providerTenant, github: { ...row.grant.providerTenant!.github!, bindingSync: { id: randomUUID(), removeInstallationIds: ["101"] } }, + } }).where(eq(connectionGrants.id, row.grant.id)); + const latest = async (id: string) => (await db.select().from(connectionGrants).where(eq(connectionGrants.id, id)))[0]!; + heldBindingCompany = f.company.id; + let release!: () => void; + bindingGate = new Promise(resolve => { release = resolve; }); + const arrived = new Promise(resolve => { bindingArrived = resolve; }); + const closed = new Promise(resolve => { bindingClosed = resolve; }); + const timeout = AbortSignal.timeout.bind(AbortSignal); + const deadlines = vi.spyOn(AbortSignal, "timeout").mockImplementation(ms => timeout(ms === 15_000 ? 3_000 : ms)); + let completed = false; + const attempt = f.service.syncGitHubGrantBindings(f.connection, await latest(f.grant.id)); + const outcome = attempt.then(value => ({ value, error: null }), error => ({ value: null, error })).finally(() => { completed = true; }); + try { + await arrived; + const otherResult = await other.service.syncGitHubGrantBindings(other.connection, await latest(other.grant.id)); + expect(otherResult.providerTenant!.github!.bindingSync).toBeUndefined(); + await db.update(companies).set({ name: "Unrelated company changed while body held" }).where(eq(companies.id, other.company.id)); + expect(completed).toBe(false); + expect((await outcome).error).toMatchObject({ code: "CONNECTOR_BAD_RESPONSE" }); + await closed; + expect(deadlines).toHaveBeenCalledWith(15_000); + expect((await latest(f.grant.id)).providerTenant!.github!.bindingSync!.removeInstallationIds).toEqual(["101"]); + heldBindingCompany = undefined; release(); + const restarted = createGitHubGrantContinuity(db, f.dependencies); + expect(await restarted.sweepGitHubConnectionContinuity(f.company.id)).toMatchObject({ due: 1, refreshed: 1, failed: 0 }); + expect((await latest(f.grant.id)).providerTenant!.github!.bindingSync).toBeUndefined(); + } finally { heldBindingCompany = undefined; release(); await outcome; bindingGate = undefined; bindingArrived = undefined; bindingClosed = undefined; } + }); + it("lease acquire and cleanup cannot replace a concurrent webhook metadata revision", async () => { const f = await fixture(["101"]); let interleave = true; @@ -257,7 +298,9 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( const [unrelatedConnection] = await db.update(toolConnections).set({ config: {}, transportConfig: {}, transport: "rest_api", authKind: "none", credentialRefs: [], credentialSecretRefs: [] }).where(eq(toolConnections.id, other.connection.id)).returning(); await db.update(connectionGrants).set({ credentialSecretRefs: [] }).where(eq(connectionGrants.id, other.grant.id)); const api = toolAccessService(db); - const load = vi.spyOn(f.service, "githubConnectionContinuitySchedule"); + const loadTimes: number[] = []; + const schedule = f.service.githubConnectionContinuitySchedule.bind(f.service); + const load = vi.spyOn(f.service, "githubConnectionContinuitySchedule").mockImplementation(companyId => { loadTimes.push(Date.now()); return schedule(companyId); }); const sweep = vi.spyOn(f.service, "sweepGitHubConnectionContinuity"); const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true }); const continuity = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => true); @@ -266,7 +309,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( await Promise.all([continuity.ready, events.ready]); await vi.waitFor(() => expect(leases).toBe(1)); await vi.waitFor(() => expect(load).toHaveBeenCalledTimes(1)); - load.mockClear(); sweep.mockClear(); + load.mockClear(); loadTimes.length = 0; sweep.mockClear(); for (let i = 0; i < 30; i++) { await db.transaction(async tx => { await notifyToolConnectionHealth(tx, other.connection); await notifyChatEndpointConfiguration(tx, { id: randomUUID(), provider: "discord" }); }); await api.updateConnection(unrelatedConnection!.id, { name: `Unrelated patch ${i}` }); @@ -274,13 +317,19 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP ( await vi.advanceTimersByTimeAsync(10_000); expect(load).not.toHaveBeenCalled(); expect(sweep).not.toHaveBeenCalled(); expect(leases).toBe(1); for (let index = 0; index < 30; index++) await db.transaction(async tx => { await notifyGitHubCredentialChange(tx, f.connection); }); - expect(leases).toBe(1); + expect(leases).toBeLessThanOrEqual(2); await vi.advanceTimersByTimeAsync(6_000); await vi.waitFor(() => expect(load).toHaveBeenCalledWith(f.company.id)); expect(load.mock.calls.every(([companyId]) => companyId === f.company.id)).toBe(true); expect(sweep).not.toHaveBeenCalled(); - await vi.waitFor(() => expect(leases).toBe(2)); - expect(load).toHaveBeenCalledTimes(1); + await vi.waitFor(() => expect(leases).toBeGreaterThanOrEqual(2)); + // The first hint can run immediately; changes during it get one spaced pass. + expect(leases).toBeLessThanOrEqual(3); expect(load.mock.calls.length).toBeLessThanOrEqual(2); + if (loadTimes.length === 2) expect(loadTimes[1]! - loadTimes[0]!).toBeGreaterThanOrEqual(5_000); + if (leaseTimes.length === 3) expect(leaseTimes[2]! - leaseTimes[1]!).toBeGreaterThanOrEqual(5_000); + const quiet = { leases, loads: load.mock.calls.length }; + await vi.advanceTimersByTimeAsync(60_000); + expect({ leases, loads: load.mock.calls.length }).toEqual(quiet); } finally { await events.stop(); await continuity.stop(); await coordinator.stop(); } }); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index 5db8585ffcf..b0cc2f2fdd4 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -10,7 +10,7 @@ import { toolConnections, type Db, } from "@paperclipai/db"; -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, ne, sql } from "drizzle-orm"; import { logActivity, publishActivity, @@ -456,6 +456,7 @@ export function githubConnectionEventService( eq(connectionEventDeliveries.companyId, companyId), eq(connectionEventDeliveries.provider, event.provider), eq(connectionEventDeliveries.providerDeliveryId, event.id), + ne(connectionEventDeliveries.status, "processed"), )); throw error; }