| @@ -0,0 +1,570 @@ |
| 1 | +// Cloud: a refreshed OAuth credential has to be PERSISTED, and persisting it is |
| 2 | +// a pair of version-checked writes into WorkOS Vault — the rotated refresh |
| 3 | +// token and the new access token, one after the other, not atomically. Two |
| 4 | +// production failures live in that gap. |
| 5 | +// |
| 6 | +// 1. Contention. Two concurrent probes of one connection each run a refresh and |
| 7 | +// each write the same two objects, so a version-checked write comes back 409 |
| 8 | +// ("Current version does not match expected version"). A write that retries |
| 9 | +// with no wait re-collides inside the peer's own round trip, drains its |
| 10 | +// attempts in microseconds, and the refresh fails outright — the user saw a |
| 11 | +// 500 on a connection health request. |
| 12 | +// 2. Ordering. If the write that fails is the one carrying the rotated refresh |
| 13 | +// token, the credential is gone for good: minting already spent the refresh |
| 14 | +// token we sent, so nothing can mint again and every later use of the |
| 15 | +// connection comes back `invalid_grant`. The access token, by contrast, is |
| 16 | +// disposable — one more grant re-mints it. |
| 17 | +// |
| 18 | +// Both are pinned here at the product surface, black box. Failures are armed on |
| 19 | +// the WorkOS emulator that the product's own WorkOS client talks to; no product |
| 20 | +// code, stubs, or internals are touched. Contention is modelled twice, because |
| 21 | +// the two halves of the write policy fail differently: by a COUNT of collisions |
| 22 | +// (does the loop have enough attempts, and does it still land the value when it |
| 23 | +// runs out?) and by a WINDOW of time (does it wait long enough between attempts |
| 24 | +// to still be trying when the peer lets go?). Every scenario also reads the |
| 25 | +// emulator's ledger back to prove the collisions it armed were really served — |
| 26 | +// a fault whose pattern stopped matching would otherwise leave a test that |
| 27 | +// passes without ever contending. |
| 28 | +import { randomBytes } from "node:crypto"; |
| 29 | +import { createServer } from "node:http"; |
| 30 | + |
| 31 | +import { expect } from "@effect/vitest"; |
| 32 | +import { Effect } from "effect"; |
| 33 | +import { composePluginApi } from "@executor-js/api/server"; |
| 34 | +import { |
| 35 | + type ArmedFault, |
| 36 | + connectEmulator, |
| 37 | + type EmulatorClient, |
| 38 | + type FaultArmInput, |
| 39 | +} from "@executor-js/emulate"; |
| 40 | +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; |
| 41 | +import { |
| 42 | + AuthTemplateSlug, |
| 43 | + ConnectionName, |
| 44 | + IntegrationSlug, |
| 45 | + OAuthClientSlug, |
| 46 | +} from "@executor-js/sdk/shared"; |
| 47 | +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; |
| 48 | + |
| 49 | +import { scenario } from "../src/scenario"; |
| 50 | +import { Api, Mcp, Target } from "../src/services"; |
| 51 | +import { WORKOS_EMULATOR_PORT } from "../targets/cloud"; |
| 52 | + |
| 53 | +const api = composePluginApi([openApiHttpPlugin()] as const); |
| 54 | + |
| 55 | +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; |
| 56 | + |
| 57 | +const VAULT_CONFLICT_RESPONSE = { |
| 58 | + status: 409, |
| 59 | + body: { code: "conflict", message: "Current version does not match expected version" }, |
| 60 | +} as const; |
| 61 | + |
| 62 | +/** Every PUT the vault serves for a stored object — the version-checked write. */ |
| 63 | +const VAULT_WRITE = { method: "PUT", pathPattern: "/vault/v1/kv/*" } as const; |
| 64 | + |
| 65 | +type UpstreamHandle = { |
| 66 | + readonly url: string; |
| 67 | + /** Stop honouring the bearer(s) seen so far, exactly as a revocation would — |
| 68 | + * the deterministic way to make the very next call run a refresh. */ |
| 69 | + readonly revokeSeenBearers: () => void; |
| 70 | + readonly close: () => void; |
| 71 | +}; |
| 72 | + |
| 73 | +/** Upstream on 127.0.0.1 that authenticates for real: `GET /issues` is 200 for |
| 74 | + * any bearer it has not been told to reject, and 401 for one it has. */ |
| 75 | +const serveUpstream = () => |
| 76 | + Effect.acquireRelease( |
| 77 | + Effect.callback<UpstreamHandle>((resume) => { |
| 78 | + const seen: string[] = []; |
| 79 | + const revoked = new Set<string>(); |
| 80 | + const server = createServer((request, response) => { |
| 81 | + const bearer = (request.headers.authorization ?? "").replace(/^Bearer\s+/i, ""); |
| 82 | + if (request.method === "GET" && (request.url ?? "").startsWith("/issues")) { |
| 83 | + seen.push(bearer); |
| 84 | + if (revoked.has(bearer)) { |
| 85 | + response.writeHead(401, { "content-type": "application/json" }); |
| 86 | + response.end(JSON.stringify({ error: "invalid_token" })); |
| 87 | + return; |
| 88 | + } |
| 89 | + response.writeHead(200, { "content-type": "application/json" }); |
| 90 | + response.end(JSON.stringify({ issues: [{ id: 1, title: "first" }] })); |
| 91 | + return; |
| 92 | + } |
| 93 | + response.writeHead(404, { "content-type": "application/json" }); |
| 94 | + response.end(JSON.stringify({ error: "not_found" })); |
| 95 | + }); |
| 96 | + server.listen(0, "127.0.0.1", () => { |
| 97 | + const address = server.address(); |
| 98 | + const port = typeof address === "object" && address ? address.port : 0; |
| 99 | + resume( |
| 100 | + Effect.succeed({ |
| 101 | + url: `http://127.0.0.1:${port}`, |
| 102 | + revokeSeenBearers: () => { |
| 103 | + for (const bearer of seen) revoked.add(bearer); |
| 104 | + }, |
| 105 | + close: () => { |
| 106 | + server.close(); |
| 107 | + server.closeAllConnections(); |
| 108 | + }, |
| 109 | + }), |
| 110 | + ); |
| 111 | + }); |
| 112 | + }), |
| 113 | + (server) => Effect.sync(server.close), |
| 114 | + ); |
| 115 | + |
| 116 | +const spec = ( |
| 117 | + baseUrl: string, |
| 118 | + oauth: { readonly authorizationEndpoint: string; readonly tokenEndpoint: string }, |
| 119 | +): string => |
| 120 | + JSON.stringify({ |
| 121 | + openapi: "3.0.3", |
| 122 | + info: { title: "Issues API", version: "1.0.0" }, |
| 123 | + servers: [{ url: baseUrl }], |
| 124 | + paths: { |
| 125 | + "/issues": { |
| 126 | + get: { |
| 127 | + operationId: "listIssues", |
| 128 | + summary: "List issues", |
| 129 | + security: [{ oauth: ["issues.read"] }], |
| 130 | + responses: { "200": { description: "issues" } }, |
| 131 | + }, |
| 132 | + }, |
| 133 | + }, |
| 134 | + components: { |
| 135 | + securitySchemes: { |
| 136 | + oauth: { |
| 137 | + type: "oauth2", |
| 138 | + flows: { |
| 139 | + authorizationCode: { |
| 140 | + authorizationUrl: oauth.authorizationEndpoint, |
| 141 | + tokenUrl: oauth.tokenEndpoint, |
| 142 | + scopes: { "issues.read": "Read issues" }, |
| 143 | + }, |
| 144 | + }, |
| 145 | + }, |
| 146 | + }, |
| 147 | + }, |
| 148 | + }); |
| 149 | + |
| 150 | +const invokeByAddressCode = (address: string, args: unknown) => ` |
| 151 | +const segments = ${JSON.stringify(address)}.split(".").slice(1); |
| 152 | +let node = tools; |
| 153 | +for (const segment of segments) node = node[segment]; |
| 154 | +const result = await node(${JSON.stringify(args)}); |
| 155 | +return JSON.stringify(result); |
| 156 | +`; |
| 157 | + |
| 158 | +/** One vault object as the emulated WorkOS itself sees it. Read from the |
| 159 | + * emulator control plane — upstream state, never the product's own. */ |
| 160 | +type VaultObject = { readonly id: string; readonly name: string }; |
| 161 | + |
| 162 | +/** Read a vault row out of the emulator's generic store snapshot, or null when |
| 163 | + * it is not shaped like one. */ |
| 164 | +const readVaultObject = (row: Record<string, unknown>): VaultObject | null => |
| 165 | + typeof row.workos_id === "string" && typeof row.name === "string" |
| 166 | + ? { id: row.workos_id, name: row.name } |
| 167 | + : null; |
| 168 | + |
| 169 | +/** Every vault object this connection owns. The object name embeds the logical |
| 170 | + * item id, so the connection's own objects are the ones carrying its slug. */ |
| 171 | +const vaultObjectsFor = (workos: EmulatorClient, slug: string): Effect.Effect<VaultObject[]> => |
| 172 | + Effect.promise(async () => { |
| 173 | + const snapshot = await workos.state(); |
| 174 | + const items = snapshot.collections["workos.vault_objects"]?.items ?? []; |
| 175 | + return items |
| 176 | + .map((item) => readVaultObject({ ...item })) |
| 177 | + .filter((object): object is VaultObject => object !== null && object.name.includes(slug)); |
| 178 | + }); |
| 179 | + |
| 180 | +/** Arm a fault on the shared emulator and take it back down with the scope, by |
| 181 | + * id — a blanket `faults.clear()` would also disarm a neighbouring scenario's. */ |
| 182 | +const armFault = (workos: EmulatorClient, input: FaultArmInput) => |
| 183 | + Effect.acquireRelease( |
| 184 | + Effect.promise(() => workos.faults.arm(input)), |
| 185 | + (armed) => Effect.promise(() => workos.faults.clear(armed.id)).pipe(Effect.ignore), |
| 186 | + ); |
| 187 | + |
| 188 | +/** How many responses this armed fault actually injected, read back from the |
| 189 | + * emulator's ledger. This is the proof that the failure under test HAPPENED: |
| 190 | + * an armed fault whose pattern never matched leaves the product's writes |
| 191 | + * untouched, and a scenario asserting only "the call worked" would pass |
| 192 | + * without ever exercising the retry it claims to cover. */ |
| 193 | +const faultsServed = (workos: EmulatorClient, armed: ArmedFault): Effect.Effect<number> => |
| 194 | + Effect.promise(async () => { |
| 195 | + const entries = await workos.ledger.list(200); |
| 196 | + return entries.filter((entry) => entry.faultId === armed.id).length; |
| 197 | + }); |
| 198 | + |
| 199 | +/** Keep the vault's version-checked writes losing for a WINDOW of time that |
| 200 | + * starts at the first collision the product actually suffers — a peer writer |
| 201 | + * that holds the object for a while, rather than a fixed number of collisions. |
| 202 | + * This is the shape of the production failure: attempts spaced by a wait |
| 203 | + * outlast the peer's round trip; attempts fired back to back all land inside |
| 204 | + * it, drain, and the refresh dies. |
| 205 | + * |
| 206 | + * Anchoring on the first collision (not on arming) is what makes it |
| 207 | + * deterministic — a refresh only begins several round trips after the fault is |
| 208 | + * armed, so a window measured from arming would already be over. */ |
| 209 | +const holdWritesInConflict = (workos: EmulatorClient, windowMillis: number) => |
| 210 | + Effect.gen(function* () { |
| 211 | + // Far more conflicts than any bounded retry policy can consume, so the |
| 212 | + // window — not a counter — is what ends the contention. |
| 213 | + const armed = yield* armFault(workos, { |
| 214 | + match: VAULT_WRITE, |
| 215 | + response: VAULT_CONFLICT_RESPONSE, |
| 216 | + times: 64, |
| 217 | + }); |
| 218 | + const released = (async () => { |
| 219 | + const deadline = Date.now() + 30_000; |
| 220 | + while (Date.now() < deadline) { |
| 221 | + const live = (await workos.faults.list()).find((fault) => fault.id === armed.id); |
| 222 | + if (!live || live.remaining < armed.times) break; |
| 223 | + await new Promise((resolve) => setTimeout(resolve, 15)); |
| 224 | + } |
| 225 | + await new Promise((resolve) => setTimeout(resolve, windowMillis)); |
| 226 | + await workos.faults.clear(armed.id); |
| 227 | + })().catch(() => undefined); |
| 228 | + return { armed, released }; |
| 229 | + }); |
| 230 | + |
| 231 | +type CallResult = { readonly ok: boolean; readonly text: string }; |
| 232 | + |
| 233 | +type ToolEnvelope = { |
| 234 | + readonly ok: boolean; |
| 235 | + readonly data?: unknown; |
| 236 | + readonly error?: { |
| 237 | + readonly code?: string; |
| 238 | + readonly message?: string; |
| 239 | + readonly status?: number; |
| 240 | + }; |
| 241 | +}; |
| 242 | + |
| 243 | +/** An OAuth-backed integration, connected and proven working, plus the handles |
| 244 | + * needed to break its credential writes: the upstream that can revoke a bearer |
| 245 | + * (the deterministic refresh trigger) and the WorkOS emulator that serves the |
| 246 | + * vault. Everything it creates is torn down by scope finalizers. */ |
| 247 | +const connectIntegration = Effect.gen(function* () { |
| 248 | + const target = yield* Target; |
| 249 | + const { client: makeClient } = yield* Api; |
| 250 | + const mcp = yield* Mcp; |
| 251 | + const identity = yield* target.newIdentity(); |
| 252 | + const client = yield* makeClient(api, identity); |
| 253 | + const upstream = yield* serveUpstream(); |
| 254 | + // Long-lived tokens: the proactive expiry check must never fire, so the |
| 255 | + // upstream's 401 is the only thing that triggers the refresh whose |
| 256 | + // persistence is under test. |
| 257 | + const oauth = yield* serveOAuthTestServer({ |
| 258 | + scopes: ["issues.read"], |
| 259 | + tokenExpiresInSeconds: 3600, |
| 260 | + }); |
| 261 | + const workos = yield* Effect.promise(() => |
| 262 | + connectEmulator({ baseUrl: `http://127.0.0.1:${WORKOS_EMULATOR_PORT}` }), |
| 263 | + ); |
| 264 | + const slug = unique("credwrite"); |
| 265 | + const clientSlug = OAuthClientSlug.make(unique("credwriteclient")); |
| 266 | + |
| 267 | + yield* client.openapi.addSpec({ |
| 268 | + payload: { |
| 269 | + spec: { kind: "blob", value: spec(upstream.url, oauth) }, |
| 270 | + slug, |
| 271 | + baseUrl: upstream.url, |
| 272 | + authenticationTemplate: [ |
| 273 | + { |
| 274 | + slug: "oauth", |
| 275 | + kind: "oauth2", |
| 276 | + authorizationUrl: oauth.authorizationEndpoint, |
| 277 | + tokenUrl: oauth.tokenEndpoint, |
| 278 | + scopes: ["issues.read"], |
| 279 | + }, |
| 280 | + ], |
| 281 | + }, |
| 282 | + }); |
| 283 | + yield* Effect.addFinalizer(() => |
| 284 | + client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore), |
| 285 | + ); |
| 286 | + |
| 287 | + yield* client.oauth.createClient({ |
| 288 | + payload: { |
| 289 | + owner: "org", |
| 290 | + slug: clientSlug, |
| 291 | + grant: "authorization_code", |
| 292 | + authorizationUrl: oauth.authorizationEndpoint, |
| 293 | + tokenUrl: oauth.tokenEndpoint, |
| 294 | + clientId: "test-client", |
| 295 | + clientSecret: "test-secret", |
| 296 | + originIntegration: IntegrationSlug.make(slug), |
| 297 | + }, |
| 298 | + }); |
| 299 | + yield* Effect.addFinalizer(() => |
| 300 | + client.oauth |
| 301 | + .removeClient({ params: { slug: clientSlug }, payload: { owner: "org" } }) |
| 302 | + .pipe(Effect.ignore), |
| 303 | + ); |
| 304 | + |
| 305 | + const started = yield* client.oauth.start({ |
| 306 | + payload: { |
| 307 | + client: clientSlug, |
| 308 | + clientOwner: "org", |
| 309 | + owner: "org", |
| 310 | + name: ConnectionName.make("main"), |
| 311 | + integration: IntegrationSlug.make(slug), |
| 312 | + template: AuthTemplateSlug.make("oauth"), |
| 313 | + }, |
| 314 | + }); |
| 315 | + expect(started.status, "oauth.start redirects to the authorization server").toBe("redirect"); |
| 316 | + if (started.status !== "redirect") return yield* Effect.die("no redirect"); |
| 317 | + |
| 318 | + // Drive the test IdP's consent by hand (authorize -> login -> code). |
| 319 | + const code = yield* Effect.promise(async () => { |
| 320 | + const authorize = await fetch(started.authorizationUrl, { redirect: "manual" }); |
| 321 | + const loginUrl = authorize.headers.get("location"); |
| 322 | + if (!loginUrl) throw new Error(`authorize did not redirect: ${authorize.status}`); |
| 323 | + const login = await fetch(loginUrl, { |
| 324 | + method: "POST", |
| 325 | + headers: { |
| 326 | + authorization: `Basic ${Buffer.from("alice:password").toString("base64")}`, |
| 327 | + }, |
| 328 | + redirect: "manual", |
| 329 | + }); |
| 330 | + const callbackUrl = login.headers.get("location"); |
| 331 | + if (!callbackUrl) throw new Error(`login did not redirect: ${login.status}`); |
| 332 | + const minted = new URL(callbackUrl).searchParams.get("code"); |
| 333 | + if (!minted) throw new Error("callback carried no authorization code"); |
| 334 | + return minted; |
| 335 | + }); |
| 336 | + yield* client.oauth.complete({ payload: { state: started.state, code } }); |
| 337 | + yield* Effect.addFinalizer(() => |
| 338 | + client.connections |
| 339 | + .remove({ |
| 340 | + params: { |
| 341 | + owner: "org", |
| 342 | + integration: IntegrationSlug.make(slug), |
| 343 | + name: ConnectionName.make("main"), |
| 344 | + }, |
| 345 | + }) |
| 346 | + .pipe(Effect.ignore), |
| 347 | + ); |
| 348 | + |
| 349 | + const tools = yield* client.tools.list({ query: {} }); |
| 350 | + const address = tools |
| 351 | + .filter((tool) => String(tool.integration) === slug) |
| 352 | + .map((tool) => String(tool.address)) |
| 353 | + .find((addr) => addr.endsWith("listIssues")); |
| 354 | + expect(address, "the listIssues tool is in the catalog").toBeDefined(); |
| 355 | + |
| 356 | + const session = mcp.session(identity); |
| 357 | + /** Invoke the connected tool and report what came back, success or not. */ |
| 358 | + const attempt = (): Effect.Effect<CallResult, unknown> => |
| 359 | + Effect.gen(function* () { |
| 360 | + let called = yield* session.call("execute", { |
| 361 | + code: invokeByAddressCode(address!, {}), |
| 362 | + }); |
| 363 | + let guard = 0; |
| 364 | + while (called.text.includes("executionId:") && guard < 10) { |
| 365 | + called = yield* session.approvePaused(called.text); |
| 366 | + guard += 1; |
| 367 | + } |
| 368 | + return { ok: called.ok, text: called.text }; |
| 369 | + }); |
| 370 | + |
| 371 | + /** Invoke the tool and require it to have worked end to end. */ |
| 372 | + const call = (label: string): Effect.Effect<ToolEnvelope, unknown> => |
| 373 | + Effect.gen(function* () { |
| 374 | + const result = yield* attempt(); |
| 375 | + expect( |
| 376 | + result.ok, |
| 377 | + `the ${label} call reached the upstream and came back (got: ${result.text.slice(0, 400)})`, |
| 378 | + ).toBe(true); |
| 379 | + const envelope = JSON.parse(result.text) as ToolEnvelope; |
| 380 | + expect(envelope.ok, `the ${label} call succeeded`).toBe(true); |
| 381 | + return envelope; |
| 382 | + }); |
| 383 | + |
| 384 | + return { attempt, call, oauth, slug, upstream, workos } as const; |
| 385 | +}); |
| 386 | + |
| 387 | +/** Is this one of the authorization server's refresh grants? Counting them is |
| 388 | + * how a "success" is shown to involve a real re-mint, not a cached token. */ |
| 389 | +const isRefreshGrant = (request: { |
| 390 | + readonly path: string; |
| 391 | + readonly method: string; |
| 392 | + readonly body: string; |
| 393 | +}) => |
| 394 | + request.path === "/token" && |
| 395 | + request.method === "POST" && |
| 396 | + request.body.includes("grant_type=refresh_token"); |
| 397 | + |
| 398 | +scenario( |
| 399 | + "Credential persistence · contention on the vault write does not lose the refreshed credential", |
| 400 | + {}, |
| 401 | + Effect.scoped( |
| 402 | + Effect.gen(function* () { |
| 403 | + const { call, oauth, upstream, workos } = yield* connectIntegration; |
| 404 | + |
| 405 | + // Baseline: the connection works on its freshly minted token. |
| 406 | + yield* call("baseline"); |
| 407 | + |
| 408 | + // Transient contention: the next few version-checked writes lose their |
| 409 | + // race, exactly as a peer writer persisting the same connection's |
| 410 | + // credential would make them lose it. Three is as many as the pre-fix |
| 411 | + // policy could ever absorb. |
| 412 | + const transient = yield* Effect.scoped( |
| 413 | + Effect.gen(function* () { |
| 414 | + const armed = yield* armFault(workos, { |
| 415 | + match: VAULT_WRITE, |
| 416 | + response: VAULT_CONFLICT_RESPONSE, |
| 417 | + times: 3, |
| 418 | + }); |
| 419 | + // Revoking upstream forces the very next call to refresh, so the |
| 420 | + // contended write happens inside a real user-visible request. |
| 421 | + upstream.revokeSeenBearers(); |
| 422 | + yield* call("transiently-contended-refresh"); |
| 423 | + return yield* faultsServed(workos, armed); |
| 424 | + }), |
| 425 | + ); |
| 426 | + expect(transient, "the transiently-contended refresh really did collide three times").toBe(3); |
| 427 | + |
| 428 | + // Sustained contention: every version-checked attempt loses. Dropping a |
| 429 | + // credential that has just been minted is the one unrecoverable outcome, |
| 430 | + // so the write still has to land. |
| 431 | + const sustained = yield* Effect.scoped( |
| 432 | + Effect.gen(function* () { |
| 433 | + const armed = yield* armFault(workos, { |
| 434 | + match: VAULT_WRITE, |
| 435 | + response: VAULT_CONFLICT_RESPONSE, |
| 436 | + times: 5, |
| 437 | + }); |
| 438 | + upstream.revokeSeenBearers(); |
| 439 | + yield* call("continuously-contended-refresh"); |
| 440 | + return yield* faultsServed(workos, armed); |
| 441 | + }), |
| 442 | + ); |
| 443 | + expect( |
| 444 | + sustained, |
| 445 | + "the continuously-contended refresh exhausted every version-checked attempt", |
| 446 | + ).toBe(5); |
| 447 | + |
| 448 | + // The durability half: the credential those contended writes were |
| 449 | + // carrying is the rotated, single-use refresh token. If any of it had been |
| 450 | + // dropped, the connection could never mint anything again. |
| 451 | + upstream.revokeSeenBearers(); |
| 452 | + yield* call("post-contention"); |
| 453 | + |
| 454 | + expect( |
| 455 | + (yield* oauth.requests).filter(isRefreshGrant).length, |
| 456 | + "each revocation was absorbed by a real refresh grant", |
| 457 | + ).toBeGreaterThanOrEqual(3); |
| 458 | + }), |
| 459 | + ), |
| 460 | +); |
| 461 | + |
| 462 | +// The peer writer's round trip, in milliseconds. Long enough that a retry loop |
| 463 | +// firing its attempts back to back drains all of them (and its last-resort |
| 464 | +// write) inside the window; short enough that a loop waiting between attempts is |
| 465 | +// still trying when the window closes — the shipped policy's fourth and fifth |
| 466 | +// attempts fall no earlier than 175ms and 375ms after the first collision. |
| 467 | +const CONTENTION_WINDOW_MS = 200; |
| 468 | + |
| 469 | +scenario( |
| 470 | + "Credential persistence · a refresh outlasts contention that lasts longer than a round trip", |
| 471 | + {}, |
| 472 | + Effect.scoped( |
| 473 | + Effect.gen(function* () { |
| 474 | + const { call, oauth, upstream, workos } = yield* connectIntegration; |
| 475 | + |
| 476 | + yield* call("baseline"); |
| 477 | + |
| 478 | + // Contention bounded by TIME rather than by a count of collisions: this is |
| 479 | + // the production shape, where the peer holds the object for as long as its |
| 480 | + // own write takes and the loser has to still be trying when it lets go. |
| 481 | + const { armed, released } = yield* holdWritesInConflict(workos, CONTENTION_WINDOW_MS); |
| 482 | + upstream.revokeSeenBearers(); |
| 483 | + yield* call("refresh-under-a-contention-window"); |
| 484 | + yield* Effect.promise(() => released); |
| 485 | + |
| 486 | + expect( |
| 487 | + yield* faultsServed(workos, armed), |
| 488 | + "the refresh really was fighting a contended vault write", |
| 489 | + ).toBeGreaterThanOrEqual(3); |
| 490 | + |
| 491 | + // And the credential that survived the window is usable: the next |
| 492 | + // revocation is absorbed by another real refresh. |
| 493 | + upstream.revokeSeenBearers(); |
| 494 | + yield* call("post-window"); |
| 495 | + |
| 496 | + expect( |
| 497 | + (yield* oauth.requests).filter(isRefreshGrant).length, |
| 498 | + "both revocations were absorbed by real refresh grants", |
| 499 | + ).toBeGreaterThanOrEqual(2); |
| 500 | + }), |
| 501 | + ), |
| 502 | +); |
| 503 | + |
| 504 | +scenario( |
| 505 | + "Credential persistence · a write that fails mid-refresh leaves the connection able to refresh", |
| 506 | + {}, |
| 507 | + Effect.scoped( |
| 508 | + Effect.gen(function* () { |
| 509 | + const { attempt, call, oauth, slug, upstream, workos } = yield* connectIntegration; |
| 510 | + |
| 511 | + yield* call("baseline"); |
| 512 | + |
| 513 | + // A refresh persists two objects. Break the ACCESS TOKEN's object only, |
| 514 | + // once: whichever write runs first survives and whichever runs second is |
| 515 | + // lost, so failing this one is the direct question "which of the two does |
| 516 | + // the product persist first?". |
| 517 | + const objects = yield* vaultObjectsFor(workos, slug); |
| 518 | + const refreshObject = objects.find((object) => object.name.endsWith("refresh")); |
| 519 | + expect(refreshObject, "the connection stored a refresh token in the vault").toBeDefined(); |
| 520 | + const accessObject = objects.find( |
| 521 | + (object) => object !== refreshObject && refreshObject!.name.startsWith(object.name), |
| 522 | + ); |
| 523 | + expect(accessObject, "the connection stored an access token in the vault").toBeDefined(); |
| 524 | + |
| 525 | + const interrupted = yield* Effect.scoped( |
| 526 | + Effect.gen(function* () { |
| 527 | + const armed = yield* armFault(workos, { |
| 528 | + match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, |
| 529 | + response: { |
| 530 | + status: 503, |
| 531 | + body: { code: "unavailable", message: "vault unavailable" }, |
| 532 | + }, |
| 533 | + times: 1, |
| 534 | + }); |
| 535 | + |
| 536 | + upstream.revokeSeenBearers(); |
| 537 | + const result = yield* attempt(); |
| 538 | + expect( |
| 539 | + yield* faultsServed(workos, armed), |
| 540 | + "the access token's write is the one that broke", |
| 541 | + ).toBe(1); |
| 542 | + return result; |
| 543 | + }), |
| 544 | + ); |
| 545 | + expect( |
| 546 | + interrupted.ok, |
| 547 | + `the call whose credential write was interrupted reports the failure (got: ${interrupted.text.slice(0, 200)})`, |
| 548 | + ).toBe(false); |
| 549 | + |
| 550 | + // The interrupted refresh still spent the refresh token it sent: the |
| 551 | + // authorization server rotated it and will not honour the old one again. |
| 552 | + // The access token in the vault is the stale, revoked one, so the very |
| 553 | + // next call has to refresh — and can only do so if the ROTATED refresh |
| 554 | + // token is what survived the half-finished write. |
| 555 | + const recovered = yield* attempt(); |
| 556 | + expect( |
| 557 | + recovered.ok, |
| 558 | + `the connection still refreshes after the interrupted write (got: ${recovered.text.slice(0, 400)})`, |
| 559 | + ).toBe(true); |
| 560 | + expect((JSON.parse(recovered.text) as ToolEnvelope).ok, "the recovered call succeeded").toBe( |
| 561 | + true, |
| 562 | + ); |
| 563 | + |
| 564 | + expect( |
| 565 | + (yield* oauth.requests).filter(isRefreshGrant).length, |
| 566 | + "the interrupted refresh and the recovery were both real refresh grants", |
| 567 | + ).toBeGreaterThanOrEqual(2); |
| 568 | + }), |
| 569 | + ), |
| 570 | +); |