Unified · Split
| @@ -17,14 +17,26 @@ · 16 unmodified lines | |||
| 17 | // | 17 | // |
| 18 | // Both are pinned here at the product surface, black box. Failures are armed on | 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 | 19 | // the WorkOS emulator that the product's own WorkOS client talks to; no product |
| 20 | // code, stubs, or internals are touched. | 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. | ||
| 21 | import { randomBytes } from "node:crypto"; | 28 | import { randomBytes } from "node:crypto"; |
| 22 | import { createServer } from "node:http"; | 29 | import { createServer } from "node:http"; |
| 23 | 30 | ||
| 24 | import { expect } from "@effect/vitest"; | 31 | import { expect } from "@effect/vitest"; |
| 25 | import { Effect } from "effect"; | 32 | import { Effect } from "effect"; |
| 26 | import { composePluginApi } from "@executor-js/api/server"; | 33 | import { composePluginApi } from "@executor-js/api/server"; |
| 27 | import { connectEmulator, type EmulatorClient } from "@executor-js/emulate"; | 34 | import { |
| 35 | type ArmedFault, | ||
| 36 | connectEmulator, | ||
| 37 | type EmulatorClient, | ||
| 38 | type FaultArmInput, | ||
| 39 | } from "@executor-js/emulate"; | ||
| 28 | import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; | 40 | import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; |
| 29 | import { | 41 | import { |
| 30 | AuthTemplateSlug, | 42 | AuthTemplateSlug, |
| @@ -47,6 +59,9 @@ const VAULT_CONFLICT_RESPONSE = { · 16 unmodified lines | |||
| 47 | body: { code: "conflict", message: "Current version does not match expected version" }, | 59 | body: { code: "conflict", message: "Current version does not match expected version" }, |
| 48 | } as const; | 60 | } as const; |
| 49 | 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 | |||
| 50 | type UpstreamHandle = { | 65 | type UpstreamHandle = { |
| 51 | readonly url: string; | 66 | readonly url: string; |
| 52 | /** Stop honouring the bearer(s) seen so far, exactly as a revocation would — | 67 | /** 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<Va · 109 unmodified lines | |||
| 162 | .filter((object): object is VaultObject => object !== null && object.name.includes(slug)); | 177 | .filter((object): object is VaultObject => object !== null && object.name.includes(slug)); |
| 163 | }); | 178 | }); |
| 164 | 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 | |||
| 165 | type CallResult = { readonly ok: boolean; readonly text: string }; | 231 | type CallResult = { readonly ok: boolean; readonly text: string }; |
| 166 | 232 | ||
| 167 | type ToolEnvelope = { | 233 | type ToolEnvelope = { |
| @@ -198,8 +264,6 @@ const connectIntegration = Effect.gen(function* () { · 30 unmodified lines | |||
| 198 | const slug = unique("credwrite"); | 264 | const slug = unique("credwrite"); |
| 199 | const clientSlug = OAuthClientSlug.make(unique("credwriteclient")); | 265 | const clientSlug = OAuthClientSlug.make(unique("credwriteclient")); |
| 200 | 266 | ||
| 201 | yield* Effect.addFinalizer(() => Effect.promise(() => workos.faults.clear()).pipe(Effect.ignore)); | ||
| 202 | |||
| 203 | yield* client.openapi.addSpec({ | 267 | yield* client.openapi.addSpec({ |
| 204 | payload: { | 268 | payload: { |
| 205 | spec: { kind: "blob", value: spec(upstream.url, oauth) }, | 269 | spec: { kind: "blob", value: spec(upstream.url, oauth) }, |
| @@ -343,35 +407,43 @@ scenario( · 137 unmodified lines | |||
| 343 | 407 | ||
| 344 | // Transient contention: the next few version-checked writes lose their | 408 | // Transient contention: the next few version-checked writes lose their |
| 345 | // race, exactly as a peer writer persisting the same connection's | 409 | // race, exactly as a peer writer persisting the same connection's |
| 346 | // credential would make them lose it. Three is as many as a tight, | 410 | // credential would make them lose it. Three is as many as the pre-fix |
| 347 | // wait-free retry loop could ever absorb — a policy that waits between | 411 | // policy could ever absorb. |
| 348 | // attempts outlasts them, one that does not, cannot. | 412 | const transient = yield* Effect.scoped( |
| 349 | yield* Effect.promise(() => | 413 | Effect.gen(function* () { |
| 350 | workos.faults.arm({ | 414 | const armed = yield* armFault(workos, { |
| 351 | match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, | 415 | match: VAULT_WRITE, |
| 352 | response: VAULT_CONFLICT_RESPONSE, | 416 | response: VAULT_CONFLICT_RESPONSE, |
| 353 | times: 3, | 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); | ||
| 354 | }), | 424 | }), |
| 355 | ); | 425 | ); |
| 356 | // Revoking upstream forces the very next call to refresh, so the contended | 426 | expect(transient, "the transiently-contended refresh really did collide three times").toBe(3); |
| 357 | // write happens inside a real user-visible request. | ||
| 358 | upstream.revokeSeenBearers(); | ||
| 359 | yield* call("transiently-contended-refresh"); | ||
| 360 | yield* Effect.promise(() => workos.faults.clear()); | ||
| 361 | 427 | ||
| 362 | // Sustained contention: every version-checked attempt loses. Dropping a | 428 | // Sustained contention: every version-checked attempt loses. Dropping a |
| 363 | // credential that has just been minted is the one unrecoverable outcome, | 429 | // credential that has just been minted is the one unrecoverable outcome, |
| 364 | // so the write still has to land. | 430 | // so the write still has to land. |
| 365 | yield* Effect.promise(() => | 431 | const sustained = yield* Effect.scoped( |
| 366 | workos.faults.arm({ | 432 | Effect.gen(function* () { |
| 367 | match: { method: "PUT", pathPattern: "/vault/v1/kv/*" }, | 433 | const armed = yield* armFault(workos, { |
| 368 | response: VAULT_CONFLICT_RESPONSE, | 434 | match: VAULT_WRITE, |
| 369 | times: 5, | 435 | response: VAULT_CONFLICT_RESPONSE, |
| 436 | times: 5, | ||
| 437 | }); | ||
| 438 | upstream.revokeSeenBearers(); | ||
| 439 | yield* call("continuously-contended-refresh"); | ||
| 440 | return yield* faultsServed(workos, armed); | ||
| 370 | }), | 441 | }), |
| 371 | ); | 442 | ); |
| 372 | upstream.revokeSeenBearers(); | 443 | expect( |
| 373 | yield* call("continuously-contended-refresh"); | 444 | sustained, |
| 374 | yield* Effect.promise(() => workos.faults.clear()); | 445 | "the continuously-contended refresh exhausted every version-checked attempt", |
| 446 | ).toBe(5); | ||
| 375 | 447 | ||
| 376 | // The durability half: the credential those contended writes were | 448 | // The durability half: the credential those contended writes were |
| 377 | // carrying is the rotated, single-use refresh token. If any of it had been | 449 | // carrying is the rotated, single-use refresh token. If any of it had been |
| @@ -387,6 +459,48 @@ scenario( · 9 unmodified lines | |||
| 387 | ), | 459 | ), |
| 388 | ); | 460 | ); |
| 389 | 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 | |||
| 390 | scenario( | 504 | scenario( |
| 391 | "Credential persistence · a write that fails mid-refresh leaves the connection able to refresh", | 505 | "Credential persistence · a write that fails mid-refresh leaves the connection able to refresh", |
| 392 | {}, | 506 | {}, |
| @@ -408,21 +522,30 @@ scenario( · 15 unmodified lines | |||
| 408 | ); | 522 | ); |
| 409 | expect(accessObject, "the connection stored an access token in the vault").toBeDefined(); | 523 | expect(accessObject, "the connection stored an access token in the vault").toBeDefined(); |
| 410 | 524 | ||
| 411 | yield* Effect.promise(() => | 525 | const interrupted = yield* Effect.scoped( |
| 412 | workos.faults.arm({ | 526 | Effect.gen(function* () { |
| 413 | match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, | 527 | const armed = yield* armFault(workos, { |
| 414 | response: { status: 503, body: { code: "unavailable", message: "vault unavailable" } }, | 528 | match: { method: "PUT", pathPattern: `/vault/v1/kv/${accessObject!.id}` }, |
| 415 | times: 1, | 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; | ||
| 416 | }), | 543 | }), |
| 417 | ); | 544 | ); |
| 418 | |||
| 419 | upstream.revokeSeenBearers(); | ||
| 420 | const interrupted = yield* attempt(); | ||
| 421 | expect( | 545 | expect( |
| 422 | interrupted.ok, | 546 | interrupted.ok, |
| 423 | `the call whose credential write was interrupted reports the failure (got: ${interrupted.text.slice(0, 200)})`, | 547 | `the call whose credential write was interrupted reports the failure (got: ${interrupted.text.slice(0, 200)})`, |
| 424 | ).toBe(false); | 548 | ).toBe(false); |
| 425 | yield* Effect.promise(() => workos.faults.clear()); | ||
| 426 | 549 | ||
| 427 | // The interrupted refresh still spent the refresh token it sent: the | 550 | // The interrupted refresh still spent the refresh token it sent: the |
| 428 | // authorization server rotated it and will not honour the old one again. | 551 | // authorization server rotated it and will not honour the old one again. |