From c576efa8d0b54e870f86561fc797d7d12d25c969 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 13:07:13 -0700 Subject: [PATCH 1/2] Back off vault version-check retries and persist the rotated refresh token first --- e2e/cloud/credential-write-durability.test.ts | 447 ++++++++++++++++++ packages/core/sdk/src/executor.ts | 19 +- .../workos-vault/src/sdk/secret-store.test.ts | 4 +- .../workos-vault/src/sdk/secret-store.ts | 105 +++- 4 files changed, 552 insertions(+), 23 deletions(-) create mode 100644 e2e/cloud/credential-write-durability.test.ts diff --git a/e2e/cloud/credential-write-durability.test.ts b/e2e/cloud/credential-write-durability.test.ts new file mode 100644 index 000000000..6569ed78a --- /dev/null +++ b/e2e/cloud/credential-write-durability.test.ts @@ -0,0 +1,447 @@ +// Cloud: a refreshed OAuth credential has to be PERSISTED, and persisting it is +// a pair of version-checked writes into WorkOS Vault — the rotated refresh +// token and the new access token, one after the other, not atomically. Two +// production failures live in that gap. +// +// 1. Contention. Two concurrent probes of one connection each run a refresh and +// each write the same two objects, so a version-checked write comes back 409 +// ("Current version does not match expected version"). A write that retries +// with no wait re-collides inside the peer's own round trip, drains its +// attempts in microseconds, and the refresh fails outright — the user saw a +// 500 on a connection health request. +// 2. Ordering. If the write that fails is the one carrying the rotated refresh +// token, the credential is gone for good: minting already spent the refresh +// token we sent, so nothing can mint again and every later use of the +// connection comes back `invalid_grant`. The access token, by contrast, is +// disposable — one more grant re-mints it. +// +// Both are pinned here at the product surface, black box. Failures are armed on +// the WorkOS emulator that the product's own WorkOS client talks to; no product +// code, stubs, or internals are touched. +import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { connectEmulator, type EmulatorClient } from "@executor-js/emulate"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, +} from "@executor-js/sdk/shared"; +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; +import { WORKOS_EMULATOR_PORT } from "../targets/cloud"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +const VAULT_CONFLICT_RESPONSE = { + status: 409, + body: { code: "conflict", message: "Current version does not match expected version" }, +} as const; + +type UpstreamHandle = { + readonly url: string; + /** Stop honouring the bearer(s) seen so far, exactly as a revocation would — + * the deterministic way to make the very next call run a refresh. */ + readonly revokeSeenBearers: () => void; + readonly close: () => void; +}; + +/** Upstream on 127.0.0.1 that authenticates for real: `GET /issues` is 200 for + * any bearer it has not been told to reject, and 401 for one it has. */ +const serveUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + const seen: string[] = []; + const revoked = new Set(); + const server = createServer((request, response) => { + const bearer = (request.headers.authorization ?? "").replace(/^Bearer\s+/i, ""); + if (request.method === "GET" && (request.url ?? "").startsWith("/issues")) { + seen.push(bearer); + if (revoked.has(bearer)) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "invalid_token" })); + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ issues: [{ id: 1, title: "first" }] })); + 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}`, + revokeSeenBearers: () => { + for (const bearer of seen) revoked.add(bearer); + }, + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const spec = ( + baseUrl: string, + oauth: { readonly authorizationEndpoint: string; readonly tokenEndpoint: string }, +): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Issues API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/issues": { + get: { + operationId: "listIssues", + summary: "List issues", + security: [{ oauth: ["issues.read"] }], + responses: { "200": { description: "issues" } }, + }, + }, + }, + components: { + securitySchemes: { + oauth: { + type: "oauth2", + flows: { + authorizationCode: { + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: { "issues.read": "Read issues" }, + }, + }, + }, + }, + }, + }); + +const invokeByAddressCode = (address: string, args: unknown) => ` +const segments = ${JSON.stringify(address)}.split(".").slice(1); +let node = tools; +for (const segment of segments) node = node[segment]; +const result = await node(${JSON.stringify(args)}); +return JSON.stringify(result); +`; + +/** One vault object as the emulated WorkOS itself sees it. Read from the + * emulator control plane — upstream state, never the product's own. */ +type VaultObject = { readonly id: string; readonly name: string }; + +/** Read a vault row out of the emulator's generic store snapshot, or null when + * it is not shaped like one. */ +const readVaultObject = (row: Record): VaultObject | null => + typeof row.workos_id === "string" && typeof row.name === "string" + ? { id: row.workos_id, name: row.name } + : null; + +/** Every vault object this connection owns. The object name embeds the logical + * item id, so the connection's own objects are the ones carrying its slug. */ +const vaultObjectsFor = (workos: EmulatorClient, slug: string): Effect.Effect => + Effect.promise(async () => { + const snapshot = await workos.state(); + const items = snapshot.collections["workos.vault_objects"]?.items ?? []; + return items + .map((item) => readVaultObject({ ...item })) + .filter((object): object is VaultObject => object !== null && object.name.includes(slug)); + }); + +type CallResult = { readonly ok: boolean; readonly text: string }; + +type ToolEnvelope = { + readonly ok: boolean; + readonly data?: unknown; + readonly error?: { + readonly code?: string; + readonly message?: string; + readonly status?: number; + }; +}; + +/** An OAuth-backed integration, connected and proven working, plus the handles + * needed to break its credential writes: the upstream that can revoke a bearer + * (the deterministic refresh trigger) and the WorkOS emulator that serves the + * vault. Everything it creates is torn down by scope finalizers. */ +const connectIntegration = Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream(); + // Long-lived tokens: the proactive expiry check must never fire, so the + // upstream's 401 is the only thing that triggers the refresh whose + // persistence is under test. + const oauth = yield* serveOAuthTestServer({ + scopes: ["issues.read"], + tokenExpiresInSeconds: 3600, + }); + const workos = yield* Effect.promise(() => + connectEmulator({ baseUrl: `http://127.0.0.1:${WORKOS_EMULATOR_PORT}` }), + ); + const slug = unique("credwrite"); + const clientSlug = OAuthClientSlug.make(unique("credwriteclient")); + + yield* Effect.addFinalizer(() => Effect.promise(() => workos.faults.clear()).pipe(Effect.ignore)); + + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: spec(upstream.url, oauth) }, + slug, + baseUrl: upstream.url, + authenticationTemplate: [ + { + slug: "oauth", + kind: "oauth2", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: ["issues.read"], + }, + ], + }, + }); + yield* Effect.addFinalizer(() => + client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore), + ); + + yield* client.oauth.createClient({ + payload: { + owner: "org", + slug: clientSlug, + grant: "authorization_code", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + clientId: "test-client", + clientSecret: "test-secret", + originIntegration: IntegrationSlug.make(slug), + }, + }); + yield* Effect.addFinalizer(() => + client.oauth + .removeClient({ params: { slug: clientSlug }, payload: { owner: "org" } }) + .pipe(Effect.ignore), + ); + + const started = yield* client.oauth.start({ + payload: { + client: clientSlug, + clientOwner: "org", + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make(slug), + template: AuthTemplateSlug.make("oauth"), + }, + }); + expect(started.status, "oauth.start redirects to the authorization server").toBe("redirect"); + if (started.status !== "redirect") return yield* Effect.die("no redirect"); + + // Drive the test IdP's consent by hand (authorize -> login -> code). + const code = yield* Effect.promise(async () => { + const authorize = await fetch(started.authorizationUrl, { redirect: "manual" }); + const loginUrl = authorize.headers.get("location"); + if (!loginUrl) throw new Error(`authorize did not redirect: ${authorize.status}`); + const login = await fetch(loginUrl, { + method: "POST", + headers: { + authorization: `Basic ${Buffer.from("alice:password").toString("base64")}`, + }, + redirect: "manual", + }); + const callbackUrl = login.headers.get("location"); + if (!callbackUrl) throw new Error(`login did not redirect: ${login.status}`); + const minted = new URL(callbackUrl).searchParams.get("code"); + if (!minted) throw new Error("callback carried no authorization code"); + return minted; + }); + yield* client.oauth.complete({ payload: { state: started.state, code } }); + yield* Effect.addFinalizer(() => + client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("main"), + }, + }) + .pipe(Effect.ignore), + ); + + const tools = yield* client.tools.list({ query: {} }); + const address = tools + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((addr) => addr.endsWith("listIssues")); + expect(address, "the listIssues tool is in the catalog").toBeDefined(); + + const session = mcp.session(identity); + /** Invoke the connected tool and report what came back, success or not. */ + const attempt = (): Effect.Effect => + Effect.gen(function* () { + let called = yield* session.call("execute", { + code: invokeByAddressCode(address!, {}), + }); + let guard = 0; + while (called.text.includes("executionId:") && guard < 10) { + called = yield* session.approvePaused(called.text); + guard += 1; + } + return { ok: called.ok, text: called.text }; + }); + + /** Invoke the tool and require it to have worked end to end. */ + const call = (label: string): Effect.Effect => + Effect.gen(function* () { + const result = yield* attempt(); + expect( + result.ok, + `the ${label} call reached the upstream and came back (got: ${result.text.slice(0, 400)})`, + ).toBe(true); + const envelope = JSON.parse(result.text) as ToolEnvelope; + expect(envelope.ok, `the ${label} call succeeded`).toBe(true); + return envelope; + }); + + return { attempt, call, oauth, slug, upstream, workos } as const; +}); + +/** Is this one of the authorization server's refresh grants? Counting them is + * how a "success" is shown to involve a real re-mint, not a cached token. */ +const isRefreshGrant = (request: { + readonly path: string; + readonly method: string; + readonly body: string; +}) => + request.path === "/token" && + request.method === "POST" && + request.body.includes("grant_type=refresh_token"); + +scenario( + "Credential persistence · contention on the vault write does not lose the refreshed credential", + {}, + Effect.scoped( + Effect.gen(function* () { + const { call, oauth, upstream, workos } = yield* connectIntegration; + + // Baseline: the connection works on its freshly minted token. + yield* call("baseline"); + + // Transient contention: the next few version-checked writes lose their + // race, exactly as a peer writer persisting the same connection's + // credential would make them lose it. Three is as many as a tight, + // wait-free retry loop could ever absorb — a policy that waits between + // attempts outlasts them, one that does not, cannot. + yield* Effect.promise(() => + workos.faults.arm({ + match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, + response: VAULT_CONFLICT_RESPONSE, + times: 3, + }), + ); + // Revoking upstream forces the very next call to refresh, so the contended + // write happens inside a real user-visible request. + upstream.revokeSeenBearers(); + yield* call("transiently-contended-refresh"); + yield* Effect.promise(() => workos.faults.clear()); + + // Sustained contention: every version-checked attempt loses. Dropping a + // credential that has just been minted is the one unrecoverable outcome, + // so the write still has to land. + yield* Effect.promise(() => + workos.faults.arm({ + match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, + response: VAULT_CONFLICT_RESPONSE, + times: 5, + }), + ); + upstream.revokeSeenBearers(); + yield* call("continuously-contended-refresh"); + yield* Effect.promise(() => workos.faults.clear()); + + // The durability half: the credential those contended writes were + // carrying is the rotated, single-use refresh token. If any of it had been + // dropped, the connection could never mint anything again. + upstream.revokeSeenBearers(); + yield* call("post-contention"); + + expect( + (yield* oauth.requests).filter(isRefreshGrant).length, + "each revocation was absorbed by a real refresh grant", + ).toBeGreaterThanOrEqual(3); + }), + ), +); + +scenario( + "Credential persistence · a write that fails mid-refresh leaves the connection able to refresh", + {}, + Effect.scoped( + Effect.gen(function* () { + const { attempt, call, oauth, slug, upstream, workos } = yield* connectIntegration; + + yield* call("baseline"); + + // A refresh persists two objects. Break the ACCESS TOKEN's object only, + // once: whichever write runs first survives and whichever runs second is + // lost, so failing this one is the direct question "which of the two does + // the product persist first?". + const objects = yield* vaultObjectsFor(workos, slug); + const refreshObject = objects.find((object) => object.name.endsWith("refresh")); + expect(refreshObject, "the connection stored a refresh token in the vault").toBeDefined(); + const accessObject = objects.find( + (object) => object !== refreshObject && refreshObject!.name.startsWith(object.name), + ); + expect(accessObject, "the connection stored an access token in the vault").toBeDefined(); + + yield* Effect.promise(() => + workos.faults.arm({ + match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, + response: { status: 503, body: { code: "unavailable", message: "vault unavailable" } }, + times: 1, + }), + ); + + upstream.revokeSeenBearers(); + const interrupted = yield* attempt(); + expect( + interrupted.ok, + `the call whose credential write was interrupted reports the failure (got: ${interrupted.text.slice(0, 200)})`, + ).toBe(false); + yield* Effect.promise(() => workos.faults.clear()); + + // The interrupted refresh still spent the refresh token it sent: the + // authorization server rotated it and will not honour the old one again. + // The access token in the vault is the stale, revoked one, so the very + // next call has to refresh — and can only do so if the ROTATED refresh + // token is what survived the half-finished write. + const recovered = yield* attempt(); + expect( + recovered.ok, + `the connection still refreshes after the interrupted write (got: ${recovered.text.slice(0, 400)})`, + ).toBe(true); + expect((JSON.parse(recovered.text) as ToolEnvelope).ok, "the recovered call succeeded").toBe( + true, + ); + + expect( + (yield* oauth.requests).filter(isRefreshGrant).length, + "the interrupted refresh and the recovery were both real refresh grants", + ).toBeGreaterThanOrEqual(2); + }), + ), +); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 2ce9a3922..a877946b6 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1860,11 +1860,20 @@ export const createExecutor = { }), ); - it.effect("retries value writes on 409 version conflicts", () => + // `it.live` (real clock): a conflicted write now waits before retrying, and + // the TestClock never advances that wait on its own. + it.live("retries value writes on 409 version conflicts", () => Effect.gen(function* () { const provider = makeProvider(makeFakeClient({ conflictOnNextSecretUpdate: true })); diff --git a/packages/plugins/workos-vault/src/sdk/secret-store.ts b/packages/plugins/workos-vault/src/sdk/secret-store.ts index 97341056e..4d8c6e6b6 100644 --- a/packages/plugins/workos-vault/src/sdk/secret-store.ts +++ b/packages/plugins/workos-vault/src/sdk/secret-store.ts @@ -22,7 +22,14 @@ import { export const WORKOS_VAULT_PROVIDER_KEY = ProviderKey.make("workos-vault"); const DEFAULT_OBJECT_PREFIX = "executor"; -const MAX_WRITE_ATTEMPTS = 3; +// Version-checked write attempts before the last-resort un-versioned write. +// Contention here is a peer writer persisting the SAME credential (two +// concurrent refreshes of one connection), so retries must be spaced: a tight +// loop re-reads and re-writes inside the peer's own round trip and collides +// again deterministically. +const MAX_WRITE_ATTEMPTS = 5; +const WRITE_CONFLICT_BACKOFF_BASE_MS = 50; +const WRITE_CONFLICT_BACKOFF_CAP_MS = 800; // WorkOS creates a per-context KEK just-in-time on first write; a create // call immediately after that provisioning step can race with the KEK // becoming usable and return a transient error whose message ends in @@ -243,34 +250,98 @@ const loadSecretObject = ( }), ); +/** Half-jittered exponential backoff for attempt `n` (1-based): a wait drawn + * from [d/2, d) where d = base * 2^(n-1), capped. The jitter matters more than + * the delay — two writers that back off by the same amount simply collide + * again one beat later. */ +const writeConflictBackoffMillis = (attempt: number): number => { + const ceiling = Math.min( + WRITE_CONFLICT_BACKOFF_BASE_MS * 2 ** (attempt - 1), + WRITE_CONFLICT_BACKOFF_CAP_MS, + ); + return ceiling / 2 + Math.random() * (ceiling / 2); +}; + +/** A version-checked write either landed or lost the race to a peer writer. */ +type WriteOutcome = "written" | "contended"; + +/** Did this update failure mean "someone else wrote first"? 409 is the + * documented answer. WorkOS also answers 400 when the version we checked + * against is several generations behind, and an update carries no name and no + * context — the only inputs are the id, the value, and the version — so a 400 + * here is contention too, not a malformed request. If it really were + * malformed, the un-versioned last-resort write fails the same way and the + * error still surfaces. */ +const isWriteContention = (error: WorkOSVaultClientError): boolean => + isStatusError(error, 409) || isStatusError(error, 400); + const upsertSecretValue = ( client: WorkOSVaultClient, name: string, value: string, context: Record, ): Effect.Effect => { - const attemptWrite = ( - remainingConflictAttempts: number, - remainingKekAttempts: number, - ): Effect.Effect => + const attemptOnce = (): Effect.Effect => Effect.gen(function* () { const existing = yield* loadSecretObject(client, name); if (existing) { - yield* client.updateObject({ - id: existing.id, - value, - versionCheck: existing.metadata.versionId, - }); + return yield* client + .updateObject({ id: existing.id, value, versionCheck: existing.metadata.versionId }) + .pipe( + Effect.as("written"), + Effect.catch((error: WorkOSVaultClientError) => + isWriteContention(error) + ? Effect.succeed("contended") + : Effect.fail(error), + ), + ); + } + + return yield* client.createObject({ name, value, context }).pipe( + Effect.as("written"), + // A peer created the object between our read and our create; the next + // attempt re-reads it and takes the update path. + Effect.catch((error: WorkOSVaultClientError) => + isStatusError(error, 409) + ? Effect.succeed("contended") + : Effect.fail(error), + ), + ); + }); + + /** Last resort once every version-checked attempt lost: write without a + * version check. Both racers are persisting a credential they just minted, + * so last-writer-wins keeps a usable value; failing instead discards a + * freshly minted, often single-use credential, which is the one outcome + * that leaves a connection with nothing it can use. */ + const blindWrite = (): Effect.Effect => + Effect.gen(function* () { + const existing = yield* loadSecretObject(client, name); + if (!existing) { + yield* client.createObject({ name, value, context }); return; } + console.warn( + `[workos-vault] version-checked write for object=${name} lost ` + + `${MAX_WRITE_ATTEMPTS} races; writing without a version check`, + ); + yield* client.updateObject({ id: existing.id, value }); + }); - yield* client.createObject({ name, value, context }); - }).pipe( + const attemptWrite = ( + attempt: number, + remainingKekAttempts: number, + ): Effect.Effect => + attemptOnce().pipe( + Effect.flatMap((outcome: WriteOutcome) => { + if (outcome === "written") return Effect.void; + if (attempt >= MAX_WRITE_ATTEMPTS) return blindWrite(); + return Effect.sleep(writeConflictBackoffMillis(attempt)).pipe( + Effect.flatMap(() => attemptWrite(attempt + 1, remainingKekAttempts)), + ); + }), Effect.catch((error: WorkOSVaultClientError) => { - if (remainingConflictAttempts > 1 && isStatusError(error, 409)) { - return attemptWrite(remainingConflictAttempts - 1, remainingKekAttempts); - } if (remainingKekAttempts > 1 && isKekNotReadyError(error)) { console.warn( `[workos-vault] KEK not ready for object=${name} — ` + @@ -278,7 +349,7 @@ const upsertSecretValue = ( `(${MAX_KEK_NOT_READY_ATTEMPTS - remainingKekAttempts + 1}/${MAX_KEK_NOT_READY_ATTEMPTS})`, ); return Effect.sleep(KEK_NOT_READY_BACKOFF_MS).pipe( - Effect.flatMap(() => attemptWrite(remainingConflictAttempts, remainingKekAttempts - 1)), + Effect.flatMap(() => attemptWrite(attempt, remainingKekAttempts - 1)), ); } if (isKekNotReadyError(error)) { @@ -291,7 +362,7 @@ const upsertSecretValue = ( }), ); - return attemptWrite(MAX_WRITE_ATTEMPTS, MAX_KEK_NOT_READY_ATTEMPTS); + return attemptWrite(1, MAX_KEK_NOT_READY_ATTEMPTS); }; const deleteSecretValue = ( From 5a9ae3d6c30fbdbc3bd9f8d1c121ef93ede6cada Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 14:45:39 -0700 Subject: [PATCH 2/2] Prove the vault write contention is real and time-bounded in e2e --- e2e/cloud/credential-write-durability.test.ts | 191 ++++++++++++++---- 1 file changed, 157 insertions(+), 34 deletions(-) diff --git a/e2e/cloud/credential-write-durability.test.ts b/e2e/cloud/credential-write-durability.test.ts index 6569ed78a..589b050db 100644 --- a/e2e/cloud/credential-write-durability.test.ts +++ b/e2e/cloud/credential-write-durability.test.ts @@ -17,14 +17,26 @@ // // Both are pinned here at the product surface, black box. Failures are armed on // the WorkOS emulator that the product's own WorkOS client talks to; no product -// code, stubs, or internals are touched. +// code, stubs, or internals are touched. Contention is modelled twice, because +// the two halves of the write policy fail differently: by a COUNT of collisions +// (does the loop have enough attempts, and does it still land the value when it +// runs out?) and by a WINDOW of time (does it wait long enough between attempts +// to still be trying when the peer lets go?). Every scenario also reads the +// emulator's ledger back to prove the collisions it armed were really served — +// a fault whose pattern stopped matching would otherwise leave a test that +// passes without ever contending. import { randomBytes } from "node:crypto"; import { createServer } from "node:http"; import { expect } from "@effect/vitest"; import { Effect } from "effect"; import { composePluginApi } from "@executor-js/api/server"; -import { connectEmulator, type EmulatorClient } from "@executor-js/emulate"; +import { + type ArmedFault, + connectEmulator, + type EmulatorClient, + type FaultArmInput, +} from "@executor-js/emulate"; import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; import { AuthTemplateSlug, @@ -47,6 +59,9 @@ const VAULT_CONFLICT_RESPONSE = { body: { code: "conflict", message: "Current version does not match expected version" }, } as const; +/** Every PUT the vault serves for a stored object — the version-checked write. */ +const VAULT_WRITE = { method: "PUT", pathPattern: "/vault/v1/kv/*" } as const; + type UpstreamHandle = { readonly url: string; /** Stop honouring the bearer(s) seen so far, exactly as a revocation would — @@ -162,6 +177,57 @@ const vaultObjectsFor = (workos: EmulatorClient, slug: string): Effect.Effect object !== null && object.name.includes(slug)); }); +/** Arm a fault on the shared emulator and take it back down with the scope, by + * id — a blanket `faults.clear()` would also disarm a neighbouring scenario's. */ +const armFault = (workos: EmulatorClient, input: FaultArmInput) => + Effect.acquireRelease( + Effect.promise(() => workos.faults.arm(input)), + (armed) => Effect.promise(() => workos.faults.clear(armed.id)).pipe(Effect.ignore), + ); + +/** How many responses this armed fault actually injected, read back from the + * emulator's ledger. This is the proof that the failure under test HAPPENED: + * an armed fault whose pattern never matched leaves the product's writes + * untouched, and a scenario asserting only "the call worked" would pass + * without ever exercising the retry it claims to cover. */ +const faultsServed = (workos: EmulatorClient, armed: ArmedFault): Effect.Effect => + Effect.promise(async () => { + const entries = await workos.ledger.list(200); + return entries.filter((entry) => entry.faultId === armed.id).length; + }); + +/** Keep the vault's version-checked writes losing for a WINDOW of time that + * starts at the first collision the product actually suffers — a peer writer + * that holds the object for a while, rather than a fixed number of collisions. + * This is the shape of the production failure: attempts spaced by a wait + * outlast the peer's round trip; attempts fired back to back all land inside + * it, drain, and the refresh dies. + * + * Anchoring on the first collision (not on arming) is what makes it + * deterministic — a refresh only begins several round trips after the fault is + * armed, so a window measured from arming would already be over. */ +const holdWritesInConflict = (workos: EmulatorClient, windowMillis: number) => + Effect.gen(function* () { + // Far more conflicts than any bounded retry policy can consume, so the + // window — not a counter — is what ends the contention. + const armed = yield* armFault(workos, { + match: VAULT_WRITE, + response: VAULT_CONFLICT_RESPONSE, + times: 64, + }); + const released = (async () => { + const deadline = Date.now() + 30_000; + while (Date.now() < deadline) { + const live = (await workos.faults.list()).find((fault) => fault.id === armed.id); + if (!live || live.remaining < armed.times) break; + await new Promise((resolve) => setTimeout(resolve, 15)); + } + await new Promise((resolve) => setTimeout(resolve, windowMillis)); + await workos.faults.clear(armed.id); + })().catch(() => undefined); + return { armed, released }; + }); + type CallResult = { readonly ok: boolean; readonly text: string }; type ToolEnvelope = { @@ -198,8 +264,6 @@ const connectIntegration = Effect.gen(function* () { const slug = unique("credwrite"); const clientSlug = OAuthClientSlug.make(unique("credwriteclient")); - yield* Effect.addFinalizer(() => Effect.promise(() => workos.faults.clear()).pipe(Effect.ignore)); - yield* client.openapi.addSpec({ payload: { spec: { kind: "blob", value: spec(upstream.url, oauth) }, @@ -343,35 +407,43 @@ scenario( // Transient contention: the next few version-checked writes lose their // race, exactly as a peer writer persisting the same connection's - // credential would make them lose it. Three is as many as a tight, - // wait-free retry loop could ever absorb — a policy that waits between - // attempts outlasts them, one that does not, cannot. - yield* Effect.promise(() => - workos.faults.arm({ - match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, - response: VAULT_CONFLICT_RESPONSE, - times: 3, + // credential would make them lose it. Three is as many as the pre-fix + // policy could ever absorb. + const transient = yield* Effect.scoped( + Effect.gen(function* () { + const armed = yield* armFault(workos, { + match: VAULT_WRITE, + response: VAULT_CONFLICT_RESPONSE, + times: 3, + }); + // Revoking upstream forces the very next call to refresh, so the + // contended write happens inside a real user-visible request. + upstream.revokeSeenBearers(); + yield* call("transiently-contended-refresh"); + return yield* faultsServed(workos, armed); }), ); - // Revoking upstream forces the very next call to refresh, so the contended - // write happens inside a real user-visible request. - upstream.revokeSeenBearers(); - yield* call("transiently-contended-refresh"); - yield* Effect.promise(() => workos.faults.clear()); + expect(transient, "the transiently-contended refresh really did collide three times").toBe(3); // Sustained contention: every version-checked attempt loses. Dropping a // credential that has just been minted is the one unrecoverable outcome, // so the write still has to land. - yield* Effect.promise(() => - workos.faults.arm({ - match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, - response: VAULT_CONFLICT_RESPONSE, - times: 5, + const sustained = yield* Effect.scoped( + Effect.gen(function* () { + const armed = yield* armFault(workos, { + match: VAULT_WRITE, + response: VAULT_CONFLICT_RESPONSE, + times: 5, + }); + upstream.revokeSeenBearers(); + yield* call("continuously-contended-refresh"); + return yield* faultsServed(workos, armed); }), ); - upstream.revokeSeenBearers(); - yield* call("continuously-contended-refresh"); - yield* Effect.promise(() => workos.faults.clear()); + expect( + sustained, + "the continuously-contended refresh exhausted every version-checked attempt", + ).toBe(5); // The durability half: the credential those contended writes were // carrying is the rotated, single-use refresh token. If any of it had been @@ -387,6 +459,48 @@ scenario( ), ); +// The peer writer's round trip, in milliseconds. Long enough that a retry loop +// firing its attempts back to back drains all of them (and its last-resort +// write) inside the window; short enough that a loop waiting between attempts is +// still trying when the window closes — the shipped policy's fourth and fifth +// attempts fall no earlier than 175ms and 375ms after the first collision. +const CONTENTION_WINDOW_MS = 200; + +scenario( + "Credential persistence · a refresh outlasts contention that lasts longer than a round trip", + {}, + Effect.scoped( + Effect.gen(function* () { + const { call, oauth, upstream, workos } = yield* connectIntegration; + + yield* call("baseline"); + + // Contention bounded by TIME rather than by a count of collisions: this is + // the production shape, where the peer holds the object for as long as its + // own write takes and the loser has to still be trying when it lets go. + const { armed, released } = yield* holdWritesInConflict(workos, CONTENTION_WINDOW_MS); + upstream.revokeSeenBearers(); + yield* call("refresh-under-a-contention-window"); + yield* Effect.promise(() => released); + + expect( + yield* faultsServed(workos, armed), + "the refresh really was fighting a contended vault write", + ).toBeGreaterThanOrEqual(3); + + // And the credential that survived the window is usable: the next + // revocation is absorbed by another real refresh. + upstream.revokeSeenBearers(); + yield* call("post-window"); + + expect( + (yield* oauth.requests).filter(isRefreshGrant).length, + "both revocations were absorbed by real refresh grants", + ).toBeGreaterThanOrEqual(2); + }), + ), +); + scenario( "Credential persistence · a write that fails mid-refresh leaves the connection able to refresh", {}, @@ -408,21 +522,30 @@ scenario( ); expect(accessObject, "the connection stored an access token in the vault").toBeDefined(); - yield* Effect.promise(() => - workos.faults.arm({ - match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, - response: { status: 503, body: { code: "unavailable", message: "vault unavailable" } }, - times: 1, + const interrupted = yield* Effect.scoped( + Effect.gen(function* () { + const armed = yield* armFault(workos, { + match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, + response: { + status: 503, + body: { code: "unavailable", message: "vault unavailable" }, + }, + times: 1, + }); + + upstream.revokeSeenBearers(); + const result = yield* attempt(); + expect( + yield* faultsServed(workos, armed), + "the access token's write is the one that broke", + ).toBe(1); + return result; }), ); - - upstream.revokeSeenBearers(); - const interrupted = yield* attempt(); expect( interrupted.ok, `the call whose credential write was interrupted reports the failure (got: ${interrupted.text.slice(0, 200)})`, ).toBe(false); - yield* Effect.promise(() => workos.faults.clear()); // The interrupted refresh still spent the refresh token it sent: the // authorization server rotated it and will not honour the old one again.