From 1a4a18712dd958b6966d0ffaa5986e46719ef1da Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 6 Jul 2026 21:08:04 -0700 Subject: [PATCH 01/13] Reject duplicate connection names on create instead of overwriting connections.create silently upserted when a connection with the same (owner, integration, name) already existed, clobbering the stored credential and metadata. Fail with ConnectionAlreadyExistsError (409) instead, checked before any provider write and re-checked inside the insert transaction. OAuth minting keeps its intentional re-mint upsert. --- .changeset/connection-create-conflict.md | 5 ++ e2e/scenarios/no-auth-connection.test.ts | 26 +++++++ packages/core/api/src/connections/api.ts | 5 ++ packages/core/sdk/src/connections.test.ts | 92 +++++++++++++++++++++++ packages/core/sdk/src/core-tools.ts | 25 +++++- packages/core/sdk/src/errors.ts | 20 +++++ packages/core/sdk/src/executor.ts | 84 +++++++++++---------- packages/core/sdk/src/index.ts | 1 + packages/core/sdk/src/plugin.ts | 2 + packages/core/sdk/src/shared.ts | 1 + packages/plugins/mcp/src/sdk/plugin.ts | 4 + 11 files changed, 223 insertions(+), 42 deletions(-) create mode 100644 .changeset/connection-create-conflict.md diff --git a/.changeset/connection-create-conflict.md b/.changeset/connection-create-conflict.md new file mode 100644 index 0000000000..d0a74659e0 --- /dev/null +++ b/.changeset/connection-create-conflict.md @@ -0,0 +1,5 @@ +--- +"@executor-js/sdk": patch +--- + +`connections.create` now fails with `ConnectionAlreadyExistsError` (HTTP 409) when a connection with the same owner, integration, and name already exists, instead of silently overwriting the existing connection and its stored credential. Remove the existing connection first or pick a different name. OAuth reconnects are unaffected: they intentionally re-mint the same connection. diff --git a/e2e/scenarios/no-auth-connection.test.ts b/e2e/scenarios/no-auth-connection.test.ts index 7b47555bfe..2135b88705 100644 --- a/e2e/scenarios/no-auth-connection.test.ts +++ b/e2e/scenarios/no-auth-connection.test.ts @@ -93,6 +93,19 @@ const created = await tools.executor.coreTools.connections.create({ return created.ok ? { ok: true, connection: created.data } : { ok: false, error: created.error }; `; +// Create is never a replace: re-creating the SAME (owner, integration, name) +// must fail with ConnectionAlreadyExistsError instead of silently upserting +// over the existing connection. +const createDuplicateConnectionCode = (slug: string) => ` +const created = await tools.executor.coreTools.connections.create({ + owner: "org", + name: "public", + integration: ${JSON.stringify(slug)}, + template: "none", +}); +return created.ok ? { ok: true, connection: created.data } : { ok: false, error: created.error }; +`; + // The relaxed filter must still reject an origin on a no-auth create — an // empty `inputs: {}` is a (degenerate) origin and a credential the connection // can't hold, so it stays a validation failure. @@ -189,6 +202,19 @@ scenario( rejected.ok, `a no-auth create with an empty inputs origin is rejected: ${JSON.stringify(rejected)}`, ).toBe(false); + + // 5. Create is never a replace: re-creating the same (owner, + // integration, name) fails with a conflict instead of silently + // editing over the existing connection. + const duplicate = yield* executeJson(session, createDuplicateConnectionCode(integration)); + expect( + duplicate.ok, + `re-creating an existing connection name is rejected: ${JSON.stringify(duplicate)}`, + ).toBe(false); + expect( + JSON.stringify(duplicate.error), + "the failure names the conflict so callers can act on it", + ).toContain("already exists"); }).pipe( // Selfhost shares one workspace identity — leaked connections fail other // scenarios' zero-state assertions, so drop everything this run made. diff --git a/packages/core/api/src/connections/api.ts b/packages/core/api/src/connections/api.ts index 5ba4878729..65aefbc6a9 100644 --- a/packages/core/api/src/connections/api.ts +++ b/packages/core/api/src/connections/api.ts @@ -13,6 +13,7 @@ import { Predicate, Schema } from "effect"; import { AuthTemplateSlug, ConnectionAddress, + ConnectionAlreadyExistsError, ConnectionName, ConnectionNotFoundError, CredentialProviderNotRegisteredError, @@ -166,6 +167,9 @@ const ConnectionNotFound = ConnectionNotFoundError.annotate({ const IntegrationNotFound = IntegrationNotFoundError.annotate({ httpApiStatus: 404, }); +const ConnectionAlreadyExists = ConnectionAlreadyExistsError.annotate({ + httpApiStatus: 409, +}); const CredentialProviderNotRegistered = CredentialProviderNotRegisteredError.annotate({ httpApiStatus: 409, }); @@ -192,6 +196,7 @@ export const ConnectionsApi = HttpApiGroup.make("connections") error: [ InternalError, IntegrationNotFound, + ConnectionAlreadyExists, CredentialProviderNotRegistered, InvalidConnectionInput, ], diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index a183ac501d..56fc35883c 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -124,6 +124,98 @@ describe("connections.create", () => { }), ); + // Create is never a replace: a second create with the same (owner, + // integration, name) must fail with ConnectionAlreadyExistsError and leave + // the first connection fully intact — including its stored secret, which a + // silent upsert would overwrite (the pasted value's item id is derived from + // the name, so the provider write alone clobbers it). + it.effect("rejects a duplicate (owner, integration, name) and keeps the original intact", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "original-token", + description: "original", + }); + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "clobbered-token", + description: "clobbered", + }), + ); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ConnectionAlreadyExistsError")(result.failure)).toBe(true); + + // The original row and its secret both survived. + const connections = yield* executor.connections.list(); + expect(connections.length).toBe(1); + expect(connections[0]?.description).toBe("original"); + const value = yield* executor.demo.resolveValue("org", "main"); + expect(value).toBe("original-token"); + }), + ); + + // Names collide AFTER identifier normalization: "my-api-key" and "my api key" + // both normalize to myApiKey, so the second must be rejected even though the + // raw inputs differ. + it.effect("rejects a duplicate that only collides after name normalization", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("my-api-key"), + integration: INTEG, + template: TEMPLATE, + value: "v1", + }); + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("my api key"), + integration: INTEG, + template: TEMPLATE, + value: "v2", + }), + ); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("ConnectionAlreadyExistsError")(result.failure)).toBe(true); + const value = yield* executor.demo.resolveValue("org", "myApiKey"); + expect(value).toBe("v1"); + }), + ); + + it.effect("allows the same name under a different owner", () => + Effect.gen(function* () { + const executor = yield* setup(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "org-token", + }); + const personal = yield* executor.connections.create({ + owner: "user", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "user-token", + }); + expect(String(personal.address)).toBe("tools.vercel.user.main"); + expect((yield* executor.connections.list()).length).toBe(2); + }), + ); + it.effect("external `from` references a provider item without writing it", () => Effect.gen(function* () { const executor = yield* setup(); diff --git a/packages/core/sdk/src/core-tools.ts b/packages/core/sdk/src/core-tools.ts index a61aa85a92..595e1c226b 100644 --- a/packages/core/sdk/src/core-tools.ts +++ b/packages/core/sdk/src/core-tools.ts @@ -24,6 +24,7 @@ import { import { definePlugin, tool, type StaticToolSchema } from "./plugin"; import { ToolPolicyActionSchema } from "./policies"; import type { Tool } from "./tool"; +import { ToolResult } from "./tool-result"; const schemaToStandard = (schema: Schema.Decoder): StaticToolSchema => Schema.toStandardSchemaV1(Schema.toStandardJSONSchemaV1(schema) as never) as StaticToolSchema< @@ -578,7 +579,7 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { tool({ name: "connections.create", description: - 'Low-level create or replace for a saved connection from provider item references. For a no-auth integration (public MCP server, public REST API), pass `template: "none"` with no `from`/`inputs` to wire it up directly. For normal API keys/tokens, use `connections.createHandoff` so the user enters the credential in the web UI. OAuth credentials should use `oauth.start`.', + 'Low-level create for a saved connection from provider item references. Fails if a connection with the same owner, integration, and name already exists (remove it first, or pick a different name). For a no-auth integration (public MCP server, public REST API), pass `template: "none"` with no `from`/`inputs` to wire it up directly. For normal API keys/tokens, use `connections.createHandoff` so the user enters the credential in the web UI. OAuth credentials should use `oauth.start`.', inputSchema: ConnectionCreateInputStd, outputSchema: ConnectionOutputStd, // Creating a connection binds a credential reference and roots a new @@ -589,9 +590,25 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { // approval-gated (the v1 `sources.configure` carried the same guard). annotations: { requiresApproval: true }, execute: (input: typeof ConnectionCreateInput.Type, { ctx }) => - Effect.map( - ctx.connections.create(createConnectionInputFromTool(input)), - connectionToOutput, + ctx.connections.create(createConnectionInputFromTool(input)).pipe( + Effect.map(connectionToOutput), + // Expected, caller-actionable failures resolve as ToolResult.fail + // (the sandbox sees `{ ok: false, error }`); anything else stays + // a defect and surfaces as the opaque internal-error generic. + Effect.catchTags({ + ConnectionAlreadyExistsError: (error) => + Effect.succeed( + ToolResult.fail({ code: "connection_already_exists", message: error.message }), + ), + IntegrationNotFoundError: (error) => + Effect.succeed( + ToolResult.fail({ code: "integration_not_found", message: error.message }), + ), + InvalidConnectionInputError: (error) => + Effect.succeed( + ToolResult.fail({ code: "invalid_connection_input", message: error.message }), + ), + }), ), }), tool({ diff --git a/packages/core/sdk/src/errors.ts b/packages/core/sdk/src/errors.ts index b0a6c95033..1177d8f382 100644 --- a/packages/core/sdk/src/errors.ts +++ b/packages/core/sdk/src/errors.ts @@ -130,6 +130,26 @@ export class ConnectionNotFoundError extends Schema.TaggedErrorClass()( + "ConnectionAlreadyExistsError", + { + owner: Owner, + integration: IntegrationSlug, + name: ConnectionName, + }, + { httpApiStatus: 409 }, +) { + override get message(): string { + return `A connection named "${this.name}" already exists for ${this.integration} (${this.owner}). Choose a different name, or remove the existing connection first.`; + } +} + /** A connection create request was rejected before anything was written: the * input is structurally invalid (no credential inputs for a credentialed * template, mixed pasted/external origins, …) or targets owner `user` in a diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 071d4a21b6..9640d6c0b9 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -53,6 +53,7 @@ import { export type { OnElicitation, InvokeOptions } from "./elicitation"; import { + ConnectionAlreadyExistsError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, CredentialResolutionError, @@ -297,6 +298,7 @@ export type Executor = { ) => Effect.Effect< Connection, | IntegrationNotFoundError + | ConnectionAlreadyExistsError | CredentialProviderNotRegisteredError | InvalidConnectionInputError | StorageFailure @@ -2167,6 +2169,7 @@ export const createExecutor = = { + if (existing) { + return yield* new ConnectionAlreadyExistsError({ + owner: input.owner, + integration: input.integration, + name, + }); + } + yield* core.create("connection", { + tenant: keys.tenant, + owner: keys.owner, + subject: keys.subject, + integration: String(input.integration), + name: String(name), template: String(input.template), provider: providerKey, item_ids: itemIds, identity_label: input.identityLabel ?? null, - // Re-saving a credential keeps an existing curated description - // unless the caller explicitly provides one. - ...(input.description !== undefined ? { description: input.description } : {}), + description: input.description ?? null, + oauth_client: null, + refresh_item_id: null, + expires_at: null, + oauth_scope: null, + provider_state: null, + created_at: now, updated_at: now, - }; - if (existing) { - yield* core.updateMany("connection", { - where: (b: AnyCb) => - b.and( - byOwner(input.owner)(b), - b("integration", "=", String(input.integration)), - b("name", "=", String(name)), - ), - set, - }); - } else { - yield* core.create("connection", { - tenant: keys.tenant, - owner: keys.owner, - subject: keys.subject, - integration: String(input.integration), - name: String(name), - template: String(input.template), - provider: providerKey, - item_ids: itemIds, - identity_label: input.identityLabel ?? null, - description: input.description ?? null, - oauth_client: null, - refresh_item_id: null, - expires_at: null, - oauth_scope: null, - provider_state: null, - created_at: now, - updated_at: now, - }); - } + }); }), ); @@ -2348,8 +2354,10 @@ export const createExecutor = => diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index fc11fb9bf3..474f9e2323 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -68,6 +68,7 @@ export { IntegrationNotFoundError, IntegrationAlreadyExistsError, IntegrationRemovalNotAllowedError, + ConnectionAlreadyExistsError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, CredentialResolutionError, diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 31dfa07eb6..bd2ef3ef48 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -38,6 +38,7 @@ import type { ElicitationResponse, } from "./elicitation"; import type { + ConnectionAlreadyExistsError, ConnectionNotFoundError, CredentialProviderNotRegisteredError, IntegrationNotFoundError, @@ -204,6 +205,7 @@ export interface PluginCtx { ) => Effect.Effect< Connection, | IntegrationNotFoundError + | ConnectionAlreadyExistsError | CredentialProviderNotRegisteredError | InvalidConnectionInputError | StorageFailure diff --git a/packages/core/sdk/src/shared.ts b/packages/core/sdk/src/shared.ts index 931a332e1c..63a94b5a77 100644 --- a/packages/core/sdk/src/shared.ts +++ b/packages/core/sdk/src/shared.ts @@ -58,6 +58,7 @@ export { IntegrationNotFoundError, IntegrationAlreadyExistsError, IntegrationRemovalNotAllowedError, + ConnectionAlreadyExistsError, ConnectionNotFoundError, InvalidConnectionInputError, CredentialProviderNotRegisteredError, diff --git a/packages/plugins/mcp/src/sdk/plugin.ts b/packages/plugins/mcp/src/sdk/plugin.ts index cc4194af6c..e50ca1394c 100644 --- a/packages/plugins/mcp/src/sdk/plugin.ts +++ b/packages/plugins/mcp/src/sdk/plugin.ts @@ -917,6 +917,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { Effect.fail( new McpConnectionError({ transport: "stdio", message: cause.message }), ), + ConnectionAlreadyExistsError: (cause) => + Effect.fail( + new McpConnectionError({ transport: "stdio", message: cause.message }), + ), CredentialProviderNotRegisteredError: (cause) => Effect.fail( new McpConnectionError({ transport: "stdio", message: cause.message }), From ac6a29d54c605d04f581635b580ebf7cbecbdd01 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 6 Jul 2026 21:17:14 -0700 Subject: [PATCH 02/13] Update e2e scenarios that leaned on the connection-create upsert health-checks swapped in a dead key by re-creating the connection; the fixture server now revokes the token instead. The cloud credentials scenario pinned replace-in-place; it now pins the conflict rejection. --- e2e/cloud/connections-credentials.test.ts | 38 ++++++++++-------- e2e/scenarios/health-checks.test.ts | 47 ++++++++++------------- 2 files changed, 43 insertions(+), 42 deletions(-) diff --git a/e2e/cloud/connections-credentials.test.ts b/e2e/cloud/connections-credentials.test.ts index d94ce374a8..2386c7db79 100644 --- a/e2e/cloud/connections-credentials.test.ts +++ b/e2e/cloud/connections-credentials.test.ts @@ -3,8 +3,9 @@ // identified by (owner, integration, name), with its value stored through the // real vault. The product promises under test: the secret goes in but NEVER // comes back out of any endpoint; metadata round-trips; re-creating the same -// connection replaces it instead of duplicating; removal really removes; and -// unknown connections fail with a typed not-found error. +// connection is rejected as a conflict instead of silently replacing it; +// removal really removes; and unknown connections fail with a typed +// not-found error. import { randomBytes } from "node:crypto"; import { expect } from "@effect/vitest"; @@ -102,7 +103,7 @@ scenario( ); scenario( - "Connections · re-creating the same connection replaces it instead of duplicating", + "Connections · re-creating the same connection is rejected and leaves the original intact", {}, Effect.gen(function* () { const target = yield* Target; @@ -128,21 +129,28 @@ scenario( "the first create stores one row with its label", ).toEqual(["first key"]); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration, - template: TEMPLATE_API_KEY, - identityLabel: "rotated key", - value: "second-value", - }, - }); + const error = yield* client.connections + .create({ + payload: { + owner: "org", + name, + integration, + template: TEMPLATE_API_KEY, + identityLabel: "clobber attempt", + value: "second-value", + }, + }) + .pipe(Effect.flip); + expect( + (error as { _tag?: string })._tag, + "re-creating the same (owner, integration, name) fails with the typed conflict", + ).toBe("ConnectionAlreadyExistsError"); + const second = yield* client.connections.list({ query: { integration } }); expect( second.filter((connection) => connection.name === name).map((c) => c.identityLabel), - "re-creating the same (owner, integration, name) updates the row in place", - ).toEqual(["rotated key"]); + "the rejected create left the original row untouched", + ).toEqual(["first key"]); }), ); diff --git a/e2e/scenarios/health-checks.test.ts b/e2e/scenarios/health-checks.test.ts index c4db028d4c..cb8adf53c0 100644 --- a/e2e/scenarios/health-checks.test.ts +++ b/e2e/scenarios/health-checks.test.ts @@ -141,14 +141,20 @@ const discriminatedUnionSpec = (baseUrl: string): string => { }; /** A real node:http identity API on 127.0.0.1. `GET /me` returns the account - * JSON only when the bearer token matches `validToken`; any other token is a - * 401 (the "the dev token got revoked" case the health check classifies as - * expired). Closed by the scope's finalizer. */ + * JSON only when the bearer token matches the CURRENT `validToken`; any other + * token is a 401 (the "the dev token got revoked" case the health check + * classifies as expired). `revoke()` rotates the server-side token so a saved + * key stops working mid-scenario. Closed by the scope's finalizer. */ const serveIdentityApi = (validToken: string) => Effect.acquireRelease( - Effect.callback<{ readonly url: string; readonly close: () => void }>((resume) => { + Effect.callback<{ + readonly url: string; + readonly revoke: () => void; + readonly close: () => void; + }>((resume) => { + let currentToken = validToken; const server = createServer((request, response) => { - const authorized = request.headers["authorization"] === `Bearer ${validToken}`; + const authorized = request.headers["authorization"] === `Bearer ${currentToken}`; if (request.method === "GET" && (request.url ?? "").startsWith("/me")) { if (!authorized) { response.writeHead(401, { "content-type": "application/json" }); @@ -168,6 +174,9 @@ const serveIdentityApi = (validToken: string) => resume( Effect.succeed({ url: `http://127.0.0.1:${port}`, + revoke: () => { + currentToken = `revoked_${validToken}`; + }, close: () => { server.close(); server.closeAllConnections(); @@ -323,17 +332,9 @@ scenario( expect(healthy.httpStatus, "the saved probe saw the 200").toBe(200); expect(healthy.identity, "the saved probe derives the account identity").toBe(IDENTITY); - // Re-creating the same (owner, integration, name) replaces the stored - // key in place: now the connection holds a key the server rejects. - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration: slug, - template: TEMPLATE, - value: "rotated-away", - }, - }); + // The server revokes the key: the connection's saved value now gets + // a 401 (the "dev token got revoked" case). + server.revoke(); const expired = yield* client.connections.checkHealth({ params: { owner: "org", integration: slug, name }, query: {}, @@ -642,22 +643,14 @@ scenario( }, }); - // Seed a verdict, then re-create the connection with a DEAD key: the - // persisted verdict (healthy) and reality (expired) now disagree. + // Seed a verdict, then revoke the key server-side: the persisted + // verdict (healthy) and reality (expired) now disagree. const seeded = yield* client.connections.checkHealth({ params: { owner: "org", integration: slug, name }, query: {}, }); expect(seeded.status, "the seed probe is healthy").toBe("healthy"); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration: slug, - template: TEMPLATE, - value: "rotated-away", - }, - }); + server.revoke(); // Within the freshness window the CACHED verdict comes back with no // probe, so the dead key still reads healthy (stale by design). From 34b1790e45a10791b010eb4e48bbcab80ff56718 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 7 Jul 2026 12:53:49 -0700 Subject: [PATCH 03/13] Show a specific toast when the connection name is taken The conflict error message is a getter, so it does not survive the wire; match the tag client-side and rebuild the message, mirroring the integration already-exists handling. Adds a browser scenario recording the rejected duplicate. --- .../connection-duplicate-name-demo.test.ts | 89 +++++++++++++++++++ .../src/components/add-account-modal.tsx | 26 +++++- 2 files changed, 114 insertions(+), 1 deletion(-) create mode 100644 e2e/selfhost/connection-duplicate-name-demo.test.ts diff --git a/e2e/selfhost/connection-duplicate-name-demo.test.ts b/e2e/selfhost/connection-duplicate-name-demo.test.ts new file mode 100644 index 0000000000..e02c432ab5 --- /dev/null +++ b/e2e/selfhost/connection-duplicate-name-demo.test.ts @@ -0,0 +1,89 @@ +// Demo recording (AFTER the fix): adding a second connection without typing a +// name derives the same default name as the first — the create is REJECTED +// with the conflict error and the original connection survives untouched. +// Video is the artifact. +import { randomBytes } from "node:crypto"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { IntegrationSlug } from "@executor-js/sdk/shared"; + +import { scenario } from "../src/scenario"; +import { Api, Browser, Target } from "../src/services"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const bearerSpec = (): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Bearer Fixture", version: "1.0.0" }, + servers: [{ url: "https://api.bearerfix.test" }], + security: [{ bearerAuth: [] }], + components: { securitySchemes: { bearerAuth: { type: "http", scheme: "bearer" } } }, + paths: { + "/ping": { get: { operationId: "ping", responses: { "200": { description: "ok" } } } }, + }, + }); + +scenario( + "Connections · a second connection with the default name is rejected, not overwritten", + {}, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeApiClient } = yield* Api; + const browser = yield* Browser; + const identity = yield* target.newIdentity(); + const apiClient = yield* makeApiClient(api, identity); + const slug = `dup_name_demo_${randomBytes(4).toString("hex")}`; + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* apiClient.openapi.addSpec({ + payload: { spec: { kind: "blob", value: bearerSpec() }, slug }, + }); + + yield* browser.session(identity, async ({ page, step }) => { + const addConnection = async (key: string) => { + await page.getByRole("button", { name: "Add connection" }).first().click(); + await page.getByRole("heading", { name: /Add connection/ }).waitFor(); + const dialog = page.getByRole("dialog", { name: /Add connection/ }); + await dialog.locator('input[type="password"]').first().fill(key); + await dialog.getByRole("button", { name: "Continue" }).click(); + // Leave the display name empty: the default derives the SAME + // connection name both times. + await dialog.getByRole("button", { name: "Add connection" }).click(); + }; + + await step("Add the first connection with the default name", async () => { + await page.goto(`/integrations/${slug}`, { waitUntil: "networkidle" }); + await page.getByText("Connections").first().waitFor(); + await addConnection("first-key"); + await page.getByText("Connection added").waitFor(); + }); + + await step("Add a second connection, also leaving the name empty", async () => { + await page.waitForTimeout(1_000); // let the first toast clear + await addConnection("second-key"); + }); + + await step("The duplicate is rejected with the conflict error", async () => { + await page.getByText(/already exists/).waitFor(); + await page.waitForTimeout(2_000); // hold the error on screen + }); + }); + + const connections = yield* apiClient.connections.list({ + query: { integration: IntegrationSlug.make(slug) }, + }); + expect(connections.length, "the original connection is the only one").toBe(1); + }), + apiClient.openapi + .removeSpec({ params: { slug: IntegrationSlug.make(slug) } }) + .pipe(Effect.ignore), + ); + }), + ), +); diff --git a/packages/react/src/components/add-account-modal.tsx b/packages/react/src/components/add-account-modal.tsx index 7259b199ce..da47c82d33 100644 --- a/packages/react/src/components/add-account-modal.tsx +++ b/packages/react/src/components/add-account-modal.tsx @@ -1,6 +1,8 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useAtomSet, useAtomValue } from "@effect/atom-react"; import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"; import { ConnectionName, @@ -543,6 +545,19 @@ export const connectionLabelForHost = ( organizationId: string | null, ): string => label.trim() || `${ownerLabelForHost(owner, organizationId)} ${integrationName}`; +/** The create endpoint rejected the name as taken (409). Like + * `isIntegrationAlreadyExistsExit`: the error's `message` is a getter derived + * from its fields, so it doesn't survive the wire — match the tag and rebuild + * the message client-side. */ +export const isConnectionAlreadyExistsExit = (exit: Exit.Exit): boolean => + Option.match(Exit.findErrorOption(exit), { + onNone: () => false, + onSome: Predicate.isTagged("ConnectionAlreadyExistsError"), + }); + +export const connectionExistsMessage = (label: string): string => + `A connection named "${label}" already exists. Pick a different name, or remove the existing connection first.`; + /** The default owner a new connection is saved under when the user makes no * explicit choice. Personal: a connection is most often a personal credential. */ export const DEFAULT_CONNECTION_OWNER: Owner = "user"; @@ -1746,7 +1761,16 @@ function AddAccountModalView(props: AddAccountModalProps) { }); if (Exit.isFailure(exit)) { setSubmitting(false); - toast.error(messageFromExit(exit, "Failed to add connection")); + // The conflict error's message is a getter derived from its fields, so + // it doesn't survive the wire — rebuild it from the tag (the same + // pattern as isIntegrationAlreadyExistsExit). + toast.error( + isConnectionAlreadyExistsExit(exit) + ? connectionExistsMessage( + connectionLabelForHost(label, owner, integrationName, organizationId), + ) + : messageFromExit(exit, "Failed to add connection"), + ); return; } toast.success("Connection added"); From fdd0288a80bac6993f87cf535c74ea128f92e399 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 7 Jul 2026 13:49:21 -0700 Subject: [PATCH 04/13] Guard the OAuth flow against overwriting an existing connection The callback mint upserts by (owner, integration, name), so a fresh oauth.start aimed at an existing name silently replaced that connection when the user returned from the provider. start now rejects the name up front and complete re-checks it (a connection can appear mid-flight), both leaving the existing connection untouched. Reconnect flows pass reconnect: true and keep re-minting the same connection. --- .changeset/connection-create-conflict.md | 2 +- packages/core/api/src/handlers/oauth.ts | 1 + packages/core/api/src/oauth/api.ts | 4 + packages/core/sdk/src/core-tools.ts | 4 +- packages/core/sdk/src/executor.ts | 8 + packages/core/sdk/src/oauth-client.ts | 5 + packages/core/sdk/src/oauth-flow.test.ts | 167 ++++++++++++++++++ packages/core/sdk/src/oauth-service.ts | 64 +++++++ .../graphql/src/react/GraphqlSignInButton.tsx | 3 + .../src/components/add-account-modal.tsx | 1 + packages/react/src/plugins/oauth-reconnect.ts | 3 + packages/react/src/plugins/oauth-sign-in.tsx | 4 + 12 files changed, 264 insertions(+), 2 deletions(-) diff --git a/.changeset/connection-create-conflict.md b/.changeset/connection-create-conflict.md index d0a74659e0..f27d498f78 100644 --- a/.changeset/connection-create-conflict.md +++ b/.changeset/connection-create-conflict.md @@ -2,4 +2,4 @@ "@executor-js/sdk": patch --- -`connections.create` now fails with `ConnectionAlreadyExistsError` (HTTP 409) when a connection with the same owner, integration, and name already exists, instead of silently overwriting the existing connection and its stored credential. Remove the existing connection first or pick a different name. OAuth reconnects are unaffected: they intentionally re-mint the same connection. +Creating a connection over an existing one is now rejected instead of silently overwriting it. `connections.create` fails with `ConnectionAlreadyExistsError` (HTTP 409) when the (owner, integration, name) is already taken, and `oauth.start` / `oauth.complete` reject a fresh OAuth connect that targets an existing connection name. Reconnect flows pass `reconnect: true` and keep re-minting the same connection for re-consent and token refresh. diff --git a/packages/core/api/src/handlers/oauth.ts b/packages/core/api/src/handlers/oauth.ts index cea145f2e5..e7425d3c3d 100644 --- a/packages/core/api/src/handlers/oauth.ts +++ b/packages/core/api/src/handlers/oauth.ts @@ -161,6 +161,7 @@ export const OAuthHandlers = HttpApiBuilder.group(ExecutorApi, "oauth", (handler template: payload.template, identityLabel: payload.identityLabel, redirectUri: payload.redirectUri, + reconnect: payload.reconnect, }); return startResultToResponse(result); }), diff --git a/packages/core/api/src/oauth/api.ts b/packages/core/api/src/oauth/api.ts index 66404af5eb..d6705d039a 100644 --- a/packages/core/api/src/oauth/api.ts +++ b/packages/core/api/src/oauth/api.ts @@ -160,6 +160,10 @@ const StartPayload = Schema.Struct({ template: AuthTemplateSlug, identityLabel: Schema.optional(Schema.NullOr(Schema.String)), redirectUri: Schema.optional(Schema.NullOr(Schema.String)), + /** True when re-running OAuth for an existing connection (re-consent): the + * flow re-mints the same (owner, integration, name). A fresh connect + * targeting an existing name is rejected at start. */ + reconnect: Schema.optional(Schema.Boolean), }); const StartResponse = Schema.Union([ diff --git a/packages/core/sdk/src/core-tools.ts b/packages/core/sdk/src/core-tools.ts index 595e1c226b..7f9804be80 100644 --- a/packages/core/sdk/src/core-tools.ts +++ b/packages/core/sdk/src/core-tools.ts @@ -310,6 +310,7 @@ const OAuthStartInput = Schema.Struct({ template: Schema.String, identityLabel: Schema.optional(Schema.NullOr(Schema.String)), redirectUri: Schema.optional(Schema.NullOr(Schema.String)), + reconnect: Schema.optional(Schema.Boolean), }); const OAuthStartOutput = Schema.Union([ Schema.Struct({ @@ -832,7 +833,7 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { tool({ name: "oauth.start", description: - "Start OAuth through a registered client to mint a connection for an integration. `client_credentials` clients return `connected`; authorization-code clients return an authorization URL and state.", + "Start OAuth through a registered client to mint a connection for an integration. `client_credentials` clients return `connected`; authorization-code clients return an authorization URL and state. Fails if the connection name is already taken unless `reconnect: true` re-runs the flow for that existing connection.", inputSchema: OAuthStartInputStd, outputSchema: OAuthStartOutputStd, // This is the materialization step that turns a registered client @@ -854,6 +855,7 @@ export const coreToolsPlugin = definePlugin((options: CoreToolsPluginOptions = { template: AuthTemplateSlug.make(input.template), identityLabel: input.identityLabel, redirectUri: input.redirectUri, + reconnect: input.reconnect ?? undefined, }), (result) => result.status === "connected" diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 9640d6c0b9..3fdcb45254 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3615,6 +3615,14 @@ export const createExecutor = ownedKeys(owner), defaultWritableProvider, mintOAuthConnection: (input: MintOAuthConnectionInput) => mintOAuthConnection(input), + // Normalize the name exactly as the mint does, so the fresh-connect + // guard sees the same row the callback would overwrite. + findConnection: (ref) => + findConnectionRow({ + owner: ref.owner, + integration: ref.integration, + name: connectionIdentifier(String(ref.name)), + }).pipe(Effect.map((row) => (row ? rowToConnection(row) : null))), // One integration-row read + one projector run. Resolve the method this // template selects exactly as the runtime's `selectAuthMethod` does — // exact slug match, else the sole declared method (single-method diff --git a/packages/core/sdk/src/oauth-client.ts b/packages/core/sdk/src/oauth-client.ts index b9ff81a232..c22db3ec18 100644 --- a/packages/core/sdk/src/oauth-client.ts +++ b/packages/core/sdk/src/oauth-client.ts @@ -124,6 +124,11 @@ export interface OAuthStartInput { readonly identityLabel?: string | null; /** Browser-facing callback URL for this flow. Defaults to the executor's configured redirectUri. */ readonly redirectUri?: string | null; + /** True when this flow re-runs OAuth for an EXISTING connection (re-consent, + * widened scopes, fresh refresh token): the completed flow re-mints the same + * (owner, integration, name) row. A fresh connect (the default) targeting an + * existing name fails at start instead of silently overwriting it. */ + readonly reconnect?: boolean; } export interface OAuthCompleteInput { diff --git a/packages/core/sdk/src/oauth-flow.test.ts b/packages/core/sdk/src/oauth-flow.test.ts index 971554dc96..9581f1df33 100644 --- a/packages/core/sdk/src/oauth-flow.test.ts +++ b/packages/core/sdk/src/oauth-flow.test.ts @@ -487,6 +487,173 @@ describe("oauth.start / oauth.complete", () => { }), ), ); + + // The OAuth callback mints by upsert, so without a guard a FRESH connect + // aimed at an existing name would silently overwrite that connection when + // the user returns from the provider. `start` rejects it up front (and + // `complete` re-checks for connections created mid-flight); an explicit + // `reconnect: true` keeps the intentional re-mint for re-consent flows. + it.effect("a fresh start targeting an existing connection name is rejected", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const { executor } = yield* makeTestWorkspaceHarness({ plugins }); + yield* executor.acme.seed(); + + // An existing connection occupies the name (a static credential — + // exactly what a silent OAuth overwrite would destroy). + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: AuthTemplateSlug.make("apiKey"), + value: "precious-static-key", + }); + + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + + const startError = yield* executor.oauth + .start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }) + .pipe(Effect.flip); + expect(startError).toBeInstanceOf(OAuthStartError); + expect(startError.message).toContain("already exists"); + + // The existing connection is untouched. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.template)).toBe("apiKey"); + }), + ), + ); + + it.effect("reconnect: true re-mints the same connection through the full flow", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const { executor } = yield* makeTestWorkspaceHarness({ plugins }); + yield* executor.acme.seed(); + + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + + // First connect mints the connection. + const first = yield* executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(first.status).toBe("redirect"); + if (first.status !== "redirect") return; + const firstCallback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: first.authorizationUrl, + }); + yield* executor.oauth.complete({ state: first.state, code: firstCallback.code }); + + // Reconnect (re-consent) explicitly targets the SAME connection. + const again = yield* executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + reconnect: true, + }); + expect(again.status).toBe("redirect"); + if (again.status !== "redirect") return; + const againCallback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: again.authorizationUrl, + }); + const reminted = yield* executor.oauth.complete({ + state: again.state, + code: againCallback.code, + }); + expect(String(reminted.name)).toBe("main"); + expect((yield* executor.connections.list()).length).toBe(1); + }), + ), + ); + + // The start-time guard has a window: a connection created AFTER start but + // BEFORE the callback lands. `complete` re-checks and refuses to redeem. + it.effect("complete refuses when the name was taken mid-flight", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const { executor } = yield* makeTestWorkspaceHarness({ plugins }); + yield* executor.acme.seed(); + + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + + const started = yield* executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + + // While the user is at the provider, the name gets taken. + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: AuthTemplateSlug.make("apiKey"), + value: "created-mid-flight", + }); + + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + const completeError = yield* executor.oauth + .complete({ state: started.state, code: callback.code }) + .pipe(Effect.flip); + expect(Predicate.isTagged("OAuthCompleteError")(completeError)).toBe(true); + expect(completeError.message).toContain("already exists"); + + // The mid-flight connection survives with its static credential. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.template)).toBe("apiKey"); + }), + ), + ); }); describe("oauth token refresh in resolveConnectionValue", () => { diff --git a/packages/core/sdk/src/oauth-service.ts b/packages/core/sdk/src/oauth-service.ts index da5303416f..b51409b527 100644 --- a/packages/core/sdk/src/oauth-service.ts +++ b/packages/core/sdk/src/oauth-service.ts @@ -125,6 +125,15 @@ export interface OAuthServiceDeps { readonly mintOAuthConnection: ( input: MintOAuthConnectionInput, ) => Effect.Effect; + /** Look up a saved connection by (owner, integration, name), with the name + * normalized the way the mint normalizes it. Backs the fresh-connect guard: + * a non-reconnect `start` targeting an existing connection is rejected + * instead of silently re-minting over it at the callback. */ + readonly findConnection: (ref: { + readonly owner: Owner; + readonly integration: IntegrationSlug; + readonly name: ConnectionName; + }) => Effect.Effect; /** * Resolve the OAuth scope policy for a `(integration, template)`: * - `{ kind: "scopes", scopes }`: the scopes the integration's auth template @@ -220,6 +229,18 @@ const requestedScopesFromPayload = (payload: unknown): readonly string[] | null return Array.isArray(value) ? value.filter((s): s is string => typeof s === "string") : null; }; +/** Read the `reconnect` flag `start` recorded on the session payload. Missing + * (legacy sessions) reads as false — the safe default, since only an explicit + * Reconnect flow may overwrite an existing connection. */ +const reconnectFromPayload = (payload: unknown): boolean => { + const decoded = + typeof payload === "string" + ? decodeJsonPayload(payload).pipe(Option.getOrElse(() => payload)) + : payload; + if (decoded === null || typeof decoded !== "object") return false; + return (decoded as Record).reconnect === true; +}; + /** Read the app owner `start` recorded on the session payload. Null when absent * (same-owner connects, or sessions written before this field), so `complete` * falls back to the session owner. */ @@ -961,6 +982,25 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { }); } + // Fresh connects never replace: the callback's mint upserts by + // (owner, integration, name), so without this guard a new flow aimed at + // an existing name would silently overwrite that connection when the + // user returns from the provider. Reject at start — before any session + // or token exists — unless the caller explicitly marked the flow a + // reconnect (re-consent / widened scopes re-mint the SAME connection). + if (input.reconnect !== true) { + const existing = yield* deps.findConnection({ + owner: input.owner, + integration: input.integration, + name: input.name, + }); + if (existing) { + return yield* new OAuthStartError({ + message: `A connection named "${input.name}" already exists for ${input.integration}. Choose a different name, remove the existing connection first, or reconnect it instead.`, + }); + } + } + // Declared scopes win (driven by the selected auth template). MCP-style // integrations declare none and discover them from the client's protected // resource / authorization server metadata at connect. @@ -1077,6 +1117,10 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { owner: input.owner, clientOwner: input.clientOwner, requestedScopes: authorizationRequestedScopes, + // Recorded so `complete` can re-check the fresh-connect guard: a + // connection created between start and the callback must not be + // clobbered by the mint. + reconnect: input.reconnect === true, }, expires_at: expiresAt, created_at: now, @@ -1144,6 +1188,7 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { // owner for same-owner connects. clientOwner: clientOwnerFromPayload(sessionRow.payload) ?? (String(sessionRow.owner) as Owner), + reconnect: reconnectFromPayload(sessionRow.payload), }; // Expired sessions are not redeemable — drop + treat as not found. @@ -1152,6 +1197,25 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { return yield* new OAuthSessionNotFoundError({ state: input.state }); } + // Re-check the fresh-connect guard from `start`: the user may have + // created a connection under this name while the browser hop was in + // flight, and the mint below upserts. Checked BEFORE the code exchange + // so nothing is redeemed or stored for a flow that must not land. + if (!session.reconnect) { + const existing = yield* deps.findConnection({ + owner: session.owner, + integration: session.integration, + name: session.name, + }); + if (existing) { + yield* deleteSession(input.state); + return yield* new OAuthCompleteError({ + message: `A connection named "${session.name}" already exists for ${session.integration}. Choose a different name, remove the existing connection first, or reconnect it instead.`, + restartRequired: true, + }); + } + } + // Reload the SAME app `start` resolved, by its explicit recorded owner. const client = yield* loadClient(session.clientOwner, session.clientSlug); if (!client) { diff --git a/packages/plugins/graphql/src/react/GraphqlSignInButton.tsx b/packages/plugins/graphql/src/react/GraphqlSignInButton.tsx index cd2656c286..860bab09ac 100644 --- a/packages/plugins/graphql/src/react/GraphqlSignInButton.tsx +++ b/packages/plugins/graphql/src/react/GraphqlSignInButton.tsx @@ -59,6 +59,9 @@ export default function GraphqlSignInButton(props: { integration: IntegrationSlug.make(String(props.slug)), template: AuthTemplateSlug.make(String(props.template)), identityLabel: `${props.displayName} OAuth`, + // The button doubles as Reconnect (deterministic per-owner name): + // a second sign-in intentionally re-mints the same connection. + ...(isConnected ? { reconnect: true } : {}), }, onSuccess: (payload: OAuthCompletionPayload) => { // Touch the minted connection name to satisfy the success contract; the diff --git a/packages/react/src/components/add-account-modal.tsx b/packages/react/src/components/add-account-modal.tsx index da47c82d33..71da567bc8 100644 --- a/packages/react/src/components/add-account-modal.tsx +++ b/packages/react/src/components/add-account-modal.tsx @@ -1651,6 +1651,7 @@ function AddAccountModalView(props: AddAccountModalProps) { integration, template: oauthMethod.template, ...(handoff.identityLabel !== undefined ? { identityLabel: handoff.identityLabel } : {}), + reconnect: true, }, onAuthorizationStarted: () => { trackEvent("connection_reconnected", { diff --git a/packages/react/src/plugins/oauth-reconnect.ts b/packages/react/src/plugins/oauth-reconnect.ts index 388545b7c1..3d5d516e7e 100644 --- a/packages/react/src/plugins/oauth-reconnect.ts +++ b/packages/react/src/plugins/oauth-reconnect.ts @@ -40,6 +40,9 @@ export function oauthReconnectPayload(connection: Connection): OAuthStartPayload integration: connection.integration, template: connection.template, identityLabel: connection.identityLabel ?? undefined, + // Re-minting the SAME connection is the point of Reconnect; without this + // flag oauth.start rejects the existing name as a fresh-connect conflict. + reconnect: true, }; } diff --git a/packages/react/src/plugins/oauth-sign-in.tsx b/packages/react/src/plugins/oauth-sign-in.tsx index e270c1cdb8..844167be42 100644 --- a/packages/react/src/plugins/oauth-sign-in.tsx +++ b/packages/react/src/plugins/oauth-sign-in.tsx @@ -70,6 +70,10 @@ export type OAuthStartPayload = { readonly template: AuthTemplateSlug; readonly identityLabel?: string; readonly redirectUri?: string; + /** True when re-running OAuth for an EXISTING connection (Reconnect): the + * flow re-mints the same (owner, integration, name). Fresh connects omit it + * and are rejected when the name is already taken. */ + readonly reconnect?: boolean; }; export type StartOAuthPopupInput = { From 07312bdeace73609c59eda8d0fe1f9252c1b2aac Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:29:23 -0700 Subject: [PATCH 05/13] Surface OAuth name conflicts in the UI instead of a silent popup blip The popup flow hook dropped the reconnect flag from its start payload, so every Reconnect that went through it hit the new oauth.start guard: the popup opened and closed with no message. Forward the flag, toast the reconnect handoff failure, and pre-check the connection list before opening a popup for fresh BYO/CIMD/DCR connects so a taken name surfaces as an error without the window blip. --- .../mcp-oauth-reconnect-health.test.ts | 16 +++++++- .../src/components/add-account-modal.tsx | 38 ++++++++++++++++++- packages/react/src/plugins/oauth-sign-in.tsx | 1 + 3 files changed, 52 insertions(+), 3 deletions(-) diff --git a/e2e/selfhost/mcp-oauth-reconnect-health.test.ts b/e2e/selfhost/mcp-oauth-reconnect-health.test.ts index b363c38dd9..28cce20289 100644 --- a/e2e/selfhost/mcp-oauth-reconnect-health.test.ts +++ b/e2e/selfhost/mcp-oauth-reconnect-health.test.ts @@ -257,20 +257,34 @@ scenario( const oauthRequest = page .waitForRequest((request) => oauthReconnectRequest(request.url()), { timeout: 30_000 }) .then((request) => request.url()); + // Reconnect re-mints the SAME connection name, so the start call must + // carry `reconnect: true` and succeed. A 4xx here is the regression + // where the flag was dropped and the popup blipped open and closed. + const startResponse = page.waitForResponse( + (response) => + response.url().includes("/api/oauth/start") && response.request().method() === "POST", + { timeout: 30_000 }, + ); await menuTrigger.click(); await page.getByRole("menuitem", { name: "Reconnect" }).click(); await dialog.waitFor({ state: "visible", timeout: 30_000 }); const reachedOAuth = await oauthRequest; + const started = await startResponse; + const startedBody = await started.text(); await page.waitForTimeout(2_000); await dialog.waitFor({ state: "visible", timeout: 1_000 }); console.info( `[MCP OAuth repro] reconnect dialog stayed open; OAuth requests: ${ oauthRequests.join(", ") || reachedOAuth - }`, + }; start: ${started.status()} ${startedBody}`, ); expect(reachedOAuth, "Reconnect should issue an OAuth request").toBeTruthy(); + expect( + started.status(), + `reconnect OAuth start must succeed (reconnect flag reached the server); body: ${startedBody}`, + ).toBe(200); }); }); }), diff --git a/packages/react/src/components/add-account-modal.tsx b/packages/react/src/components/add-account-modal.tsx index 71da567bc8..1276523df3 100644 --- a/packages/react/src/components/add-account-modal.tsx +++ b/packages/react/src/components/add-account-modal.tsx @@ -12,6 +12,7 @@ import { ProviderKey, identityPathTier, rankResponseSample, + type Connection, type HealthCheckCandidate, type HealthCheckResult, type HealthCheckSpec, @@ -1251,6 +1252,20 @@ function AddAccountModalView(props: AddAccountModalProps) { () => buildUsageMap(AsyncResult.isSuccess(connectionsResult) ? connectionsResult.value : []), [connectionsResult], ); + // A fresh OAuth connect targeting an existing (owner, name) is rejected by + // `oauth.start`, but only after the popup has already been reserved — the + // user sees a window blip open and close. Check against the loaded + // connection list first so the conflict surfaces as an error with no popup. + const connectionNameTaken = useCallback( + (connectionOwner: Owner, name: ConnectionName): boolean => + (AsyncResult.isSuccess(connectionsResult) ? connectionsResult.value : []).some( + (connection: Connection) => + connection.owner === connectionOwner && + connection.integration === integration && + String(connection.name) === String(name), + ), + [connectionsResult, integration], + ); const method = useMemo( () => allMethods.find((m: AuthMethod) => m.id === methodId) ?? allMethods[0], @@ -1660,7 +1675,8 @@ function AddAccountModalView(props: AddAccountModalProps) { success: true, }); }, - onError: () => { + onError: (message: string) => { + toast.error(message); trackEvent("connection_reconnected", { integration_slug: String(integration), owner: connectionOwner, @@ -1907,11 +1923,21 @@ function AddAccountModalView(props: AddAccountModalProps) { integrationName, organizationId, ); + const connectionName = connectionNameFrom( + label, + connectionOwner, + integrationName, + organizationId, + ); + if (connectionNameTaken(connectionOwner, connectionName)) { + oauthPopup.setError(connectionExistsMessage(identityLabel)); + return; + } const payload = { client: chosenClient.slug, clientOwner: chosenClient.owner, owner: connectionOwner, - name: connectionNameFrom(label, connectionOwner, integrationName, organizationId), + name: connectionName, integration, template: method.template, identityLabel, @@ -1985,6 +2011,10 @@ function AddAccountModalView(props: AddAccountModalProps) { const cimdOwner = owner; const connectionName = connectionNameFrom(label, cimdOwner, integrationName, organizationId); const identityLabel = connectionLabelForHost(label, cimdOwner, integrationName, organizationId); + if (connectionNameTaken(cimdOwner, connectionName)) { + toast.error(connectionExistsMessage(identityLabel)); + return; + } setCimdBusy(true); const outcome = await runCimdConnect( { @@ -2061,6 +2091,10 @@ function AddAccountModalView(props: AddAccountModalProps) { const dcrOwner = owner; const connectionName = connectionNameFrom(label, dcrOwner, integrationName, organizationId); const identityLabel = connectionLabelForHost(label, dcrOwner, integrationName, organizationId); + if (connectionNameTaken(dcrOwner, connectionName)) { + toast.error(connectionExistsMessage(identityLabel)); + return; + } setDcrBusy(true); const outcome = await runDcrConnect( { diff --git a/packages/react/src/plugins/oauth-sign-in.tsx b/packages/react/src/plugins/oauth-sign-in.tsx index 844167be42..f1429cf398 100644 --- a/packages/react/src/plugins/oauth-sign-in.tsx +++ b/packages/react/src/plugins/oauth-sign-in.tsx @@ -386,6 +386,7 @@ export function useOAuthPopupFlow< template: input.payload.template, identityLabel: input.payload.identityLabel, redirectUri: input.payload.redirectUri ?? oauthCallbackUrl(callbackPath), + reconnect: input.payload.reconnect, }, }).then((exit) => Exit.isSuccess(exit) From 62d6c90151e290c31c72b6ee0ae74a98869bce63 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:33:06 -0700 Subject: [PATCH 06/13] Use a toast for the BYO pre-check conflict, matching CIMD and DCR --- packages/react/src/components/add-account-modal.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/react/src/components/add-account-modal.tsx b/packages/react/src/components/add-account-modal.tsx index 1276523df3..7dcca36735 100644 --- a/packages/react/src/components/add-account-modal.tsx +++ b/packages/react/src/components/add-account-modal.tsx @@ -1930,7 +1930,7 @@ function AddAccountModalView(props: AddAccountModalProps) { organizationId, ); if (connectionNameTaken(connectionOwner, connectionName)) { - oauthPopup.setError(connectionExistsMessage(identityLabel)); + toast.error(connectionExistsMessage(identityLabel)); return; } const payload = { From c4b8217929c433b6714757c9472c12b7772e1b9f Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:05:10 -0700 Subject: [PATCH 07/13] Write connection credentials only after winning the row insert --- e2e/cloud/connections-credentials.test.ts | 160 ++++++++++++++------ packages/core/sdk/src/connections.test.ts | 172 ++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 55 ++++++- 3 files changed, 337 insertions(+), 50 deletions(-) diff --git a/e2e/cloud/connections-credentials.test.ts b/e2e/cloud/connections-credentials.test.ts index 2386c7db79..5f970a0a86 100644 --- a/e2e/cloud/connections-credentials.test.ts +++ b/e2e/cloud/connections-credentials.test.ts @@ -7,6 +7,7 @@ // removal really removes; and unknown connections fail with a typed // not-found error. import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; import { expect } from "@effect/vitest"; import { Effect } from "effect"; @@ -23,7 +24,8 @@ type Client = HttpApiClient.ForApi; const TEMPLATE_API_KEY = AuthTemplateSlug.make("apiKey"); -/** Minimal OpenAPI spec with a single GET /ping — never contacted here. */ +/** Minimal OpenAPI spec with a single GET /ping — only the conflict scenario + * ever contacts it, to prove which stored secret the connection resolves. */ const pingSpec = JSON.stringify({ openapi: "3.0.3", info: { title: "Ping API", version: "1.0.0" }, @@ -34,15 +36,58 @@ const pingSpec = JSON.stringify({ }, }); -/** Registers a fresh apiKey-authenticated integration for connections to bind to. */ -const registerIntegration = (client: Client) => +type CaptureUpstream = { + readonly url: string; + /** Every Authorization header `GET /ping` has received, in order. */ + readonly authorizationHeaders: () => readonly string[]; + readonly close: () => void; +}; + +/** Upstream on 127.0.0.1 that records the Authorization header of every + * `GET /ping`. This is how a scenario proves WHICH stored secret a connection + * resolves, since no endpoint ever echoes the value itself. */ +const serveCaptureUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + const headers: string[] = []; + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith("/ping")) { + headers.push(request.headers.authorization ?? ""); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ pong: true })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + authorizationHeaders: () => [...headers], + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (upstream) => Effect.sync(upstream.close), + ); + +/** Registers a fresh apiKey-authenticated integration for connections to bind + * to. Identifier-safe slug: it becomes a property path in sandbox code. */ +const registerIntegration = (client: Client, baseUrl = "http://127.0.0.1:59999") => Effect.gen(function* () { - const slug = IntegrationSlug.make(`conn-scn-${randomBytes(4).toString("hex")}`); + const slug = IntegrationSlug.make(`connscn${randomBytes(4).toString("hex")}`); yield* client.openapi.addSpec({ payload: { spec: { kind: "blob", value: pingSpec }, slug, - baseUrl: "http://127.0.0.1:59999", // never contacted during registration + baseUrl, // the default is never contacted during registration authenticationTemplate: [ { slug: "apiKey", @@ -105,53 +150,80 @@ scenario( scenario( "Connections · re-creating the same connection is rejected and leaves the original intact", {}, - Effect.gen(function* () { - const target = yield* Target; - const { client: apiClient } = yield* Api; - const identity = yield* target.newIdentity(); - const client = yield* apiClient(api, identity); - const integration = yield* registerIntegration(client); - const name = freshConnectionName(); + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: apiClient } = yield* Api; + const identity = yield* target.newIdentity(); + const client = yield* apiClient(api, identity); + const upstream = yield* serveCaptureUpstream(); + const integration = yield* registerIntegration(client, upstream.url); + const name = freshConnectionName(); - yield* client.connections.create({ - payload: { - owner: "org", - name, - integration, - template: TEMPLATE_API_KEY, - identityLabel: "first key", - value: "first-value", - }, - }); - const first = yield* client.connections.list({ query: { integration } }); - expect( - first.filter((connection) => connection.name === name).map((c) => c.identityLabel), - "the first create stores one row with its label", - ).toEqual(["first key"]); - - const error = yield* client.connections - .create({ + yield* client.connections.create({ payload: { owner: "org", name, integration, template: TEMPLATE_API_KEY, - identityLabel: "clobber attempt", - value: "second-value", + identityLabel: "first key", + value: "first-value", }, - }) - .pipe(Effect.flip); - expect( - (error as { _tag?: string })._tag, - "re-creating the same (owner, integration, name) fails with the typed conflict", - ).toBe("ConnectionAlreadyExistsError"); + }); + const first = yield* client.connections.list({ query: { integration } }); + expect( + first.filter((connection) => connection.name === name).map((c) => c.identityLabel), + "the first create stores one row with its label", + ).toEqual(["first key"]); - const second = yield* client.connections.list({ query: { integration } }); - expect( - second.filter((connection) => connection.name === name).map((c) => c.identityLabel), - "the rejected create left the original row untouched", - ).toEqual(["first key"]); - }), + const error = yield* client.connections + .create({ + payload: { + owner: "org", + name, + integration, + template: TEMPLATE_API_KEY, + identityLabel: "clobber attempt", + value: "second-value", + }, + }) + .pipe(Effect.flip); + expect( + (error as { _tag?: string })._tag, + "re-creating the same (owner, integration, name) fails with the typed conflict", + ).toBe("ConnectionAlreadyExistsError"); + + const second = yield* client.connections.list({ query: { integration } }); + expect( + second.filter((connection) => connection.name === name).map((c) => c.identityLabel), + "the rejected create left the original row untouched", + ).toEqual(["first key"]); + + // The stored SECRET is intact too, not just the metadata: invoking + // through the original connection must still authenticate upstream with + // the first value. The pasted value's provider item id is derived from + // the connection name, so a rejected create that wrote before losing + // would have replaced the credential while every metadata read above + // still looked untouched. + const tools = yield* client.tools.list({ query: { integration } }); + const address = tools + .map((tool) => String(tool.address)) + .find((toolAddress) => toolAddress.includes(".ping.")); + expect(address, "the ping tool is in the catalog").toBeDefined(); + if (address === undefined) return; + + const execution = yield* client.executions.execute({ + payload: { + code: [`const result = await ${address}({});`, "return result;"].join("\n"), + }, + }); + expect(execution.status, "the invoke completes").toBe("completed"); + expect( + upstream.authorizationHeaders(), + "the original connection still authenticates with the first value", + ).toEqual(["Bearer first-value"]); + }), + ), ); scenario( diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 80ae162308..06f6944757 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -20,6 +20,7 @@ import { ToolAddress, ToolName, } from "./ids"; +import { ConnectionAlreadyExistsError } from "./errors"; import { createExecutor } from "./executor"; import { StorageError, type FumaDb } from "./fuma-runtime"; import { HealthCheckResult } from "./health-check"; @@ -244,6 +245,177 @@ describe("connections.create", () => { }), ); + // The race the early duplicate check cannot answer: two creates for the same + // (owner, integration, name) in flight at once. The row insert picks the + // winner and the loser gets the typed 409 — and, the load-bearing part, the + // provider write is winner-only. A pasted value's item id is deterministic, + // so a pre-insert write from the LOSING create would silently replace the + // winner's stored secret while the winner's row keeps resolving through it. + // The gate parks the first create inside its provider write, so the second + // create runs its full duplicate handling while the first is mid-flight. + it.effect("concurrent creates: one winner, a typed 409, and the winner's secret intact", () => + Effect.scoped( + Effect.gen(function* () { + const firstWriteEntered = yield* Deferred.make(); + const releaseFirstWrite = yield* Deferred.make(); + const store = new Map(); + let writes = 0; + const gatedProvider: CredentialProvider = { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => + Effect.gen(function* () { + writes += 1; + if (writes === 1) { + yield* Deferred.succeed(firstWriteEntered, undefined); + yield* Deferred.await(releaseFirstWrite); + } + store.set(String(id), value); + }), + }; + const gatedPlugin = definePlugin(() => ({ + id: "gated" as const, + credentialProviders: [gatedProvider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [{ name: ToolName.make("deploy"), description: "deploy" }], + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + resolveValue: (name: string) => + ctx.connections.resolveValue({ + owner: "org", + integration: INTEG, + name: ConnectionName.make(name), + }), + }), + }))(); + const config = makeTestConfig({ plugins: [gatedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.gated.seed(); + + const createWith = (value: string) => + Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value, + }), + ); + + const firstFiber = yield* Effect.forkChild(createWith("first-value")); + yield* Deferred.await(firstWriteEntered); + const second = yield* createWith("second-value"); + yield* Deferred.succeed(releaseFirstWrite, undefined); + const first = yield* Fiber.join(firstFiber); + + const attempts = [ + { result: first, value: "first-value" }, + { result: second, value: "second-value" }, + ]; + const winners = attempts.filter((attempt) => Result.isSuccess(attempt.result)); + const losers = attempts.filter((attempt) => Result.isFailure(attempt.result)); + expect(winners).toHaveLength(1); + expect(losers).toHaveLength(1); + const loser = losers[0]; + if (!loser || !Result.isFailure(loser.result)) return; + expect(loser.result.failure).toBeInstanceOf(ConnectionAlreadyExistsError); + + // Exactly one connection survived, and it resolves to the WINNER's + // value — the losing create never reached the provider. + const connections = yield* executor.connections.list(); + expect(connections).toHaveLength(1); + const value = yield* executor.gated.resolveValue("main"); + expect(value).toBe(winners[0]?.value); + }), + ), + ); + + // When both creates observe absence, both reach the insert and the primary + // key breaks the tie — the loser must still get the typed 409, not a raw + // unique-constraint storage failure. The proxy blinds every connection-table + // read for the second create, so its early and transactional checks both + // miss the existing row and its insert genuinely collides in the database. + it.effect("maps a lost insert race to the typed 409, not a storage failure", () => + Effect.gen(function* () { + let blind = false; + const blindfoldConnectionReads = (db: FumaDb): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => + blind && table === "connection" + ? Promise.resolve(null) + : (target.findFirst as (t: unknown, q: unknown) => Promise)( + table, + query, + ); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); + }; + + const config = makeTestConfig({ plugins: [demoPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: blindfoldConnectionReads(config.db), + }); + yield* executor.demo.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "first-value", + }); + + blind = true; + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "second-value", + }), + ); + blind = false; + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(result.failure).toBeInstanceOf(ConnectionAlreadyExistsError); + + // The losing insert wrote nothing: the original secret is untouched. + const connections = yield* executor.connections.list(); + expect(connections).toHaveLength(1); + const value = yield* executor.demo.resolveValue("org", "main"); + expect(value).toBe("first-value"); + }), + ); + it.effect("allows the same name under a different owner", () => Effect.gen(function* () { const executor = yield* setup(); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 2727d07f7f..ef8bd780df 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3273,11 +3273,11 @@ export const createExecutor = = {}; + // Pasted-value provider writes, built here but run only AFTER this + // create wins the row insert below. + const pastedWrites: Effect.Effect[] = []; if (external.length > 0 && pasted.length > 0) { return yield* new InvalidConnectionInputError({ message: "A connection cannot mix pasted and external-provider inputs.", @@ -3352,8 +3355,12 @@ export const createExecutor = + Effect.fail( + new ConnectionAlreadyExistsError({ + owner: input.owner, + integration: input.integration, + name, + }), + ), + ), ); + // Winner-only credential write: only the create whose row insert + // committed may touch the provider, so a losing create can never + // clobber the winner's (or a pre-existing connection's) secret. While + // this row exists no concurrent create can win, so on failure the row + // is ours to remove — a surviving row whose item_ids were never stored + // would fail every invocation with `connection_value_missing`. + if (pastedWrites.length > 0) { + yield* Effect.all(pastedWrites).pipe( + Effect.tapError(() => + core + .deleteMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(input.owner)(b), + b("integration", "=", String(input.integration)), + b("name", "=", String(name)), + ), + }) + .pipe(Effect.ignore), + ), + ); + } + // Record the sighting. The request seam (`makeScopedExecutor`) already // does this for every hosted call, so this is the belt for direct // SDK/CLI callers that never pass through it — a connecting principal From 012a027de5c21b47e73b9369616a0af97a0415a8 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:31:01 -0700 Subject: [PATCH 08/13] Defer connection credential writes until the row insert is durable --- packages/core/sdk/src/connections.test.ts | 270 ++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 159 +++++++++++-- 2 files changed, 407 insertions(+), 22 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 06f6944757..dcd332eaca 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -575,6 +575,276 @@ describe("connections.create", () => { ); }); +// --------------------------------------------------------------------------- +// Credential-write durability & compensation. The row insert and the provider +// write cannot be atomic — the provider may live outside the database — so the +// create sequences them: the row must be DURABLE before the provider is +// touched, and a write that does not complete must tear down everything it +// already stored. The worst outcome is a committed row whose credentials were +// never written: it 409s every retry while resolving nothing. +// --------------------------------------------------------------------------- + +const trackingProvider = ( + store: Map, + overrides?: Partial>, +): CredentialProvider => ({ + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + delete: (id) => Effect.sync(() => void store.delete(String(id))), + ...overrides, +}); + +const durabilityPlugin = (provider: CredentialProvider) => + definePlugin(() => ({ + id: "durable" as const, + credentialProviders: [provider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + resolveValue: (name: string) => + ctx.connections.resolveValue({ + owner: "org", + integration: INTEG, + name: ConnectionName.make(name), + }), + /** A plugin composing a create into its own atomic unit — the shape that + * makes the create's inner insert a pass-through with no commit of its + * own. */ + createInTransaction: (rollback: boolean) => + ctx.transaction( + Effect.gen(function* () { + const created = yield* ctx.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + if (rollback) return yield* Effect.fail("rollback" as const); + return created; + }), + ), + }), + }))(); + +describe("connections.create credential-write durability", () => { + // `transaction` nests by pass-through: inside a plugin's `ctx.transaction` + // the create's inner insert commits nothing — the row becomes durable only + // with the OUTER commit. A credential written inline in that window outlives + // a rollback as an orphan at a deterministic item id, where the next create + // of the same name silently adopts it. The write must wait for the real + // commit, and the caller's own typed failure must pass through untouched. + it.effect("discards the credential write when an enclosing transaction rolls back", () => + Effect.gen(function* () { + const store = new Map(); + const executor = yield* makeTestExecutor({ + plugins: [durabilityPlugin(trackingProvider(store))] as const, + }); + yield* executor.durable.seed(); + + const result = yield* Effect.result(executor.durable.createInTransaction(true)); + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(result.failure).toBe("rollback"); + + // No zombie row and no orphaned credential survived the rollback. + expect(yield* executor.connections.list()).toEqual([]); + expect(store.size).toBe(0); + + // The name is fully reusable: a committed create stores its own value. + yield* executor.durable.createInTransaction(false); + expect((yield* executor.connections.list()).length).toBe(1); + expect(yield* executor.durable.resolveValue("main")).toBe("secret-token"); + }), + ); + + // Inside an enclosing transaction the create returns before its deferred + // credential write runs, so a write failure can no longer become the + // caller's typed error. What it must NOT become is a silent zombie: the + // deferred failure removes the committed row (and its tools) and reports + // the failure loudly. + it.effect("a deferred credential-write failure removes the committed row and logs", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: () => + Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })), + }); + const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); + yield* executor.durable.seed(); + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.durable.createInTransaction(false).pipe(Effect.provide(Logger.layer([capture]))), + ); + + // The transaction committed, so the create itself reported success ... + expect(Result.isSuccess(result)).toBe(true); + // ... but the failed write tore the committed row (and its tools) down + // instead of leaving it to 409 every retry while resolving nothing. + expect(yield* executor.connections.list()).toEqual([]); + expect(yield* executor.tools.list()).toEqual([]); + expect(errors.some((line) => line.includes("failed after commit"))).toBe(true); + }), + ); + + // Interruption is not an error: error-channel compensation never sees it. A + // create interrupted mid-write must still tear down what it already did — + // the committed row and every item that landed before the interrupt. + it.effect("an interrupted create removes the row and the items it already wrote", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Deferred.succeed(secondWriteEntered, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.sync(() => void store.set(String(id), value)), + }); + const executor = yield* makeTestExecutor({ + plugins: [durabilityPlugin(provider)] as const, + }); + yield* executor.durable.seed(); + + const fiber = yield* Effect.forkChild( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + yield* Deferred.await(secondWriteEntered); + yield* Fiber.interrupt(fiber); + + // The first item had landed before the interrupt; compensation removed + // it together with the row it belonged to. + expect(store.size).toBe(0); + expect(yield* executor.connections.list()).toEqual([]); + }), + ), + ); + + // One provider.set can succeed and a later one fail. Deleting only the row + // leaves the earlier secret at its deterministic item id, waiting to be + // adopted by the next create of the same name. Compensation must remove the + // items already written, not just the row. + it.effect("a failed later variable write cleans up the earlier items and the row", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); + yield* executor.durable.seed(); + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + // Neither half survives: the first item is gone with the row. + expect(store.size).toBe(0); + expect(yield* executor.connections.list()).toEqual([]); + }), + ); + + // The compensating delete can itself fail. Swallowing that failure strands + // a visible credential-less row behind an error that never mentions it. The + // create must fail with an error that NAMES the stranded connection so an + // operator can act on it. + it.effect("names the stranded connection when the compensating delete fails", () => + Effect.gen(function* () { + let failRowDelete = false; + const failConnectionDeletes = (db: FumaDb): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "deleteMany") { + return (table: unknown, query: unknown) => + failRowDelete && table === "connection" + ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + Promise.reject( + new StorageError({ message: "delete refused", cause: undefined }), + ) + : (target.deleteMany as (t: unknown, q: unknown) => Promise)( + table, + query, + ); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); + }; + + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ ...config, db: failConnectionDeletes(config.db) }); + yield* executor.durable.seed(); + failRowDelete = true; + + const result = yield* Effect.result( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + }), + ); +}); + describe("connections.list / get", () => { it.effect("only includes full health diagnostics in verbose core tool output", () => Effect.gen(function* () { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 5ba2d862ca..c571d0a500 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -2,6 +2,7 @@ import { Deferred, Duration, Effect, + Exit, Fiber, Inspectable, Layer, @@ -18,6 +19,7 @@ import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/sch import type { AnyColumn } from "@executor-js/fumadb/schema"; import { StorageError, + activeFumaDbRef, afterCommit, isStorageFailure, makeFumaClient, @@ -3388,9 +3390,15 @@ export const createExecutor = = {}; - // Pasted-value provider writes, built here but run only AFTER this - // create wins the row insert below. - const pastedWrites: Effect.Effect[] = []; + // Pasted-value provider writes, built here but run only after the row + // insert below is DURABLE (the post-insert block owns that ordering). + // Each entry carries its own undo so a write that does not complete + // can tear down exactly the items it already stored. + const pastedWrites: Array<{ + readonly itemId: ProviderItemId; + readonly write: Effect.Effect; + readonly remove: Effect.Effect | null; + }> = []; if (external.length > 0 && pasted.length > 0) { return yield* new InvalidConnectionInputError({ message: "A connection cannot mix pasted and external-provider inputs.", @@ -3431,7 +3439,12 @@ export const createExecutor = 0) { - yield* Effect.all(pastedWrites).pipe( - Effect.tapError(() => - core - .deleteMany("connection", { - where: (b: AnyCb) => - b.and( - byOwner(input.owner)(b), - b("integration", "=", String(input.integration)), - b("name", "=", String(name)), + const written: ProviderItemId[] = []; + const writeAll = Effect.gen(function* () { + for (const entry of pastedWrites) { + yield* entry.write; + written.push(entry.itemId); + } + }); + + // While the committed row exists no concurrent create can win, so + // on an incomplete write it is ours to tear down — a surviving row + // whose item_ids were never stored would 409 every retry while + // failing every invocation with `connection_value_missing`. + // Compensation covers BOTH halves: best-effort deletes of the items + // already written (the earlier secrets of a partial multi-variable + // write), then the row itself plus any tool rows a deferred create + // produced before its commit. Nothing here is silent: every failed + // undo is logged, and `rowRemoved` lets the immediate path convert + // a stranded row into an error that names it. + let rowRemoved = true; + const logContext = { + owner: input.owner, + integration: String(input.integration), + connection: String(name), + }; + const compensate = Effect.gen(function* () { + for (const entry of pastedWrites) { + if (entry.remove === null || !written.includes(entry.itemId)) continue; + yield* entry.remove.pipe( + Effect.catchCause((cause) => + Effect.logError("executor connection create failed to undo a credential write", { + ...logContext, + item: String(entry.itemId), + cause, + }), + ), + ); + } + const where = (b: AnyCb) => + b.and( + byOwner(input.owner)(b), + b("integration", "=", String(input.integration)), + b("connection", "=", String(name)), + ); + yield* Effect.gen(function* () { + yield* core.deleteMany("tool", { where }); + yield* core.deleteMany("definition", { where }); + yield* core.deleteMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(input.owner)(b), + b("integration", "=", String(input.integration)), + b("name", "=", String(name)), + ), + }); + }).pipe( + Effect.catchCause((cause) => + Effect.sync(() => { + rowRemoved = false; + }).pipe( + Effect.andThen( + Effect.logError( + "executor connection create stranded a credential-less connection row", + { ...logContext, cause }, ), - }) - .pipe(Effect.ignore), - ), + ), + ), + ), + ); + }); + + // `onExit`, not `tapError`: compensation must also run when the + // write is interrupted or dies with a defect. + const guardedWrites = writeAll.pipe( + Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)), ); + + const enclosingTransaction = yield* Effect.service(activeFumaDbRef); + if (enclosingTransaction === null) { + yield* guardedWrites.pipe( + Effect.catch((error) => + rowRemoved + ? Effect.fail(error) + : Effect.fail( + new StorageError({ + message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`, + cause: error, + }), + ), + ), + ); + } else { + // The create returns to its enclosing transaction before this + // hook runs, so a failure here can no longer become the caller's + // typed error — compensation plus a loud log IS the contract. + // Between the outer commit and this hook the row is briefly + // visible without its credentials; closing that window takes a + // reservation/staging scheme and is deliberately out of scope. + yield* afterCommit( + guardedWrites.pipe( + Effect.catchCause((cause) => + Effect.logError( + "executor connection create credential write failed after commit", + { ...logContext, rowRemoved, cause }, + ), + ), + ), + ); + } } // Record the sighting. The request seam (`makeScopedExecutor`) already From 7c9c5fe9f248afb09e000ec8bc1e30fec79deeaa Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 13:01:31 -0700 Subject: [PATCH 09/13] Run credential writes inline after the winning insert --- packages/core/sdk/src/connections.test.ts | 160 +++++++++------------- packages/core/sdk/src/executor.ts | 140 +++++++++---------- 2 files changed, 127 insertions(+), 173 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index dcd332eaca..508bde5f79 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -576,17 +576,17 @@ describe("connections.create", () => { }); // --------------------------------------------------------------------------- -// Credential-write durability & compensation. The row insert and the provider -// write cannot be atomic — the provider may live outside the database — so the -// create sequences them: the row must be DURABLE before the provider is -// touched, and a write that does not complete must tear down everything it +// Credential-write compensation. The row insert and the provider write cannot +// be atomic — the provider may live outside the database — so the create +// sequences them: the provider is touched only after this create wins the row +// insert, and a write that does not complete must tear down everything it // already stored. The worst outcome is a committed row whose credentials were // never written: it 409s every retry while resolving nothing. // --------------------------------------------------------------------------- const trackingProvider = ( store: Map, - overrides?: Partial>, + overrides?: Partial>, ): CredentialProvider => ({ key: ProviderKey.make("memory"), writable: true, @@ -607,98 +607,10 @@ const durabilityPlugin = (provider: CredentialProvider) => extension: (ctx) => ({ seed: () => ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), - resolveValue: (name: string) => - ctx.connections.resolveValue({ - owner: "org", - integration: INTEG, - name: ConnectionName.make(name), - }), - /** A plugin composing a create into its own atomic unit — the shape that - * makes the create's inner insert a pass-through with no commit of its - * own. */ - createInTransaction: (rollback: boolean) => - ctx.transaction( - Effect.gen(function* () { - const created = yield* ctx.connections.create({ - owner: "org", - name: ConnectionName.make("main"), - integration: INTEG, - template: TEMPLATE, - value: "secret-token", - }); - if (rollback) return yield* Effect.fail("rollback" as const); - return created; - }), - ), }), }))(); -describe("connections.create credential-write durability", () => { - // `transaction` nests by pass-through: inside a plugin's `ctx.transaction` - // the create's inner insert commits nothing — the row becomes durable only - // with the OUTER commit. A credential written inline in that window outlives - // a rollback as an orphan at a deterministic item id, where the next create - // of the same name silently adopts it. The write must wait for the real - // commit, and the caller's own typed failure must pass through untouched. - it.effect("discards the credential write when an enclosing transaction rolls back", () => - Effect.gen(function* () { - const store = new Map(); - const executor = yield* makeTestExecutor({ - plugins: [durabilityPlugin(trackingProvider(store))] as const, - }); - yield* executor.durable.seed(); - - const result = yield* Effect.result(executor.durable.createInTransaction(true)); - expect(Result.isFailure(result)).toBe(true); - if (!Result.isFailure(result)) return; - expect(result.failure).toBe("rollback"); - - // No zombie row and no orphaned credential survived the rollback. - expect(yield* executor.connections.list()).toEqual([]); - expect(store.size).toBe(0); - - // The name is fully reusable: a committed create stores its own value. - yield* executor.durable.createInTransaction(false); - expect((yield* executor.connections.list()).length).toBe(1); - expect(yield* executor.durable.resolveValue("main")).toBe("secret-token"); - }), - ); - - // Inside an enclosing transaction the create returns before its deferred - // credential write runs, so a write failure can no longer become the - // caller's typed error. What it must NOT become is a silent zombie: the - // deferred failure removes the committed row (and its tools) and reports - // the failure loudly. - it.effect("a deferred credential-write failure removes the committed row and logs", () => - Effect.gen(function* () { - const store = new Map(); - const provider = trackingProvider(store, { - set: () => - Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })), - }); - const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); - yield* executor.durable.seed(); - - const errors: string[] = []; - const capture = Logger.make((options) => { - if (options.logLevel === "Error") { - errors.push(Inspectable.toStringUnknown(options.message, 0)); - } - }); - const result = yield* Effect.result( - executor.durable.createInTransaction(false).pipe(Effect.provide(Logger.layer([capture]))), - ); - - // The transaction committed, so the create itself reported success ... - expect(Result.isSuccess(result)).toBe(true); - // ... but the failed write tore the committed row (and its tools) down - // instead of leaving it to 409 every retry while resolving nothing. - expect(yield* executor.connections.list()).toEqual([]); - expect(yield* executor.tools.list()).toEqual([]); - expect(errors.some((line) => line.includes("failed after commit"))).toBe(true); - }), - ); - +describe("connections.create credential-write compensation", () => { // Interruption is not an error: error-channel compensation never sees it. A // create interrupted mid-write must still tear down what it already did — // the committed row and every item that landed before the interrupt. @@ -773,6 +685,53 @@ describe("connections.create credential-write durability", () => { }), ); + // A provider can expose `set` without `delete`. Compensation then cannot + // undo the items already written — that is acceptable only if it is loud: + // a warning must name the item that may be stranded, never a silent skip. + it.effect("warns about possibly stranded items when the provider has no delete", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + delete: undefined, + }); + const executor = yield* makeTestExecutor({ plugins: [durabilityPlugin(provider)] as const }); + yield* executor.durable.seed(); + + const warnings: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Warn") { + warnings.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + expect(Predicate.isTagged("StorageError")(result.failure)).toBe(true); + // The row is gone, but the first item cannot be undone without a + // provider delete ... + expect(yield* executor.connections.list()).toEqual([]); + expect(store.size).toBe(1); + // ... and the create said so, naming the item. + expect(warnings.some((line) => line.includes("stranded"))).toBe(true); + expect(warnings.some((line) => line.includes("first"))).toBe(true); + }), + ); + // The compensating delete can itself fail. Swallowing that failure strands // a visible credential-less row behind an error that never mentions it. The // create must fail with an error that NAMES the stranded connection so an @@ -841,6 +800,19 @@ describe("connections.create credential-write durability", () => { if (!Predicate.isTagged("StorageError")(failure)) return; expect(failure.message).toContain("main"); expect(failure.message).toContain("vercel"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // Non-vacuous: the compensating delete really did fail, so the + // credential-less row the error names is still there. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); }), ); }); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index c571d0a500..5f3c6daf5f 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -19,7 +19,6 @@ import { schema as fumaSchema, type RelationsMap } from "@executor-js/fumadb/sch import type { AnyColumn } from "@executor-js/fumadb/schema"; import { StorageError, - activeFumaDbRef, afterCommit, isStorageFailure, makeFumaClient, @@ -3390,10 +3389,10 @@ export const createExecutor = = {}; - // Pasted-value provider writes, built here but run only after the row - // insert below is DURABLE (the post-insert block owns that ordering). - // Each entry carries its own undo so a write that does not complete - // can tear down exactly the items it already stored. + // Pasted-value provider writes, built here but run only AFTER this + // create wins the row insert below. Each entry carries its own undo + // so a write that does not complete can tear down exactly the items + // it already stored. const pastedWrites: Array<{ readonly itemId: ProviderItemId; readonly write: Effect.Effect; @@ -3504,20 +3503,26 @@ export const createExecutor = 0) { const written: ProviderItemId[] = []; const writeAll = Effect.gen(function* () { @@ -3533,10 +3538,9 @@ export const createExecutor = Effect.logError("executor connection create failed to undo a credential write", { @@ -3556,77 +3569,46 @@ export const createExecutor = - b.and( - byOwner(input.owner)(b), - b("integration", "=", String(input.integration)), - b("connection", "=", String(name)), - ); - yield* Effect.gen(function* () { - yield* core.deleteMany("tool", { where }); - yield* core.deleteMany("definition", { where }); - yield* core.deleteMany("connection", { + yield* core + .deleteMany("connection", { where: (b: AnyCb) => b.and( byOwner(input.owner)(b), b("integration", "=", String(input.integration)), b("name", "=", String(name)), ), - }); - }).pipe( - Effect.catchCause((cause) => - Effect.sync(() => { - rowRemoved = false; - }).pipe( - Effect.andThen( - Effect.logError( - "executor connection create stranded a credential-less connection row", - { ...logContext, cause }, + }) + .pipe( + Effect.catchCause((cause) => + Effect.sync(() => { + rowRemoved = false; + }).pipe( + Effect.andThen( + Effect.logError( + "executor connection create stranded a credential-less connection row", + { ...logContext, cause }, + ), ), ), ), - ), - ); + ); }); // `onExit`, not `tapError`: compensation must also run when the // write is interrupted or dies with a defect. - const guardedWrites = writeAll.pipe( + yield* writeAll.pipe( Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)), - ); - - const enclosingTransaction = yield* Effect.service(activeFumaDbRef); - if (enclosingTransaction === null) { - yield* guardedWrites.pipe( - Effect.catch((error) => - rowRemoved - ? Effect.fail(error) - : Effect.fail( - new StorageError({ - message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`, - cause: error, - }), - ), - ), - ); - } else { - // The create returns to its enclosing transaction before this - // hook runs, so a failure here can no longer become the caller's - // typed error — compensation plus a loud log IS the contract. - // Between the outer commit and this hook the row is briefly - // visible without its credentials; closing that window takes a - // reservation/staging scheme and is deliberately out of scope. - yield* afterCommit( - guardedWrites.pipe( - Effect.catchCause((cause) => - Effect.logError( - "executor connection create credential write failed after commit", - { ...logContext, rowRemoved, cause }, + Effect.catch((error) => + rowRemoved + ? Effect.fail(error) + : Effect.fail( + new StorageError({ + message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`, + cause: error, + }), ), - ), - ), - ); - } + ), + ); } // Record the sighting. The request seam (`makeScopedExecutor`) already From a988195a9467efcf52f2d666c99fa482b185c140 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 13:43:21 -0700 Subject: [PATCH 10/13] Guard connection-create compensation by inserted row identity --- .changeset/connection-create-conflict.md | 2 +- packages/core/sdk/src/connections.test.ts | 267 +++++++++++++++++++--- packages/core/sdk/src/executor.ts | 158 +++++++++---- 3 files changed, 348 insertions(+), 79 deletions(-) diff --git a/.changeset/connection-create-conflict.md b/.changeset/connection-create-conflict.md index 95883a9120..7bc40fa492 100644 --- a/.changeset/connection-create-conflict.md +++ b/.changeset/connection-create-conflict.md @@ -6,6 +6,6 @@ `connections.create` used to upsert: a create with the same (owner, integration, name) replaced the saved connection and, for a pasted value, overwrote the stored secret itself. It now fails with the new `ConnectionAlreadyExistsError` and leaves the existing connection untouched. Remove the connection first, or pick a different name. -This adds one error to the wire contract: the `POST /connections` endpoint can answer **HTTP 409** with tag `ConnectionAlreadyExistsError`, and the `connections.create` core tool resolves the same case as `{ ok: false, error: { code: "connection_already_exists" } }`. The change is additive — no existing status, field, or success shape moves. +This adds one error to the wire contract: the `POST /connections` endpoint can answer **HTTP 409** with tag `ConnectionAlreadyExistsError`, and the `connections.create` core tool resolves the same case as `{ ok: false, error: { code: "connection_already_exists" } }`. The core tool now also resolves the other expected input failures the same way instead of as opaque internal errors: `integration_not_found` for an unknown integration and `invalid_connection_input` for an invalid input. The change is additive — no existing status, field, or success shape moves. OAuth is unaffected. Fresh OAuth connects already resolve a taken name to the next free suffix through `newConnection`, and reconnect still re-mints the same connection on purpose. diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 508bde5f79..e1de2c82f1 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1,7 +1,9 @@ import { describe, expect, it } from "@effect/vitest"; import { + Cause, Deferred, Effect, + Exit, Fiber, Inspectable, Logger, @@ -610,6 +612,37 @@ const durabilityPlugin = (provider: CredentialProvider) => }), }))(); +/** Wrap a test `FumaDb` so deletes on the `connection` table can be made to + * fail on demand — the raw driver-level failure the compensating delete must + * survive loudly. Transactions hand out wrapped handles too, so the guarded + * delete inside the compensation transaction is covered. */ +const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "deleteMany") { + return (table: unknown, query: unknown) => + shouldFail() && table === "connection" + ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + Promise.reject(new StorageError({ message: "delete refused", cause: undefined })) + : (target.deleteMany as (t: unknown, q: unknown) => Promise)(table, query); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + describe("connections.create credential-write compensation", () => { // Interruption is not an error: error-channel compensation never sees it. A // create interrupted mid-write must still tear down what it already did — @@ -739,38 +772,6 @@ describe("connections.create credential-write compensation", () => { it.effect("names the stranded connection when the compensating delete fails", () => Effect.gen(function* () { let failRowDelete = false; - const failConnectionDeletes = (db: FumaDb): FumaDb => { - const wrap = (inner: FumaDb): FumaDb => - new Proxy(inner, { - get(target, prop) { - if (prop === "withContext") { - return (context: unknown) => - wrap((target.withContext as (c: unknown) => FumaDb)(context)); - } - if (prop === "transaction") { - return (run: (tx: FumaDb) => Promise) => - (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( - (tx) => run(wrap(tx)), - ); - } - if (prop === "deleteMany") { - return (table: unknown, query: unknown) => - failRowDelete && table === "connection" - ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle - Promise.reject( - new StorageError({ message: "delete refused", cause: undefined }), - ) - : (target.deleteMany as (t: unknown, q: unknown) => Promise)( - table, - query, - ); - } - return Reflect.get(target, prop); - }, - }); - return wrap(db); - }; - const store = new Map(); const provider = trackingProvider(store, { set: (id, value) => @@ -779,7 +780,10 @@ describe("connections.create credential-write compensation", () => { : Effect.sync(() => void store.set(String(id), value)), }); const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); - const executor = yield* createExecutor({ ...config, db: failConnectionDeletes(config.db) }); + const executor = yield* createExecutor({ + ...config, + db: failableConnectionDeletes(config.db, () => failRowDelete), + }); yield* executor.durable.seed(); failRowDelete = true; @@ -808,13 +812,208 @@ describe("connections.create credential-write compensation", () => { expect(failure.cause.message).toBe("provider write refused"); // Non-vacuous: the compensating delete really did fail, so the - // credential-less row the error names is still there. + // row the error names is still there. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + }), + ); + + // Compensation can be slow (provider calls). In that window a concurrent + // remove can free the name and a new create can take it, writing fresh + // secrets at the SAME deterministic item ids. Late compensation must then + // recognize that the row is no longer the one it inserted — identified by + // the storage surrogate row id — and touch neither the replacement row nor + // its credentials. Losing compensation to a concurrent remove is correct: + // the remover already cleaned up. + it.effect("late compensation leaves a concurrent replacement untouched", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const releaseSecondWrite = yield* Deferred.make(); + const store = new Map(); + let parkNextSecondWrite = true; + const provider = trackingProvider(store, { + set: (id, value) => { + if (String(id).endsWith(":second") && parkNextSecondWrite) { + parkNextSecondWrite = false; + return Deferred.succeed(secondWriteEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecondWrite)), + Effect.andThen( + Effect.fail( + new StorageError({ message: "provider write refused", cause: undefined }), + ), + ), + ); + } + return Effect.sync(() => void store.set(String(id), value)); + }, + }); + const executor = yield* makeTestExecutor({ + plugins: [durabilityPlugin(provider)] as const, + }); + yield* executor.durable.seed(); + + const fiber = yield* Effect.forkChild( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "a-1", second: "a-2" }, + }), + ); + yield* Deferred.await(secondWriteEntered); + + // While the first create is parked in its provider write, the user + // removes the connection and recreates it with different secrets. + yield* executor.connections.remove({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "c-1", second: "c-2" }, + }); + + // Release the parked write: the first create fails and compensates + // late, against a name it no longer owns. + yield* Deferred.succeed(releaseSecondWrite, undefined); + const exit = yield* Fiber.await(fiber); + expect(Exit.isFailure(exit)).toBe(true); + + // The replacement row AND its credentials survive. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.get("connection:org:vercel:main:first")).toBe("c-1"); + expect(store.get("connection:org:vercel:main:second")).toBe("c-2"); + }), + ), + ); + + // A provider write can die with a defect instead of failing. The stranded- + // row promise must hold there too: a defect followed by a failed + // compensating delete surfaces the same typed StorageError naming the + // stranded connection, not an anonymous crash. + it.effect("a defect followed by a failed row delete still names the stranded connection", () => + Effect.gen(function* () { + let failRowDelete = false; + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.die("provider crashed") + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConnectionDeletes(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + failRowDelete = true; + + const exit = yield* Effect.exit( + executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }), + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + // Not an anonymous defect: the typed error is on the failure channel. + expect(Cause.hasFails(exit.cause)).toBe(true); + const failure = Cause.squash(exit.cause); + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure)).toBe(true); + if (!isStorageError(failure)) return; + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("stranded"); + // The original defect is retained as the cause, not replaced. + expect(failure.cause).toBe("provider crashed"); + + // Non-vacuous: the row the error names is still there. failRowDelete = false; const rows = yield* executor.connections.list(); expect(rows.length).toBe(1); expect(String(rows[0]?.name)).toBe("main"); }), ); + + // Interruption cannot carry a typed error — interrupting wins over failing + // — so when an interrupted create cannot delete its row, the stranded row + // is reported through a loud error log and the create stays an + // interruption. The items that landed before the interrupt stay with the + // stranded row: credential teardown is gated on the row delete succeeding. + it.effect("an interrupted create with a failed row delete logs the stranded row", () => + Effect.scoped( + Effect.gen(function* () { + let failRowDelete = false; + const secondWriteEntered = yield* Deferred.make(); + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Deferred.succeed(secondWriteEntered, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConnectionDeletes(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const fiber = yield* Effect.forkChild( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + yield* Deferred.await(secondWriteEntered); + failRowDelete = true; + yield* Fiber.interrupt(fiber); + const exit = yield* Fiber.await(fiber); + + // Still an interruption — and the stranded row was reported loudly. + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true); + expect(errors.some((line) => line.includes("stranded"))).toBe(true); + + // Non-vacuous: the row survived the failed delete, and the item that + // landed before the interrupt stayed with it. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.size).toBe(1); + }), + ), + ); }); describe("connections.list / get", () => { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 5f3c6daf5f..6c504fa927 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,4 +1,5 @@ import { + Cause, Deferred, Duration, Effect, @@ -8,6 +9,7 @@ import { Layer, Option, Predicate, + Ref, Schema, Semaphore, } from "effect"; @@ -3454,7 +3456,19 @@ export const createExecutor = storageFailureFromUnknown("invalid owner", cause), }); const now = new Date(); - yield* transaction( + // The storage surrogate id of the row THIS create inserted. The + // composite key (owner, integration, name) can change hands while a + // failed create is still compensating, and `created_at` round-trips + // at second precision, so neither identifies OUR row — only `row_id` + // does. `FumaRow` deliberately hides `row_id` from domain rows, so it + // is read through a narrow cast. Every adapter generates it ORM-side + // on insert; a create result without it is a broken storage contract + // and fails here, inside the transaction, before any provider write. + const rowIdOf = (row: unknown): string | null => { + const value = row == null ? null : (row as Record)["row_id"]; + return typeof value === "string" ? value : null; + }; + const insertedRowId = yield* transaction( Effect.gen(function* () { const existing = yield* findConnectionRow({ owner: input.owner, @@ -3468,7 +3482,7 @@ export const createExecutor = ("removed"); const logContext = { owner: input.owner, integration: String(input.integration), connection: String(name), }; const compensate = Effect.gen(function* () { + const rowOutcome = yield* transaction( + Effect.gen(function* () { + const current = yield* findConnectionRow({ + owner: input.owner, + integration: input.integration, + name, + }); + if (rowIdOf(current) !== insertedRowId) { + return "superseded" as const; + } + yield* core.deleteMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(input.owner)(b), + b("integration", "=", String(input.integration)), + b("name", "=", String(name)), + // Even if the row changed hands between the read above + // and this statement, only OUR row can match. + b("row_id", "=", insertedRowId), + ), + }); + return "removed" as const; + }), + ).pipe( + Effect.catchCause((cause) => + Effect.logError( + "executor connection create stranded a connection row it could not delete", + { ...logContext, cause }, + ).pipe(Effect.as("failed" as const)), + ), + ); + yield* Ref.set(rowOutcomeRef, rowOutcome); + if (rowOutcome === "superseded") { + // A concurrent remove took our row, and a successor may already + // own the name and the item ids. The remover cleaned up; + // nothing left here is ours to touch. + yield* Effect.logInfo( + "executor connection create skipped compensation: the connection row was already removed or replaced", + logContext, + ); + return; + } + if (rowOutcome === "failed") { + // The row delete failed, so the row — still ours — keeps + // holding the name together with the items that already + // landed. Leave the items in place (they belong to the + // stranded row the caller is told to remove) and let the exit + // handling below surface the error. + return; + } for (const entry of pastedWrites) { if (!written.includes(entry.itemId)) continue; if (entry.remove === null) { @@ -3569,46 +3655,30 @@ export const createExecutor = - b.and( - byOwner(input.owner)(b), - b("integration", "=", String(input.integration)), - b("name", "=", String(name)), - ), - }) - .pipe( - Effect.catchCause((cause) => - Effect.sync(() => { - rowRemoved = false; - }).pipe( - Effect.andThen( - Effect.logError( - "executor connection create stranded a credential-less connection row", - { ...logContext, cause }, - ), - ), - ), - ), - ); }); // `onExit`, not `tapError`: compensation must also run when the - // write is interrupted or dies with a defect. - yield* writeAll.pipe( + // write is interrupted or dies with a defect. The stranded-row + // promise must hold on every one of those exit shapes, so the exit + // is captured and re-raised by hand: a typed failure or a defect + // that left the row behind becomes the StorageError below, while an + // interruption cannot carry a typed error at all (interrupting wins + // over failing) — for it the loud log inside `compensate` is the + // only signal, and the interruption is re-raised untouched. + const writeExit = yield* writeAll.pipe( Effect.onExit((exit) => (Exit.isSuccess(exit) ? Effect.void : compensate)), - Effect.catch((error) => - rowRemoved - ? Effect.fail(error) - : Effect.fail( - new StorageError({ - message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded without stored credentials and must be removed manually.`, - cause: error, - }), - ), - ), + Effect.exit, ); + if (Exit.isFailure(writeExit)) { + const rowOutcome = yield* Ref.get(rowOutcomeRef); + if (rowOutcome === "failed" && !Cause.hasInterruptsOnly(writeExit.cause)) { + return yield* new StorageError({ + message: `Failed to store credentials for connection ${input.owner}/${String(input.integration)}/${String(name)}, and the compensating delete also failed: the connection row is stranded with incomplete credentials and must be removed manually.`, + cause: Cause.squash(writeExit.cause), + }); + } + return yield* Effect.failCause(writeExit.cause); + } } // Record the sighting. The request seam (`makeScopedExecutor`) already From c9eba7ab39608a502fb791f3600f07d67f30c66a Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 14:15:43 -0700 Subject: [PATCH 11/13] Confirm the guarded compensation delete before undoing credential items --- packages/core/sdk/src/connections.test.ts | 144 ++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 54 +++++++- 2 files changed, 197 insertions(+), 1 deletion(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index e1de2c82f1..76e90fbd60 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -643,6 +643,56 @@ const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): FumaD return wrap(db); }; +/** Wrap a test `FumaDb` so one armed `connection` read observes a stale row. + * This models the read/delete race inside the compensation transaction: under + * read-committed isolation the pre-delete identity read can see this create's + * row while a concurrent remove/recreate has already replaced it by the time + * the guarded delete runs. SQLite serializes the whole transaction, so that + * interleaving cannot be produced with real concurrency here — the wrapper + * reproduces the exact observation order instead: the armed read returns the + * create's own (captured) row while the table already holds the successor; + * every other read, including the confirmation read after the guarded + * delete, sees the real table. */ +const staleCompensationRead = (db: FumaDb, state: { armed: boolean }): FumaDb => { + let captured: Record | null = null; + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "create") { + return async (table: unknown, values: unknown) => { + const row = await ( + target.create as (t: unknown, v: unknown) => Promise> + )(table, values); + // Keep the FIRST inserted connection row — the raced create's own. + if (table === "connection" && captured === null) captured = row; + return row; + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => { + if (table === "connection" && state.armed && captured !== null) { + state.armed = false; + return Promise.resolve(captured); + } + return (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + }; + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + describe("connections.create credential-write compensation", () => { // Interruption is not an error: error-channel compensation never sees it. A // create interrupted mid-write must still tear down what it already did — @@ -897,6 +947,100 @@ describe("connections.create credential-write compensation", () => { ), ); + // fumadb's `deleteMany` returns void, so the guarded delete cannot report + // whether it removed anything. Under read-committed isolation the identity + // read and the delete can straddle a concurrent remove/recreate: the read + // sees this create's row, then the delete matches ZERO rows because a + // successor already holds the name. Treating that zero-row delete as "our + // row is gone, the items are ours to undo" destroys the successor's freshly + // written secrets at the same deterministic item ids. The confirmation read + // after the guarded delete, in the same transaction, must observe the + // surviving row, skip ALL item deletion, and say so. + it.effect("a zero-row guarded delete never touches a successor's credentials", () => + Effect.scoped( + Effect.gen(function* () { + const secondWriteEntered = yield* Deferred.make(); + const releaseSecondWrite = yield* Deferred.make(); + const store = new Map(); + let parkNextSecondWrite = true; + const provider = trackingProvider(store, { + set: (id, value) => { + if (String(id).endsWith(":second") && parkNextSecondWrite) { + parkNextSecondWrite = false; + return Deferred.succeed(secondWriteEntered, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecondWrite)), + Effect.andThen( + Effect.fail( + new StorageError({ message: "provider write refused", cause: undefined }), + ), + ), + ); + } + return Effect.sync(() => void store.set(String(id), value)); + }, + }); + const raceState = { armed: false }; + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: staleCompensationRead(config.db, raceState), + }); + yield* executor.durable.seed(); + + const infos: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Info") { + infos.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const fiber = yield* Effect.forkChild( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "a-1", second: "a-2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + yield* Deferred.await(secondWriteEntered); + + // While the first create is parked in its provider write, the user + // removes the connection and recreates it with different secrets. + yield* executor.connections.remove({ + owner: "org", + integration: INTEG, + name: ConnectionName.make("main"), + }); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "c-1", second: "c-2" }, + }); + + // Arm the stale read and release the parked write: compensation's + // identity read sees the raced create's own row (the race), its + // guarded delete then removes zero rows. + raceState.armed = true; + yield* Deferred.succeed(releaseSecondWrite, undefined); + const exit = yield* Fiber.await(fiber); + expect(Exit.isFailure(exit)).toBe(true); + + // The successor row AND its credentials survive untouched, and the + // skip was reported, not silent. + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + expect(store.get("connection:org:vercel:main:first")).toBe("c-1"); + expect(store.get("connection:org:vercel:main:second")).toBe("c-2"); + expect(infos.some((line) => line.includes("removed nothing"))).toBe(true); + }), + ), + ); + // A provider write can die with a defect instead of failing. The stranded- // row promise must hold there too: a defect followed by a failed // compensating delete surfaces the same typed StorageError naming the diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 6c504fa927..0308a853bb 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3577,7 +3577,25 @@ export const createExecutor = ("removed"); + // + // Known limitations, accepted deliberately: provider credential + // stores expose no conditional delete, so perfect cleanup under a + // concurrent remove/recreate is impossible at this layer, and no + // further machinery is built for it. + // - Under concurrent remove/recreate, compensation may skip item + // deletion, leaving orphaned credential values at the + // deterministic item ids. Orphans are inert without a row and the + // next same-shaped create overwrites them; orphans are preferred + // over the alternative, clobbering a live successor's secrets. + // - A successor that overwrites one variable, fails before the + // next, and then also fails its own compensating row delete + // leaves a stranded connection that can resolve one stale + // predecessor value. Closing this needs provider-side conditional + // deletes, which do not exist; the stranded state is surfaced + // loudly as the typed StorageError below, naming the connection. + const rowOutcomeRef = yield* Ref.make<"removed" | "superseded" | "overtaken" | "failed">( + "removed", + ); const logContext = { owner: input.owner, integration: String(input.integration), @@ -3605,6 +3623,28 @@ export const createExecutor = Date: Fri, 28 Aug 2026 14:52:47 -0700 Subject: [PATCH 12/13] Report an unconfirmed compensating delete honestly on non-transactional adapters --- packages/core/sdk/src/connections.test.ts | 125 ++++++++++++++++++++++ packages/core/sdk/src/executor.ts | 52 +++++++-- 2 files changed, 170 insertions(+), 7 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 93ae55f0cd..510a617540 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -696,6 +696,54 @@ const staleCompensationRead = (db: FumaDb, state: { armed: boolean }): FumaDb => return wrap(db); }; +/** Wrap a test `FumaDb` so the confirmation read AFTER the guarded delete + * fails at the driver. While armed, the first `connection` read that follows + * a `connection` delete rejects; every other statement passes through. + * Transactions hand out wrapped handles too, so the read inside the + * compensation transaction is covered. */ +const failableConfirmationRead = (db: FumaDb, state: { armed: boolean }): FumaDb => { + let deleteSeen = false; + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "deleteMany") { + return (table: unknown, query: unknown) => { + if (state.armed && table === "connection") deleteSeen = true; + return (target.deleteMany as (t: unknown, q: unknown) => Promise)( + table, + query, + ); + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => { + if (state.armed && deleteSeen && table === "connection") { + state.armed = false; + deleteSeen = false; + // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + return Promise.reject( + new StorageError({ message: "confirmation read refused", cause: undefined }), + ); + } + return (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + }; + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + describe("connections.create credential-write compensation", () => { // Interruption is not an error: error-channel compensation never sees it. A // create interrupted mid-write must still tear down what it already did — @@ -1044,6 +1092,83 @@ describe("connections.create credential-write compensation", () => { ), ); + // The confirmation read after the guarded delete can itself fail. On an + // interactive adapter that failure rolls the delete back with the + // transaction, so "stranded" would be truthful — but on an auto-commit + // adapter (Cloudflare D1 runs `interactiveTransactions: false`) every + // statement commits immediately: the delete has already removed the row + // when the read fails, and a stranded-row claim would be false. The row + // state is genuinely unknown at this layer, so the create must say exactly + // that — skip ALL credential-item deletion and report the row as + // unconfirmed, never as stranded. + it.effect("a failed confirmation read skips item cleanup and reports the row unconfirmed", () => + Effect.gen(function* () { + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const readState = { armed: false }; + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConfirmationRead(config.db, readState), + }); + yield* executor.durable.seed(); + readState.armed = true; + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + // Non-vacuous: the armed rejection fired, so the statement that failed + // really was the confirmation read, after the guarded delete ran. + expect(readState.armed).toBe(false); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + // The error names the connection and reports the unconfirmed state; it + // must NOT claim the row is stranded — on an auto-commit adapter the + // delete already committed and the row is gone. + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("could not be confirmed"); + expect(failure.message).not.toContain("stranded"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // The unknown outcome skips ALL credential-item deletion: the item that + // landed before the failed write is untouched. + expect(store.get("connection:org:vercel:main:first")).toBe("1"); + // The log reports the unconfirmed state, not a stranded-row claim. + expect(errors.some((line) => line.includes("could not confirm"))).toBe(true); + expect(errors.every((line) => !line.includes("stranded a connection row"))).toBe(true); + }), + ); + // A provider write can die with a defect instead of failing. The stranded- // row promise must hold there too: a defect followed by a failed // compensating delete surfaces the same typed StorageError naming the diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 09397502b6..b702db8f99 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3710,15 +3710,31 @@ export const createExecutor = ( - "removed", - ); + // - On a non-transactional adapter (statements auto-commit, no + // rollback — Cloudflare D1) the guarded delete may already have + // committed when the confirmation read fails; the items are left + // in place as inert orphans. + const rowOutcomeRef = yield* Ref.make< + "removed" | "superseded" | "overtaken" | "failed" | "unknown" + >("removed"); const logContext = { owner: input.owner, integration: String(input.integration), connection: String(name), }; const compensate = Effect.gen(function* () { + // Progress marker for the transaction below. It distinguishes + // "the guarded delete itself failed" (nothing was deleted; a + // surviving row is truthfully stranded) from "the delete ran and + // the confirmation read failed". Deliberately a plain mutable + // outside the transaction: a rollback cannot un-set it, which is + // the point — it records statement execution, not committed + // state. On an interactive adapter a post-delete failure rolls + // the delete back; on an auto-commit adapter (D1) the delete has + // already committed. This layer cannot tell which world it is + // in, so a post-delete failure is reported as "unknown", never + // as a stranded row. + let rowDeleteRan = false; const rowOutcome = yield* transaction( Effect.gen(function* () { const current = yield* findConnectionRow({ @@ -3740,6 +3756,7 @@ export const createExecutor = Date: Fri, 28 Aug 2026 15:12:23 -0700 Subject: [PATCH 13/13] Classify a rejection during the guarded delete as unconfirmed, not stranded --- packages/core/sdk/src/connections.test.ts | 171 ++++++++++++++++++++-- packages/core/sdk/src/executor.ts | 57 +++++--- 2 files changed, 189 insertions(+), 39 deletions(-) diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 510a617540..eb8a52ba88 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -617,8 +617,11 @@ const durabilityPlugin = (provider: CredentialProvider) => /** Wrap a test `FumaDb` so deletes on the `connection` table can be made to * fail on demand — the raw driver-level failure the compensating delete must - * survive loudly. Transactions hand out wrapped handles too, so the guarded - * delete inside the compensation transaction is covered. */ + * survive loudly. A rejection during the delete statement is ambiguous on an + * auto-commit adapter (the statement may have executed first), so this + * failure is reported as unconfirmed, never as definitively stranded. + * Transactions hand out wrapped handles too, so the guarded delete inside + * the compensation transaction is covered. */ const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): FumaDb => { const wrap = (inner: FumaDb): FumaDb => new Proxy(inner, { @@ -646,6 +649,56 @@ const failableConnectionDeletes = (db: FumaDb, shouldFail: () => boolean): FumaD return wrap(db); }; +/** Wrap a test `FumaDb` so compensation fails strictly BEFORE its guarded + * delete statement is issued: the identity read that opens the compensation + * transaction rejects. Only a pre-attempt failure keeps the definitive + * stranded-row claim truthful — a rejection during the delete statement + * itself is reported as unconfirmed instead (see + * `failableConnectionDeletes`). The read is identified by sequence: the + * first `connection` read after this create's row insert. The conflict-check + * read runs before the insert and no other `connection` read happens in + * between, so that read is compensation's. Transactions hand out wrapped + * handles too. */ +const failableCompensationRowDelete = (db: FumaDb, shouldFail: () => boolean): FumaDb => { + let insertSeen = false; + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return (run: (tx: FumaDb) => Promise) => + (target.transaction as (r: (tx: FumaDb) => Promise) => Promise)( + (tx) => run(wrap(tx)), + ); + } + if (prop === "create") { + return async (table: unknown, values: unknown) => { + const row = await (target.create as (t: unknown, v: unknown) => Promise)( + table, + values, + ); + if (table === "connection") insertSeen = true; + return row; + }; + } + if (prop === "findFirst") { + return (table: unknown, query: unknown) => + shouldFail() && insertSeen && table === "connection" + ? // oxlint-disable-next-line executor/no-promise-reject -- boundary: the proxy fakes a driver-level rejection from the raw FumaDb handle + Promise.reject( + new StorageError({ message: "identity read refused", cause: undefined }), + ) + : (target.findFirst as (t: unknown, q: unknown) => Promise)(table, query); + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + /** Wrap a test `FumaDb` so one armed `connection` read observes a stale row. * This models the read/delete race inside the compensation transaction: under * read-committed isolation the pre-delete identity read can see this create's @@ -866,10 +919,13 @@ describe("connections.create credential-write compensation", () => { }), ); - // The compensating delete can itself fail. Swallowing that failure strands - // a visible credential-less row behind an error that never mentions it. The - // create must fail with an error that NAMES the stranded connection so an - // operator can act on it. + // The compensating delete can fail before its guarded delete statement is + // even issued (here: the identity read that opens the compensation + // transaction rejects). Nothing can have been deleted, so the surviving + // credential-less row is definitively stranded — and swallowing that + // failure hides it behind an error that never mentions it. The create must + // fail with an error that NAMES the stranded connection so an operator can + // act on it. it.effect("names the stranded connection when the compensating delete fails", () => Effect.gen(function* () { let failRowDelete = false; @@ -883,7 +939,7 @@ describe("connections.create credential-write compensation", () => { const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); const executor = yield* createExecutor({ ...config, - db: failableConnectionDeletes(config.db, () => failRowDelete), + db: failableCompensationRowDelete(config.db, () => failRowDelete), }); yield* executor.durable.seed(); failRowDelete = true; @@ -905,6 +961,8 @@ describe("connections.create credential-write compensation", () => { if (!Predicate.isTagged("StorageError")(failure)) return; expect(failure.message).toContain("main"); expect(failure.message).toContain("vercel"); + // Pre-attempt failure: the stranded claim is definitive and stated. + expect(failure.message).toContain("stranded"); // The original write failure is retained as the cause, not replaced. const isStorageError = (u: unknown): u is StorageError => Predicate.isTagged("StorageError")(u); @@ -1169,10 +1227,89 @@ describe("connections.create credential-write compensation", () => { }), ); + // The guarded delete STATEMENT can itself reject. On an interactive adapter + // the rejection rolls the transaction back and the row survives — but on an + // auto-commit adapter (Cloudflare D1) the statement may have executed + // before the rejection surfaced, so a definitive stranded-row claim would + // be false. Once the delete has been attempted, the row state is genuinely + // unknown at this layer: the create must skip ALL credential-item deletion + // and report the delete as unconfirmed, never as stranded. + it.effect("a rejected delete statement skips item cleanup and reports the row unconfirmed", () => + Effect.gen(function* () { + let failRowDelete = false; + const store = new Map(); + const provider = trackingProvider(store, { + set: (id, value) => + String(id).endsWith(":second") + ? Effect.fail(new StorageError({ message: "provider write refused", cause: undefined })) + : Effect.sync(() => void store.set(String(id), value)), + }); + const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); + const executor = yield* createExecutor({ + ...config, + db: failableConnectionDeletes(config.db, () => failRowDelete), + }); + yield* executor.durable.seed(); + failRowDelete = true; + + const errors: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Error") { + errors.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const result = yield* Effect.result( + executor.connections + .create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + values: { first: "1", second: "2" }, + }) + .pipe(Effect.provide(Logger.layer([capture]))), + ); + + expect(Result.isFailure(result)).toBe(true); + if (!Result.isFailure(result)) return; + const failure = result.failure; + expect(Predicate.isTagged("StorageError")(failure)).toBe(true); + if (!Predicate.isTagged("StorageError")(failure)) return; + // The error names the connection and reports the unconfirmed state; it + // must NOT claim the row is stranded — on an auto-commit adapter the + // delete may already have executed before the rejection surfaced. + expect(failure.message).toContain("main"); + expect(failure.message).toContain("vercel"); + expect(failure.message).toContain("could not be confirmed"); + expect(failure.message).not.toContain("stranded"); + // The original write failure is retained as the cause, not replaced. + const isStorageError = (u: unknown): u is StorageError => + Predicate.isTagged("StorageError")(u); + expect(isStorageError(failure.cause)).toBe(true); + if (!isStorageError(failure.cause)) return; + expect(failure.cause.message).toBe("provider write refused"); + + // The unknown outcome skips ALL credential-item deletion: the item that + // landed before the failed write is untouched. + expect(store.get("connection:org:vercel:main:first")).toBe("1"); + // The log reports the unconfirmed state, not a stranded-row claim. + expect(errors.some((line) => line.includes("could not confirm"))).toBe(true); + expect(errors.every((line) => !line.includes("stranded a connection row"))).toBe(true); + + // Non-vacuous: the armed proxy rejected the delete statement without + // running it, so on this interactive adapter the row survived. + failRowDelete = false; + const rows = yield* executor.connections.list(); + expect(rows.length).toBe(1); + expect(String(rows[0]?.name)).toBe("main"); + }), + ); + // A provider write can die with a defect instead of failing. The stranded- - // row promise must hold there too: a defect followed by a failed - // compensating delete surfaces the same typed StorageError naming the - // stranded connection, not an anonymous crash. + // row promise must hold there too: a defect followed by a compensating + // delete that fails before its delete statement is issued surfaces the + // same typed StorageError naming the stranded connection, not an anonymous + // crash. it.effect("a defect followed by a failed row delete still names the stranded connection", () => Effect.gen(function* () { let failRowDelete = false; @@ -1186,7 +1323,7 @@ describe("connections.create credential-write compensation", () => { const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); const executor = yield* createExecutor({ ...config, - db: failableConnectionDeletes(config.db, () => failRowDelete), + db: failableCompensationRowDelete(config.db, () => failRowDelete), }); yield* executor.durable.seed(); failRowDelete = true; @@ -1225,10 +1362,12 @@ describe("connections.create credential-write compensation", () => { ); // Interruption cannot carry a typed error — interrupting wins over failing - // — so when an interrupted create cannot delete its row, the stranded row - // is reported through a loud error log and the create stays an - // interruption. The items that landed before the interrupt stay with the - // stranded row: credential teardown is gated on the row delete succeeding. + // — so when an interrupted create cannot delete its row (compensation + // fails before the delete statement is issued, leaving the row + // definitively stranded), the stranded row is reported through a loud + // error log and the create stays an interruption. The items that landed + // before the interrupt stay with the stranded row: credential teardown is + // gated on the row delete succeeding. it.effect("an interrupted create with a failed row delete logs the stranded row", () => Effect.scoped( Effect.gen(function* () { @@ -1244,7 +1383,7 @@ describe("connections.create credential-write compensation", () => { const config = makeTestConfig({ plugins: [durabilityPlugin(provider)] as const }); const executor = yield* createExecutor({ ...config, - db: failableConnectionDeletes(config.db, () => failRowDelete), + db: failableCompensationRowDelete(config.db, () => failRowDelete), }); yield* executor.durable.seed(); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index b702db8f99..d8afd12d23 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3712,8 +3712,9 @@ export const createExecutor = ("removed"); @@ -3724,17 +3725,21 @@ export const createExecutor = b.and( @@ -3756,7 +3765,6 @@ export const createExecutor =