* trim to the openSession restart-retry backstop Per review: the concurrency half is superseded by #1907 (already on main, stricter). Rebased onto current main and kept only the retry loop on the documented 503 restart envelope. * Match the exact MCP restart envelope before retrying * Test queue timeout with a controlled clock ---------
Unified · Split
| @@ -19,8 +19,8 @@ · 18 unmodified lines | |||
| 19 | // the whole boot (see that file for the value and its headroom story), so | 19 | // the whole boot (see that file for the value and its headroom story), so |
| 20 | // this test can cross it with a bounded number of real sessions instead of | 20 | // this test can cross it with a bounded number of real sessions instead of |
| 21 | // registering the production default of 32. | 21 | // registering the production default of 32. |
| 22 | import { expect } from "@effect/vitest"; | 22 | import { expect, it } from "@effect/vitest"; |
| 23 | import { Effect, Schedule } from "effect"; | 23 | import { Effect, Option, Schedule, Schema } from "effect"; |
| 24 | 24 | ||
| 25 | import { scenario } from "../src/scenario"; | 25 | import { scenario } from "../src/scenario"; |
| 26 | import { Mcp, Target, Telemetry } from "../src/services"; | 26 | import { Mcp, Target, Telemetry } from "../src/services"; |
| @@ -52,6 +52,68 @@ const postJson = (mcpUrl: string, bearer: string, body: unknown, sessionId?: str · 25 unmodified lines | |||
| 52 | body: JSON.stringify(body), | 52 | body: JSON.stringify(body), |
| 53 | }); | 53 | }); |
| 54 | 54 | ||
| 55 | const decodeRestartEnvelope = Schema.decodeUnknownOption( | ||
| 56 | Schema.fromJsonString( | ||
| 57 | Schema.Struct({ | ||
| 58 | jsonrpc: Schema.Literal("2.0"), | ||
| 59 | id: Schema.Null, | ||
| 60 | error: Schema.Struct({ | ||
| 61 | code: Schema.Literal(-32001), | ||
| 62 | message: Schema.Literal("MCP session is restarting, please retry"), | ||
| 63 | }), | ||
| 64 | }), | ||
| 65 | ), | ||
| 66 | ); | ||
| 67 | |||
| 68 | const isRestartResponse = (status: number, body: string): boolean => | ||
| 69 | status === 503 && Option.isSome(decodeRestartEnvelope(body)); | ||
| 70 | |||
| 71 | it.each([ | ||
| 72 | [ | ||
| 73 | 503, | ||
| 74 | { | ||
| 75 | jsonrpc: "2.0", | ||
| 76 | id: null, | ||
| 77 | error: { code: -32001, message: "MCP session is restarting, please retry" }, | ||
| 78 | }, | ||
| 79 | true, | ||
| 80 | ], | ||
| 81 | [ | ||
| 82 | 404, | ||
| 83 | { | ||
| 84 | jsonrpc: "2.0", | ||
| 85 | id: null, | ||
| 86 | error: { code: -32001, message: "MCP session is restarting, please retry" }, | ||
| 87 | }, | ||
| 88 | false, | ||
| 89 | ], | ||
| 90 | [ | ||
| 91 | 503, | ||
| 92 | { | ||
| 93 | jsonrpc: "2.0", | ||
| 94 | id: null, | ||
| 95 | error: { code: -32603, message: "MCP session is restarting, please retry" }, | ||
| 96 | }, | ||
| 97 | false, | ||
| 98 | ], | ||
| 99 | [ | ||
| 100 | 503, | ||
| 101 | { | ||
| 102 | jsonrpc: "2.0", | ||
| 103 | id: null, | ||
| 104 | error: { code: -32001, message: "MCP session is restarting unexpectedly" }, | ||
| 105 | }, | ||
| 106 | false, | ||
| 107 | ], | ||
| 108 | [503, { error: "MCP session is restarting, please retry" }, false], | ||
| 109 | ] as const)("only retries the documented restart envelope (%s, %j)", (status, body, retry) => { | ||
| 110 | expect(isRestartResponse(status, JSON.stringify(body))).toBe(retry); | ||
| 111 | }); | ||
| 112 | |||
| 113 | it("does not retry malformed restart responses", () => { | ||
| 114 | expect(isRestartResponse(503, "MCP session is restarting, please retry")).toBe(false); | ||
| 115 | }); | ||
| 116 | |||
| 55 | /** | 117 | /** |
| 56 | * Opens one fresh MCP session under an already-minted bearer. `initialize` | 118 | * Opens one fresh MCP session under an already-minted bearer. `initialize` |
| 57 | * without an existing `mcp-session-id` always mints a new session, the same | 119 | * without an existing `mcp-session-id` always mints a new session, the same |
| @@ -73,21 +135,42 @@ const openSession = async ( · 15 unmodified lines | |||
| 73 | label: string, | 135 | label: string, |
| 74 | recordSession: (sessionId: string) => void, | 136 | recordSession: (sessionId: string) => void, |
| 75 | ): Promise<string> => { | 137 | ): Promise<string> => { |
| 76 | const initialized = await postJson(mcpUrl, bearer, { | 138 | // The platform can reset a session Durable Object while its initialize |
| 77 | jsonrpc: "2.0" as const, | 139 | // is in flight, and the server answers that with the documented restart |
| 78 | id: "initialize", | 140 | // envelope (503, -32001, "MCP session is restarting, please retry") — the |
| 79 | method: "initialize", | 141 | // same contract a streamable-http client follows: same request, after the |
| 80 | params: { | 142 | // advertised delay. Treat it as transient here instead of failing the |
| 81 | protocolVersion: PROTOCOL_VERSION, | 143 | // scenario on a retryable platform blip. |
| 82 | capabilities: {}, | 144 | const RESTART_ATTEMPTS = 8; |
| 83 | clientInfo: { name: `executor-e2e-cap-eviction-${label}`, version: "0.0.1" }, | 145 | const RESTART_DELAY_MS = 2_000; // The host advertises Retry-After: 2. |
| 84 | }, | 146 | let minted: { readonly response: Response; readonly sessionId: string } | undefined; |
| 85 | }); | 147 | for (let attempt = 0; attempt < RESTART_ATTEMPTS; attempt += 1) { |
| 86 | const sessionId = initialized.headers.get("mcp-session-id"); | 148 | const response = await postJson(mcpUrl, bearer, { |
| 87 | if (!sessionId) { | 149 | jsonrpc: "2.0" as const, |
| 150 | id: "initialize", | ||
| 151 | method: "initialize", | ||
| 152 | params: { | ||
| 153 | protocolVersion: PROTOCOL_VERSION, | ||
| 154 | capabilities: {}, | ||
| 155 | clientInfo: { name: `executor-e2e-cap-eviction-${label}`, version: "0.0.1" }, | ||
| 156 | }, | ||
| 157 | }); | ||
| 158 | const candidate = response.headers.get("mcp-session-id"); | ||
| 159 | if (candidate !== null && candidate.length > 0) { | ||
| 160 | minted = { response, sessionId: candidate }; | ||
| 161 | break; | ||
| 162 | } | ||
| 163 | const body = await response.text(); | ||
| 164 | const isRestart = isRestartResponse(response.status, body); | ||
| 165 | if (!isRestart) break; | ||
| 166 | if (attempt === RESTART_ATTEMPTS - 1) break; | ||
| 167 | await new Promise((resolve) => setTimeout(resolve, RESTART_DELAY_MS)); | ||
| 168 | } | ||
| 169 | if (!minted) { | ||
| 88 | // oxlint-disable-next-line executor/no-error-constructor -- boundary: e2e setup precondition. | 170 | // oxlint-disable-next-line executor/no-error-constructor -- boundary: e2e setup precondition. |
| 89 | throw new Error(`openSession (${label}): no mcp-session-id header`); | 171 | throw new Error(`openSession (${label}): no mcp-session-id header`); |
| 90 | } | 172 | } |
| 173 | const { response: initialized, sessionId } = minted; | ||
| 91 | // Recorded the moment the id exists — BEFORE the body read and status | 174 | // Recorded the moment the id exists — BEFORE the body read and status |
| 92 | // assertion below, either of which can throw with the session already live | 175 | // assertion below, either of which can throw with the session already live |
| 93 | // on the server. The cleanup finalizer needs the id on every one of those | 176 | // on the server. The cleanup finalizer needs the id on every one of those |