diff --git a/docs/adr/0043-reuse-warmed-tls-on-known-networks.md b/docs/adr/0043-reuse-warmed-tls-on-known-networks.md new file mode 100644 index 00000000..dc3c1473 --- /dev/null +++ b/docs/adr/0043-reuse-warmed-tls-on-known-networks.md @@ -0,0 +1,52 @@ +# 0043. Reuse warmed TLS state on the networks we already speak + +Status: accepted 2026-08-29 + +## Context + +The harness already talks to three networks: provider HTTP/SSE (and WS), +MCP Streamable HTTP, and the Codegraff gateway. Two patterns were wasting +a handshake on every first useful call: + +1. MCP Auto's modern `tools/list` probe and the legacy + `notifications/initialized` notify each built a throwaway `std.http.Client`, + so a successful probe's keep-alive died before `tools/call`, and initialized + paid a second TLS for a fire-and-forget 202. +2. Every WSS dial (`ws.WsClient.connect`) rescanned the host CA store from + disk, even though launch already warms the HTTP client's bundle. + +ADR 0002 (xAI WS full-resend), 0009/0011/0028 (prompt-cache keys), and 0035 +(deferred MCP join) stay untouched: this is transport reuse, not wire shape. + +## Decision + +- MCP HTTP probe and `notifications/initialized` use `server.transport.http`. + Do not construct a per-call client for those paths. +- WSS TLS uses a process-lifetime CA bundle warmed once (`http_warm.ensureProcessCa`). + A reconnect must not walk the host store again. +- WS→SSE fallback keeps the Agent's prewarmed HTTP pool (`postStream`). Do not + introduce a fresh client on that latch: WS never used the HTTP pool, and a + new TLS would be strictly more expensive. +- MCP Streamable HTTP advertises the std client's default Accept-Encoding + (gzip/deflate) and decompresses Content-Encoding. Do not omit it: a tools/list + catalog is the fat payload on that network, and provider POST already accepts + compression. + +## Consequences + +A modern MCP connect plus the next list is one TCP accept on keep-alive. +WSS reconnects skip the CA disk walk. MCP catalogs can travel gzip-compressed. +Revisit only if a shared `std.http.Client` is shown unsafe for the concurrent +initialized+list pair, a host CA rotation must be picked up mid-process without +restart, or a server is broken by Accept-Encoding. + +## Measured (2026-08-30, this host) + +| Path | Before | After | Left on the table | +|---|---|---|---| +| MCP modern connect + next `tools/list` | 2 TCP accepts (throwaway probe client, then the persistent one) | 1 accept, 2 POSTs | nothing — keep-alive is the rest | +| WSS CA disk walk | 1 `rescan` per connect, including every reconnect | 1 per process | launch still walks once for the HTTP client (~5–7 ms, 144 certs / 154 KB here). Cloning that bundle into the process one saves one launch scan and risks a double-free; not worth it | +| MCP `tools/list` catalog (40-tool fixture) | 6110 B raw (`Accept-Encoding` omitted) | 320 B gzip (5% of plaintext) | only if the server ignores gzip — then we still send the header and read identity | +| WS→SSE latch | prewarmed Agent HTTP pool | unchanged | a fresh client would add a TLS handshake | +| xAI / Codex WS turn body | full history | unchanged | ADR 0002: no `previous_response_id` chain | +| MCP OAuth token / login 401 probe | throwaway `std.http.Client`; login probe omits encoding | unchanged | not on the turn path | diff --git a/docs/adr/README.md b/docs/adr/README.md index e76f1c51..13abb757 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -52,6 +52,7 @@ record only when you need the evidence or the edge cases. | [0039](0039-local-tools-are-project-scripts.md) | Agent-authored local tools are project scripts under `.graff/tools/`; skills stay instructions. Runtime catalog extras, not `schema.effectiveRootSpecs`. | | [0040](0040-codedb-stays-when-licensed.md) | Ordinary reads use native `codedb` / `read_file`; codedb-pro is extra search, not the default reader. | | [0041](0041-tui-is-an-acp-client.md) | The fullscreen TUI is an in-process ACP client: session/prompt in, session/update thought/tool/text out. No child `graff acp`. | +| [0043](0043-reuse-warmed-tls-on-known-networks.md) | Reuse warmed TLS: MCP probe/initialized stay on the persistent HTTP client; WSS CA is scanned once per process; WS→SSE keeps the prewarmed pool; MCP HTTP accepts gzip. | ## When to write one diff --git a/src/agent_stream.zig b/src/agent_stream.zig index 7335450c..40eb98f5 100644 --- a/src/agent_stream.zig +++ b/src/agent_stream.zig @@ -47,10 +47,10 @@ pub fn postStream(self: *Agent, body: []const u8) ![]u8 { return postStreamWithClient(self, self.client, body); } -/// SSE stream using an explicit HTTP client. Normal traffic uses the Agent's -/// shared pool; a WebSocket failure supplies a fresh client so a stale pooled -/// keep-alive cannot poison the WS→SSE handoff and every fallback retry dials -/// from a clean pool. +/// SSE stream using an explicit HTTP client. Normal traffic and the WS→SSE +/// latch both use the Agent's prewarmed pool (agent_ws.postLive); a test can +/// pass another client. A failed SEND still poisons that connection so the +/// next retry dials fresh instead of replaying a dead keep-alive. pub fn postStreamWithClient(self: *Agent, client: *std.http.Client, body: []const u8) ![]u8 { const sink = engine_sink.forAgent(self); sink.emit(self.io, .stream_begin); diff --git a/src/http_warm.zig b/src/http_warm.zig index ff5c88cf..adb889a4 100644 --- a/src/http_warm.zig +++ b/src/http_warm.zig @@ -1,14 +1,46 @@ -//! Shared HTTP client CA-bundle warming. +//! Shared CA-bundle warming for HTTP and WSS. const std = @import("std"); const Io = std.Io; +/// Process-lifetime bundle. Always `page_allocator` so a test GPA cannot +/// free it while a later WSS dial still holds the pointer. +var process_ca: std.crypto.Certificate.Bundle = .empty; +var process_ca_rw: Io.RwLock = .init; +var process_ca_init: Io.Mutex = .init; +var process_ca_ready = std.atomic.Value(bool).init(false); + +/// Test seam: how many times the process bundle actually hit the disk. +pub var process_ca_rescans: u32 = 0; + +pub fn processCa() *std.crypto.Certificate.Bundle { + return &process_ca; +} + +pub fn processCaLock() *Io.RwLock { + return &process_ca_rw; +} + +/// Scan the host CA store once. Later WSS reconnects reuse the same bytes. +pub fn ensureProcessCa(io: Io) !void { + if (process_ca_ready.load(.acquire)) return; + process_ca_init.lockUncancelable(io); + defer process_ca_init.unlock(io); + if (process_ca_ready.load(.acquire)) return; + const now = Io.Clock.real.now(io); + process_ca.rescan(std.heap.page_allocator, io, now) catch return error.HandshakeFailed; + process_ca_rescans += 1; + process_ca_ready.store(true, .release); +} + /// Pre-load the shared HTTP client's CA bundle single-threaded so concurrent -/// agents never race Zig's lazy first-connect rescan. +/// agents never race Zig's lazy first-connect rescan. Also warms the process +/// bundle so the first WSS dial does not pay a second disk walk. pub fn prewarmCaBundle(client: *std.http.Client, gpa: std.mem.Allocator, io: Io) void { const now = Io.Clock.real.now(io); client.ca_bundle.rescan(gpa, io, now) catch return; client.now = now; + ensureProcessCa(io) catch return; } /// Warm off the launch critical path. Outbound users wait on diff --git a/src/mcp_http.zig b/src/mcp_http.zig index dd3bf7ec..80489e7d 100644 --- a/src/mcp_http.zig +++ b/src/mcp_http.zig @@ -174,7 +174,6 @@ fn httpPostUnwatched(http: *HttpTransport, body: []const u8, meta: RequestMeta, .redirect_behavior = .unhandled, .headers = .{ .content_type = .{ .override = "application/json" }, - .accept_encoding = .omit, .user_agent = .{ .override = "codegraff-mcp/1" }, }, .extra_headers = extra, @@ -224,28 +223,7 @@ fn httpPostUnwatched(http: *HttpTransport, body: []const u8, meta: RequestMeta, } } - if (response.head.content_length == 0) return null; - const is_sse = if (response.head.content_type) |content_type| - std.ascii.startsWithIgnoreCase(content_type, "text/event-stream") - else - false; - var transfer_buf: [4096]u8 = undefined; - const reader = response.reader(&transfer_buf); - if (is_sse) return readSseResponse(http.client.allocator, reader, expected_id); - - const response_buf = try http.client.allocator.alloc(u8, max_http_response); - errdefer http.client.allocator.free(response_buf); - var fixed = Io.Writer.fixed(response_buf); - _ = reader.streamRemaining(&fixed) catch |err| switch (err) { - error.WriteFailed => return error.McpResponseTooLarge, - else => return err, - }; - const len = fixed.buffered().len; - if (len == 0) { - http.client.allocator.free(response_buf); - return null; - } - return try http.client.allocator.realloc(response_buf, len); + return readResponseBody(http.client.allocator, &response, expected_id); } const HttpPostDone = union(enum) { @@ -325,7 +303,6 @@ fn probeUnwatched(http: *HttpTransport, body: []const u8, meta: RequestMeta) !Pr .redirect_behavior = .unhandled, .headers = .{ .content_type = .{ .override = "application/json" }, - .accept_encoding = .omit, .user_agent = .{ .override = "codegraff-mcp/1" }, }, .extra_headers = extra, @@ -355,17 +332,35 @@ fn probeUnwatched(http: *HttpTransport, body: []const u8, meta: RequestMeta) !Pr if (req.connection) |connection| connection.closing = true; } - if (response.head.content_length == 0) return .{ .status = status, .body = null }; + return .{ .status = status, .body = try readResponseBody(http.client.allocator, &response, null) }; +} + +fn decompressWindow(gpa: Allocator, encoding: std.http.ContentEncoding) ![]u8 { + return switch (encoding) { + .identity => &.{}, + .zstd => gpa.alloc(u8, std.compress.zstd.default_window_len), + .deflate, .gzip => gpa.alloc(u8, std.compress.flate.max_window_len), + .compress => error.UnsupportedCompressionMethod, + }; +} + +/// Decode the HTTP body, honoring Content-Encoding (gzip/deflate/zstd). +/// Callers used to omit Accept-Encoding so catalogs crossed the wire raw. +fn readResponseBody(gpa: Allocator, response: *std.http.Client.Response, expected_id: ?i64) !?[]u8 { + if (response.head.content_length == 0) return null; const is_sse = if (response.head.content_type) |content_type| std.ascii.startsWithIgnoreCase(content_type, "text/event-stream") else false; + const window = try decompressWindow(gpa, response.head.content_encoding); + defer if (window.len > 0) gpa.free(window); var transfer_buf: [4096]u8 = undefined; - const reader = response.reader(&transfer_buf); - if (is_sse) return .{ .status = status, .body = try readSseResponse(http.client.allocator, reader, null) }; + var decompress: std.http.Decompress = undefined; + const reader = response.readerDecompressing(&transfer_buf, &decompress, window); + if (is_sse) return readSseResponse(gpa, reader, expected_id); - const response_buf = try http.client.allocator.alloc(u8, max_http_response); - errdefer http.client.allocator.free(response_buf); + const response_buf = try gpa.alloc(u8, max_http_response); + errdefer gpa.free(response_buf); var fixed = Io.Writer.fixed(response_buf); _ = reader.streamRemaining(&fixed) catch |err| switch (err) { error.WriteFailed => return error.McpResponseTooLarge, @@ -373,10 +368,10 @@ fn probeUnwatched(http: *HttpTransport, body: []const u8, meta: RequestMeta) !Pr }; const len = fixed.buffered().len; if (len == 0) { - http.client.allocator.free(response_buf); - return .{ .status = status, .body = null }; + gpa.free(response_buf); + return null; } - return .{ .status = status, .body = try http.client.allocator.realloc(response_buf, len) }; + return try gpa.realloc(response_buf, len); } const ProbeDone = union(enum) { diff --git a/src/mcp_lifecycle.zig b/src/mcp_lifecycle.zig index a7386758..abdf7992 100644 --- a/src/mcp_lifecycle.zig +++ b/src/mcp_lifecycle.zig @@ -6,7 +6,6 @@ //! surfaced rather than hidden behind fallback, matching rust-sdk Auto. const std = @import("std"); -const Io = std.Io; const Value = std.json.Value; const Allocator = std.mem.Allocator; @@ -59,27 +58,12 @@ const ProbeOut = struct { id: i64, }; -fn probeMethod( - io: Io, - gpa: Allocator, - url: []const u8, - headers: []const std.http.Header, - oauth_home: ?[]const u8, - method: []const u8, - id: i64, -) ProbeOut { - var arena_state = std.heap.ArenaAllocator.init(gpa); +fn probeMethod(http: *mcp_http.HttpTransport, method: []const u8, id: i64) ProbeOut { + var arena_state = std.heap.ArenaAllocator.init(http.client.allocator); defer arena_state.deinit(); - var transport: mcp_http.HttpTransport = .{ - .url = url, - .client = .{ .allocator = gpa, .io = io }, - .headers = headers, - .oauth_home = oauth_home, - }; - defer transport.client.deinit(); const body = mcp_protocol.buildRequest(arena_state.allocator(), id, method, "{}", true) catch return .{ .reply = .{ .status = 0, .body = null }, .id = id }; - const reply = mcp_http.probe(&transport, body, .{ + const reply = mcp_http.probe(http, body, .{ .protocol_version = modern_protocol, .method = method, .modern = true, @@ -126,7 +110,7 @@ fn connectHttpAttempt(server: *mcp_rpc.Server, a: Allocator, session_alloc: Allo const id_list = server.next_id; server.next_id += 1; - const list = probeMethod(io, gpa, http.url, http.headers, http.oauth_home, "tools/list", id_list); + const list = probeMethod(http, "tools/list", id_list); defer if (list.reply.body) |b| gpa.free(b); // Any modern request may be first. A real tools/list result is the catalog @@ -232,6 +216,14 @@ test "Auto: first launch tries modern tools/list before legacy fallback" { try std.testing.expect(list_pos < fallback_pos); } +test "Auto: modern probe reuses the persistent HTTP client" { + const src = @embedFile("mcp_lifecycle.zig"); + try std.testing.expect(std.mem.indexOf(u8, src, "fn probeMethod(http: *mcp_http.HttpTransport") != null); + try std.testing.expect(std.mem.indexOf(u8, src, "const list = probeMethod(http,") != null); + const throwaway = "var transport: " ++ "mcp_http.HttpTransport"; + try std.testing.expect(std.mem.indexOf(u8, src, throwaway) == null); +} + test { _ = @import("mcp_cache.zig"); } diff --git a/src/mcp_rpc.zig b/src/mcp_rpc.zig index ed5dc316..15a7d033 100644 --- a/src/mcp_rpc.zig +++ b/src/mcp_rpc.zig @@ -236,43 +236,21 @@ pub fn connectLegacy(server: *Server, a: Allocator, session_alloc: Allocator, bo return listed; } -const InitializedJob = struct { - url: []const u8, - headers: []const std.http.Header, - oauth_home: ?[]const u8, - protocol_version: []const u8, - gpa: Allocator, -}; - -fn httpInitializedTask(io: Io, job: InitializedJob) void { - var transport: mcp_http.HttpTransport = .{ - .url = job.url, - .client = .{ .allocator = job.gpa, .io = io }, - .headers = job.headers, - .oauth_home = job.oauth_home, - }; - defer transport.client.deinit(); +fn httpInitializedTask(http: *mcp_http.HttpTransport, protocol_version: []const u8) void { const body = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\",\"params\":{}}"; - if (mcp_http.post(&transport, body, .{ - .protocol_version = job.protocol_version, + if (mcp_http.post(http, body, .{ + .protocol_version = protocol_version, .method = "notifications/initialized", .modern = false, }, null)) |maybe| { - if (maybe) |b| job.gpa.free(b); + if (maybe) |b| http.client.allocator.free(b); } else |_| {} } fn kickHttpInitialized(server: *Server) void { const http = &server.transport.http; - const job = InitializedJob{ - .url = http.url, - .headers = http.headers, - .oauth_home = http.oauth_home, - .protocol_version = server.protocol_version, - .gpa = http.client.allocator, - }; - server.pending_initialized = http.client.io.concurrent(httpInitializedTask, .{ http.client.io, job }) catch - http.client.io.async(httpInitializedTask, .{ http.client.io, job }); + server.pending_initialized = http.client.io.concurrent(httpInitializedTask, .{ http, server.protocol_version }) catch + http.client.io.async(httpInitializedTask, .{ http, server.protocol_version }); } pub fn finishInitialized(server: *Server) void { diff --git a/src/net_efficiency_test.zig b/src/net_efficiency_test.zig new file mode 100644 index 00000000..c7f401a9 --- /dev/null +++ b/src/net_efficiency_test.zig @@ -0,0 +1,244 @@ +//! Measured hot-path checks for reuse on the networks graff already speaks: +//! process-warmed WSS CA, MCP Streamable HTTP keep-alive, and the source +//! guards that keep those clients from becoming throwaways again. + +const std = @import("std"); +const builtin = @import("builtin"); +const Io = std.Io; + +const http_warm = @import("http_warm.zig"); +const mcp_http = @import("mcp_http.zig"); +const mcp_lifecycle = @import("mcp_lifecycle.zig"); +const mcp_protocol = @import("mcp_protocol.zig"); +const mcp_rpc = @import("mcp_rpc.zig"); + +test "WSS CA bundle is scanned from disk at most once per process" { + const io = std.testing.io; + const before = http_warm.process_ca_rescans; + try http_warm.ensureProcessCa(io); + const mid = http_warm.process_ca_rescans; + try http_warm.ensureProcessCa(io); + const after = http_warm.process_ca_rescans; + try std.testing.expect(mid == before or mid == before + 1); + try std.testing.expectEqual(mid, after); + try std.testing.expect(http_warm.processCa().map.count() > 0); +} + +test "MCP initialized notify reuses the persistent HTTP client" { + const src = @embedFile("mcp_rpc.zig"); + try std.testing.expect(std.mem.indexOf(u8, src, "fn httpInitializedTask(http: *mcp_http.HttpTransport") != null); + try std.testing.expect(std.mem.indexOf(u8, src, "http.client.io.concurrent(httpInitializedTask, .{ http,") != null); + const throwaway = "var transport: " ++ "mcp_http.HttpTransport"; + try std.testing.expect(std.mem.indexOf(u8, src, throwaway) == null); +} + +test "WS→SSE fallback latches the prewarmed Agent client, not a fresh pool" { + const src = @embedFile("agent_ws.zig"); + const latch = std.mem.indexOf(u8, src, "return self.postStream(body);").?; + try std.testing.expect(std.mem.indexOf(u8, src, "postStreamFresh") == null); + try std.testing.expect(std.mem.indexOf(u8, src, "using persistent prewarmed SSE") != null); + _ = latch; +} + +const ListSrv = struct { + accepts: *std.atomic.Value(u8), + posts: *std.atomic.Value(u8), + done: *std.atomic.Value(bool), + + fn run(self: *ListSrv, io: Io, listener: *std.Io.net.Server) void { + while (!self.done.load(.acquire)) { + const stream = listener.accept(io) catch { + if (self.done.load(.acquire)) return; + continue; + }; + _ = self.accepts.fetchAdd(1, .monotonic); + defer stream.close(io); + self.serveConn(io, stream) catch {}; + } + } + + fn serveConn(self: *ListSrv, io: Io, stream: std.Io.net.Stream) !void { + var rbuf: [4096]u8 = undefined; + var wbuf: [4096]u8 = undefined; + var rd = std.Io.net.Stream.Reader.init(stream, io, &rbuf); + var wr = std.Io.net.Stream.Writer.init(stream, io, &wbuf); + const r = &rd.interface; + const w = &wr.interface; + while (self.posts.load(.acquire) < 2) { + var content_len: usize = 0; + while (true) { + const line = (r.takeDelimiter('\n') catch return) orelse return; + if (line.len == 0 or (line.len == 1 and line[0] == '\r')) break; + if (std.ascii.startsWithIgnoreCase(line, "content-length:")) { + const raw = if (line[line.len - 1] == '\r') line[0 .. line.len - 1] else line; + const v = std.mem.trim(u8, raw["content-length:".len..], " \t"); + content_len = std.fmt.parseInt(usize, v, 10) catch 0; + } + } + if (content_len > 0) _ = try r.take(content_len); + const n = self.posts.fetchAdd(1, .monotonic) + 1; + const body = if (n == 1) + \\{"jsonrpc":"2.0","id":1,"result":{"tools":[],"supportedVersions":["2026-07-28"]}} + else + \\{"jsonrpc":"2.0","id":2,"result":{"tools":[]}} + ; + try w.print( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\nConnection: keep-alive\r\n\r\n{s}", + .{ body.len, body }, + ); + try w.flush(); + } + } +}; + +test "MCP modern connect + next list reuse one TCP connection" { + if (builtin.os.tag == .windows) return error.SkipZigTest; + const gpa = std.testing.allocator; + const io = std.testing.io; + + var addr = try std.Io.net.IpAddress.parseLiteral("127.0.0.1:0"); + var listener = try std.Io.net.IpAddress.listen(&addr, io, .{}); + defer listener.deinit(io); + + var accepts: std.atomic.Value(u8) = .init(0); + var posts: std.atomic.Value(u8) = .init(0); + var done: std.atomic.Value(bool) = .init(false); + var srv: ListSrv = .{ .accepts = &accepts, .posts = &posts, .done = &done }; + var fut = io.async(ListSrv.run, .{ &srv, io, &listener }); + defer fut.await(io); + defer done.store(true, .release); + defer if (std.Io.net.IpAddress.connect(&listener.socket.address, io, .{ .mode = .stream })) |s| s.close(io) else |_| {}; + + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/mcp", .{listener.socket.address.getPort()}); + + var arena_state = std.heap.ArenaAllocator.init(gpa); + defer arena_state.deinit(); + const arena = arena_state.allocator(); + + var server: mcp_rpc.Server = .{ + .name = "loop", + .transport = .{ .http = .{ + .url = url, + .client = .{ .allocator = gpa, .io = io }, + } }, + }; + defer server.transport.http.client.deinit(); + + const first = try mcp_lifecycle.connectHttp(&server, arena, arena, .unknown); + try std.testing.expect(first.object.get("result") != null); + const second = try mcp_rpc.request(&server, arena, "{}", "tools/list", null); + try std.testing.expect(second.object.get("result") != null); + + try std.testing.expectEqual(@as(u8, 1), accepts.load(.monotonic)); + try std.testing.expectEqual(@as(u8, 2), posts.load(.monotonic)); +} + +test "MCP HTTP advertises gzip and decompresses a smaller wire body" { + if (builtin.os.tag == .windows) return error.SkipZigTest; + const gpa = std.testing.allocator; + const io = std.testing.io; + + var plain_w: Io.Writer.Allocating = .init(gpa); + defer plain_w.deinit(); + try plain_w.writer.writeAll("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"tools\":["); + for (0..40) |i| { + if (i != 0) try plain_w.writer.writeByte(','); + try plain_w.writer.print( + "{{\"name\":\"tool_{d}\",\"description\":\"search the workspace and return matching files\",\"inputSchema\":{{\"type\":\"object\",\"properties\":{{\"q\":{{\"type\":\"string\"}}}}}}}}", + .{i}, + ); + } + try plain_w.writer.writeAll("],\"supportedVersions\":[\"2026-07-28\"]}}"); + const plain = plain_w.writer.buffered(); + + const gz = try gzipAlloc(gpa, plain); + defer gpa.free(gz); + try std.testing.expect(gz.len < plain.len); + + var saw_accept_gzip = std.atomic.Value(bool).init(false); + var wire_len = std.atomic.Value(usize).init(0); + + var addr = try std.Io.net.IpAddress.parseLiteral("127.0.0.1:0"); + var listener = try std.Io.net.IpAddress.listen(&addr, io, .{}); + defer listener.deinit(io); + const Srv = struct { + fn run( + io_: Io, + listener_: *std.Io.net.Server, + gz_: []const u8, + saw: *std.atomic.Value(bool), + wire: *std.atomic.Value(usize), + ) void { + const stream = listener_.accept(io_) catch return; + defer stream.close(io_); + var rbuf: [4096]u8 = undefined; + var rd = std.Io.net.Stream.Reader.init(stream, io_, &rbuf); + const r = &rd.interface; + var content_len: usize = 0; + var accept_enc: bool = false; + while (true) { + const line = (r.takeDelimiter('\n') catch return) orelse return; + if (line.len == 0 or (line.len == 1 and line[0] == '\r')) break; + if (std.ascii.startsWithIgnoreCase(line, "accept-encoding:") and + std.mem.indexOf(u8, line, "gzip") != null) accept_enc = true; + if (std.ascii.startsWithIgnoreCase(line, "content-length:")) { + const raw = if (line[line.len - 1] == '\r') line[0 .. line.len - 1] else line; + content_len = std.fmt.parseInt(usize, std.mem.trim(u8, raw["content-length:".len..], " \t"), 10) catch 0; + } + } + if (content_len > 0) _ = r.take(content_len) catch {}; + saw.store(accept_enc, .release); + wire.store(gz_.len, .release); + var wbuf: [1024]u8 = undefined; + var wr = std.Io.net.Stream.Writer.init(stream, io_, &wbuf); + wr.interface.print( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Encoding: gzip\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n", + .{gz_.len}, + ) catch return; + wr.interface.writeAll(gz_) catch return; + wr.interface.flush() catch {}; + } + }; + var fut = io.async(Srv.run, .{ io, &listener, gz, &saw_accept_gzip, &wire_len }); + defer fut.await(io); + + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/mcp", .{listener.socket.address.getPort()}); + var http: mcp_http.HttpTransport = .{ + .url = url, + .client = .{ .allocator = gpa, .io = io }, + }; + defer http.client.deinit(); + const body = (try mcp_http.post(&http, "{}", .{ + .protocol_version = mcp_protocol.modern_protocol, + .method = "tools/list", + .modern = true, + }, 1)) orelse return error.TestUnexpectedResult; + defer gpa.free(body); + + try std.testing.expect(saw_accept_gzip.load(.acquire)); + try std.testing.expectEqual(gz.len, wire_len.load(.acquire)); + try std.testing.expect(wire_len.load(.acquire) < plain.len); + try std.testing.expectEqualStrings(plain, body); +} + +test "MCP HTTP no longer omits Accept-Encoding" { + const src = @embedFile("mcp_http.zig"); + const omit = "accept_encoding = " ++ ".omit"; + try std.testing.expect(std.mem.indexOf(u8, src, omit) == null); + try std.testing.expect(std.mem.indexOf(u8, src, "readerDecompressing") != null); +} + +fn gzipAlloc(gpa: std.mem.Allocator, plain: []const u8) ![]u8 { + var out_buf: [4096]u8 = undefined; + var out: Io.Writer = .fixed(&out_buf); + const window = try gpa.alloc(u8, std.compress.flate.max_window_len); + defer gpa.free(window); + const c = try gpa.create(std.compress.flate.Compress); + defer gpa.destroy(c); + c.* = try std.compress.flate.Compress.init(&out, window, .gzip, .default); + try c.writer.writeAll(plain); + try c.finish(); + return gpa.dupe(u8, out.buffered()); +} diff --git a/src/test_hooks.zig b/src/test_hooks.zig index 15212124..08fb225e 100644 --- a/src/test_hooks.zig +++ b/src/test_hooks.zig @@ -341,4 +341,5 @@ test { _ = @import("channel_worker.zig"); _ = @import("session_wake.zig"); _ = @import("tui_acp.zig"); + _ = @import("net_efficiency_test.zig"); } diff --git a/src/ws.zig b/src/ws.zig index 186fa5af..54839878 100644 --- a/src/ws.zig +++ b/src/ws.zig @@ -21,6 +21,7 @@ const net = std.Io.net; const HostName = net.HostName; const Allocator = std.mem.Allocator; const tls = std.crypto.tls; +const http_warm = @import("http_warm.zig"); /// GRAFF_WS_DEBUG=1 → dump the handshake + frame headers to stderr. pub var g_debug: bool = false; @@ -76,8 +77,6 @@ pub const WsClient = struct { r: *Io.Reader = undefined, w: *Io.Writer = undefined, tls_client: ?tls.Client = null, - ca_bundle: std.crypto.Certificate.Bundle = .empty, - ca_lock: Io.RwLock = .init, gpa: Allocator, /// (#401) The peer is wedged or gone — tear down with a plain FIN instead /// of deinit's courtesy close frame, which is another blocking write on the @@ -123,28 +122,26 @@ pub const WsClient = struct { self.* = .{ .io = io, .stream = stream, .rd = undefined, .wr = undefined, .gpa = gpa }; self.rd = net.Stream.Reader.init(stream, io, &self.sock_rbuf); self.wr = net.Stream.Writer.init(stream, io, &self.sock_wbuf); - // From here the client owns the socket (and, for wss, the CA bundle): - // release both if the TLS or upgrade handshake fails, or every failed - // dial leaks an fd — which #401's reconnect ladder now retries into. - errdefer { - self.ca_bundle.deinit(gpa); - self.stream.close(io); - } + // From here the client owns the socket: release it if the TLS or + // upgrade handshake fails, or every failed dial leaks an fd — which + // #401's reconnect ladder now retries into. The CA bundle is the + // process-warmed one (http_warm); do not deinit it here. + errdefer self.stream.close(io); if (u.tls) { var entropy: [tls.Client.Options.entropy_len]u8 = undefined; io.random(&entropy); - if (!insecure) self.ca_bundle.rescan(gpa, io, Io.Clock.real.now(io)) catch |e| { + if (!insecure) http_warm.ensureProcessCa(io) catch |e| { dbg("ca rescan failed: {s}", .{@errorName(e)}); return error.HandshakeFailed; }; self.tls_client = tls.Client.init(&self.rd.interface, &self.wr.interface, .{ .host = if (insecure) .no_verification else .{ .explicit = u.host }, .ca = if (insecure) .no_verification else .{ .bundle = .{ - .gpa = gpa, + .gpa = std.heap.page_allocator, .io = io, - .lock = &self.ca_lock, - .bundle = &self.ca_bundle, + .lock = http_warm.processCaLock(), + .bundle = http_warm.processCa(), } }, .write_buffer = &self.tls_wbuf, .read_buffer = &self.tls_rbuf, @@ -167,7 +164,6 @@ pub const WsClient = struct { pub fn deinit(self: *WsClient, gpa: Allocator) void { if (!self.dead) self.sendFrame(.close, "") catch {}; // (#401) see `dead` - self.ca_bundle.deinit(gpa); self.stream.close(self.io); gpa.destroy(self); }