diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f9d3f16f..1ed32e6f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,15 @@ Newest first. `Unreleased` is what is on `main` and not yet tagged. ## Unreleased +### A failed shared-app approval can be tried again + +Deciding a shared-app request now commits the decision and the access it grants together. If the +approval or its audience cannot be saved, the request stays pending instead of disappearing as +approved without granting access. An administrator can retry after the write failure is resolved. +Concurrent direct approvals and request decisions take the same Bot/app transaction lock before +writing either row, so they do not deadlock by acquiring those rows in opposite orders. An owner +grant that races an administrator now leaves the administrator's approval and audience in place. + **Before upgrading.** Sixteen things change for an existing deployment: - Bots now share one computer per person instead of each having its own. Migration `0069_isolated_bots` adds `agent_profiles.isolated`, off for every existing Bot. With the diff --git a/server/src/plugins/shared-use-store.ts b/server/src/plugins/shared-use-store.ts index 59ef1f14c..053d1d169 100644 --- a/server/src/plugins/shared-use-store.ts +++ b/server/src/plugins/shared-use-store.ts @@ -32,6 +32,8 @@ import { steeringOf, } from "./shared-use"; +type Transaction = Parameters[0]>[0]; + export class SharedUseRequestDecidedError extends Error { constructor() { super("That request has already been decided."); @@ -141,8 +143,9 @@ export function createSharedUseStore( */ async function rowsOfApp( serverId: string, + executor: Database | Transaction = database, ): Promise<{ appId: string; url: string | null; ids: string[] }> { - const [row] = await database + const [row] = await executor .select({ url: mcpServers.url }) .from(mcpServers) .where(eq(mcpServers.id, serverId)) @@ -151,7 +154,7 @@ export function createSharedUseStore( const toolkit = url ? appOfUrl(url) : null; if (!toolkit) return { appId: serverId, url, ids: [serverId] }; const canonical = canonicalUrlOf(toolkit); - const siblings = await database + const siblings = await executor .select({ id: mcpServers.id }) .from(mcpServers) .where(eq(mcpServers.url, canonical)) @@ -170,8 +173,11 @@ export function createSharedUseStore( * read and write of an approval or a request goes through this first, and callers may hand in * either id and land on the same key. */ - async function appIdOf(serverId: string): Promise { - return (await rowsOfApp(serverId)).appId; + async function appIdOf( + serverId: string, + executor: Database | Transaction = database, + ): Promise { + return (await rowsOfApp(serverId, executor)).appId; } /** The approval row for this Bot at the app, whoever set it, and whether it may be read now. */ @@ -241,86 +247,116 @@ export function createSharedUseStore( }; } - async function setApproval(input: { + type ApprovalInput = { botId: string; serverId: string; approval: SharedUseApproval; by: string; /** Who is setting it. Only {@link approveOwnerGrant} says `owner`. */ as?: "admin" | "owner"; - }) { - const serverId = await appIdOf(input.serverId); - const approvedAs = input.as ?? "admin"; + }; + + async function setApproval(input: ApprovalInput) { await database.transaction(async (tx) => { - await tx - .insert(sharedUseApprovals) - .values({ + const serverId = await lockApproval(tx, input.botId, input.serverId); + await writeApproval(tx, { ...input, serverId }); + }); + } + + /** Lock the canonical Bot/app before either a request or its approval row is written. */ + async function lockApproval( + tx: Transaction, + botId: string, + serverId: string, + ) { + const appId = await appIdOf(serverId, tx); + await tx.execute( + sql`select pg_advisory_xact_lock(hashtext(${JSON.stringify(["shared-use-approval", botId, appId])}))`, + ); + return appId; + } + + /** Share the decision transaction; callers already hold the canonical Bot/app lock. */ + async function writeApproval(tx: Transaction, input: ApprovalInput) { + const serverId = input.serverId; + const approvedAs = input.as ?? "admin"; + const insertion = tx.insert(sharedUseApprovals).values({ + agentId: input.botId, + serverId, + audience: input.approval.audience, + outsideInput: input.approval.outsideInput, + approvedBy: input.by, + approvedAs, + }); + const target = [sharedUseApprovals.agentId, sharedUseApprovals.serverId]; + // An owner may fill an absence, never replace an approval another transaction just committed. + const written = + approvedAs === "owner" + ? await insertion + .onConflictDoNothing({ target }) + .returning({ botId: sharedUseApprovals.agentId }) + : await insertion + .onConflictDoUpdate({ + target, + set: { + audience: input.approval.audience, + outsideInput: input.approval.outsideInput, + approvedBy: input.by, + approvedAs, + approvedAt: new Date(), + }, + }) + .returning({ botId: sharedUseApprovals.agentId }); + if (written.length === 0) return false; + await tx + .delete(sharedUseApprovalMembers) + .where( + and( + eq(sharedUseApprovalMembers.agentId, input.botId), + eq(sharedUseApprovalMembers.serverId, serverId), + ), + ); + if ( + input.approval.audience === "people" && + input.approval.members.length > 0 + ) { + await tx.insert(sharedUseApprovalMembers).values( + input.approval.members.map((member) => ({ agentId: input.botId, serverId, - audience: input.approval.audience, - outsideInput: input.approval.outsideInput, - approvedBy: input.by, - approvedAs, - }) - .onConflictDoUpdate({ - target: [sharedUseApprovals.agentId, sharedUseApprovals.serverId], - set: { - audience: input.approval.audience, - outsideInput: input.approval.outsideInput, - approvedBy: input.by, - approvedAs, - approvedAt: new Date(), - }, - }); - await tx - .delete(sharedUseApprovalMembers) - .where( - and( - eq(sharedUseApprovalMembers.agentId, input.botId), - eq(sharedUseApprovalMembers.serverId, serverId), - ), - ); - if ( - input.approval.audience === "people" && - input.approval.members.length > 0 - ) { - await tx.insert(sharedUseApprovalMembers).values( - input.approval.members.map((member) => ({ - agentId: input.botId, - serverId, - ...member, - })), - ); - } - /* A pending request this approval now answers is closed as approved, so the inbox never asks twice. */ - const pending = await tx - .select() - .from(sharedUseRequests) - .where( - and( - eq(sharedUseRequests.agentId, input.botId), - eq(sharedUseRequests.serverId, serverId), - eq(sharedUseRequests.status, "pending"), - ), - ); - for (const request of pending) { - const proposed = { - audience: request.proposedAudience, - outsideInput: request.proposedOutsideInput, - members: request.proposedMembers, - }; - if (covers(input.approval, proposed)) { - await tx - .update(sharedUseRequests) - .set({ - status: "approved", - decidedBy: input.by, - decidedAt: new Date(), - }) - .where(eq(sharedUseRequests.id, request.id)); - } + ...member, + })), + ); + } + /* A pending request this approval now answers is closed as approved, so the inbox never asks twice. */ + const pending = await tx + .select() + .from(sharedUseRequests) + .where( + and( + eq(sharedUseRequests.agentId, input.botId), + eq(sharedUseRequests.serverId, serverId), + eq(sharedUseRequests.status, "pending"), + ), + ); + for (const request of pending) { + const proposed = { + audience: request.proposedAudience, + outsideInput: request.proposedOutsideInput, + members: request.proposedMembers, + }; + if (covers(input.approval, proposed)) { + await tx + .update(sharedUseRequests) + .set({ + status: "approved", + decidedBy: input.by, + decidedAt: new Date(), + }) + .where(eq(sharedUseRequests.id, request.id)); } - }); + } + return true; } /* @@ -372,14 +408,17 @@ export function createSharedUseStore( ...exposureOf(await botFacts(input.botId)), outsideInput: false, }; - await setApproval({ - botId: input.botId, - serverId, - approval, - by: input.by, - as: "owner", + const written = await database.transaction(async (tx) => { + const appId = await lockApproval(tx, input.botId, serverId); + return writeApproval(tx, { + botId: input.botId, + serverId: appId, + approval, + by: input.by, + as: "owner", + }); }); - return approval; + return written ? approval : null; } /** @@ -738,33 +777,50 @@ export function createSharedUseStore( by: string; decision: "approve" | "decline"; }) { - const [claimed] = await database - .update(sharedUseRequests) - .set({ - status: input.decision === "approve" ? "approved" : "declined", - decidedBy: input.by, - decidedAt: new Date(), - }) - .where( - and( - eq(sharedUseRequests.id, input.id), - eq(sharedUseRequests.status, "pending"), - ), - ) - .returning(); - if (!claimed) throw new SharedUseRequestDecidedError(); - if (input.decision === "approve") { - await setApproval({ - botId: claimed.agentId, - serverId: claimed.serverId, - by: input.by, - approval: { - audience: claimed.proposedAudience, - outsideInput: claimed.proposedOutsideInput, - members: claimed.proposedMembers, - }, - }); - } + await database.transaction(async (tx) => { + const [request] = await tx + .select({ + botId: sharedUseRequests.agentId, + serverId: sharedUseRequests.serverId, + }) + .from(sharedUseRequests) + .where( + and( + eq(sharedUseRequests.id, input.id), + eq(sharedUseRequests.status, "pending"), + ), + ) + .limit(1); + if (!request) throw new SharedUseRequestDecidedError(); + const serverId = await lockApproval(tx, request.botId, request.serverId); + const [claimed] = await tx + .update(sharedUseRequests) + .set({ + status: input.decision === "approve" ? "approved" : "declined", + decidedBy: input.by, + decidedAt: new Date(), + }) + .where( + and( + eq(sharedUseRequests.id, input.id), + eq(sharedUseRequests.status, "pending"), + ), + ) + .returning(); + if (!claimed) throw new SharedUseRequestDecidedError(); + if (input.decision === "approve") { + await writeApproval(tx, { + botId: claimed.agentId, + serverId, + by: input.by, + approval: { + audience: claimed.proposedAudience, + outsideInput: claimed.proposedOutsideInput, + members: claimed.proposedMembers, + }, + }); + } + }); const [row] = await rows(eq(sharedUseRequests.id, input.id)); /* * THE CLAIM ABOVE PROVES THE REQUEST ROW IS THERE, and this read still has to be checked: `rows` diff --git a/server/tests/shared-use-store.integration.test.ts b/server/tests/shared-use-store.integration.test.ts index 8c226408b..291558b68 100644 --- a/server/tests/shared-use-store.integration.test.ts +++ b/server/tests/shared-use-store.integration.test.ts @@ -1,7 +1,7 @@ // server/tests/shared-use-store.integration.test.ts import { afterAll, beforeEach, describe, expect, test } from "bun:test"; import { randomUUID } from "node:crypto"; -import { eq, inArray } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { createAuditStore } from "../src/audit"; import { createDatabase } from "../src/db/client"; import { @@ -273,6 +273,243 @@ describe("the shared-use gate", () => { }); }); +describe("decision durability", () => { + test("a direct approval and an administrator decision complete without a lock-order deadlock", async () => { + await call("stranger"); + const [pending] = await store.pendingFor(bot); + const blocker = createDatabase(testDatabaseUrl(), { max: 1 }); + const deciding = createDatabase(testDatabaseUrl(), { max: 1 }); + const setting = createDatabase(testDatabaseUrl(), { max: 1 }); + const decidingName = `approval_deciding_${suite}`; + const settingName = `approval_setting_${suite}`; + await deciding.execute( + sql`select set_config('application_name', ${decidingName}, false)`, + ); + await setting.execute( + sql`select set_config('application_name', ${settingName}, false)`, + ); + const locked = Promise.withResolvers(); + const release = Promise.withResolvers(); + const gate = blocker.transaction(async (tx) => { + await tx + .select() + .from(sharedUseRequests) + .where(eq(sharedUseRequests.id, pending!.id)) + .for("update"); + locked.resolve(); + await release.promise; + }); + void gate.catch(locked.reject); + const operations: Promise[] = []; + async function waitForDatabaseLock(name: string) { + const deadline = Date.now() + 5000; + while (Date.now() < deadline) { + const rows = await database.execute( + sql`select wait_event_type from pg_stat_activity where application_name = ${name} and datname = current_database()`, + ); + if (rows.some((row) => row.wait_event_type === "Lock")) return; + } + throw new Error( + `The database did not report the expected lock wait for ${name}`, + ); + } + try { + await locked.promise; + operations.push( + createSharedUseStore(deciding, "composio").decide({ + id: pending!.id, + by: id("admin"), + decision: "approve", + }), + ); + await waitForDatabaseLock(decidingName); + operations.push( + createSharedUseStore(setting, "composio").setApproval({ + botId: bot, + serverId: app, + by: id("owner"), + approval: { audience: "team", outsideInput: true, members: [] }, + }), + ); + await waitForDatabaseLock(settingName); + release.resolve(); + await gate; + const outcomes = await Promise.allSettled(operations); + expect(outcomes.map((outcome) => outcome.status)).toEqual([ + "fulfilled", + "fulfilled", + ]); + expect((await store.approvalFor(bot, app))?.audience).toBe("team"); + expect(await store.pendingFor(bot)).toHaveLength(0); + } finally { + release.resolve(); + await Promise.allSettled([gate, ...operations]); + await Promise.all([ + blocker.$client.end(), + deciding.$client.end(), + setting.$client.end(), + ]); + } + }, 15000); + + test.each(["shared_use_approvals", "shared_use_approval_members"] as const)( + "a failed %s write leaves an approval request pending and retryable", + async (table) => { + await database + .insert(teamBotPublications) + .values({ agentId: bot, publishedBy: id("owner"), audience: "people" }); + await database + .insert(teamBotAudience) + .values({ agentId: bot, kind: "group", value: "platform" }); + await call("newhire"); + const [pending] = await store.pendingFor(bot); + const fault = `approval_fault_${suite}`; + await database.execute( + sql.raw( + `CREATE FUNCTION ${fault}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'synthetic approval write fault'; END $$`, + ), + ); + await database.execute( + sql.raw( + `CREATE TRIGGER ${fault} BEFORE INSERT ON ${table} FOR EACH ROW EXECUTE FUNCTION ${fault}()`, + ), + ); + try { + await expect( + store.decide({ + id: pending!.id, + by: id("admin"), + decision: "approve", + }), + ).rejects.toThrow(); + const [row] = await database + .select() + .from(sharedUseRequests) + .where(eq(sharedUseRequests.id, pending!.id)); + expect(row?.status).toBe("pending"); + expect(row?.decidedAt).toBeNull(); + expect(await store.approvalFor(bot, app)).toBeNull(); + } finally { + await database.execute(sql.raw(`DROP TRIGGER ${fault} ON ${table}`)); + await database.execute(sql.raw(`DROP FUNCTION ${fault}()`)); + } + await store.decide({ + id: pending!.id, + by: id("admin"), + decision: "approve", + }); + expect(await store.pendingFor(bot)).toHaveLength(0); + expect((await store.approvalFor(bot, app))?.members).toEqual([ + { kind: "group", value: "platform" }, + ]); + expect((await call("newhire")).allowed).toBe(true); + expect((await call("stranger")).allowed).toBe(false); + }, + ); +}); + +describe("owner approvals", () => { + test.each(["owner", "people"] as const)( + "a concurrent owner grant cannot replace an administrator's %s approval", + async (audience) => { + const approval = { + audience, + outsideInput: false, + members: + audience === "people" + ? [{ kind: "user" as const, value: id("named") }] + : [], + }; + await database + .update(mcpServers) + .set({ ownerGrants: "any" }) + .where(eq(mcpServers.id, app)); + await database + .update(agentProfiles) + .set({ visibility: "public" }) + .where(eq(agentProfiles.agentId, bot)); + const request = await store.fileRequest({ + botId: bot, + serverId: app, + reason: "grant", + requestedBy: id("owner"), + proposed: approval, + }); + const blocker = createDatabase(testDatabaseUrl(), { max: 1 }); + const admin = createDatabase(testDatabaseUrl(), { max: 1 }); + const owner = createDatabase(testDatabaseUrl(), { max: 1 }); + const adminName = `owner_grant_admin_${suite}`; + const ownerName = `owner_grant_owner_${suite}`; + await admin.execute( + sql`select set_config('application_name', ${adminName}, false)`, + ); + await owner.execute( + sql`select set_config('application_name', ${ownerName}, false)`, + ); + const locked = Promise.withResolvers(); + const release = Promise.withResolvers(); + const gate = blocker.transaction(async (tx) => { + await tx + .select() + .from(sharedUseRequests) + .where(eq(sharedUseRequests.id, request.id)) + .for("update"); + locked.resolve(); + await release.promise; + }); + void gate.catch(locked.reject); + const adminStore = createSharedUseStore(admin, "composio"); + const ownerStore = createSharedUseStore(owner, "composio"); + const operations: Promise[] = []; + async function waitForDatabaseLock(name: string) { + const deadline = Date.now() + 5000; + while (Date.now() < deadline) { + const rows = await database.execute( + sql`select wait_event_type from pg_stat_activity where application_name = ${name} and datname = current_database()`, + ); + if (rows.some((row) => row.wait_event_type === "Lock")) return; + } + throw new Error( + `The database did not report the expected lock wait for ${name}`, + ); + } + try { + await locked.promise; + const adminWork = adminStore.setApproval({ + botId: bot, + serverId: app, + by: id("admin"), + approval, + }); + operations.push(adminWork); + await waitForDatabaseLock(adminName); + const ownerWork = ownerStore.approveOwnerGrant({ + botId: bot, + serverId: app, + by: id("owner"), + }); + operations.push(ownerWork); + await waitForDatabaseLock(ownerName); + release.resolve(); + await gate; + await adminWork; + expect(await ownerWork).toBeNull(); + expect(await store.approvalSource(bot, app)).toBe("admin"); + expect(await store.approvalFor(bot, app)).toEqual(approval); + } finally { + release.resolve(); + await Promise.allSettled([gate, ...operations]); + await Promise.all([ + blocker.$client.end(), + admin.$client.end(), + owner.$client.end(), + ]); + } + }, + 15000, + ); +}); + describe("decide", () => { test("two admins deciding one request record exactly one decision", async () => { await call("stranger");