filter: blocklist downloads never reuse a pooled connection, the peer closes it between passes

src/app.zig

@@ -482,7 +482,19 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
     // Two HTTP clients on purpose. A blocklist download streams tens of
     // megabytes and holds its connection for the whole of it; DoH queries must
     // not queue behind that, and the two have nothing to share but a type.
-    var fetch_http: std.http.Client = .{ .allocator = gpa, .io = io };
+    // `free_size = 0` makes `release` destroy every connection and the pool
+    // hand none back (`Client.zig:133`). The fetcher's `.keep_alive = false` is
+    // what sends `connection: close`, but it cannot carry this alone: `request`
+    // takes a pooled connection before it stores the flag (`Client.zig:1725`,
+    // `:1742`), and a failed send leaves a `.ready` connection that
+    // `Request.deinit` would pool (`Client.zig:890`). A pass runs once a day,
+    // so any kept connection is closed at the peer by the next one, and the
+    // client never retries a stale one — it fails the source for the day.
+    var fetch_http: std.http.Client = .{
+        .allocator = gpa,
+        .io = io,
+        .connection_pool = .{ .free_size = 0 },
+    };
     defer fetch_http.deinit();
     var dns_http: std.http.Client = .{ .allocator = gpa, .io = io };
     defer dns_http.deinit();

src/filter/fetcher.zig

@@ -1,9 +1,12 @@
 //! Blocklist download over HTTP/1.1.
 //!
-//! One `Fetcher` wraps a caller-owned `std.http.Client`, which owns the
-//! connection pool and the CA bundle, exactly as `upstream/doh_client.zig`
-//! does. This file knows nothing about parsing, files or the database: it GETs
-//! a URL and streams the bytes into a writer the caller supplies.
+//! One `Fetcher` wraps a caller-owned `std.http.Client`, which owns the CA
+//! bundle, as `upstream/doh_client.zig` does. Nothing here is pooled: every
+//! request asks for `connection: close`, and the caller is expected to give
+//! this file a client whose pool holds nothing (`free_size = 0`, set at the
+//! one construction site in `app.zig`). This file knows nothing about parsing,
+//! files or the database: it GETs a URL and streams the bytes into a writer
+//! the caller supplies.
 //!
 //! The body is never held whole. A blocklist can reach `max_body_bytes`, and
 //! the caller writes into a temporary file anyway, so nothing here allocates.
@@ -109,7 +112,16 @@ pub const Fetcher = struct {
         const uri = parseUrl(url) catch |err| return self.record(err, .connect, err, null, 0);
 
         var req = self.http.request(.GET, uri, .{
-            .keep_alive = true,
+            // Sends `connection: close`, so the peer ends the connection and
+            // `release` destroys it rather than pooling it (`Client.zig:133`,
+            // `:1167`). A pass runs once a day: a kept connection is one the
+            // peer has already closed by the next pass, and `std.http.Client`
+            // never retries a reused connection that reaches EOF before the
+            // first head byte — it returns `error.HttpConnectionClosing` and
+            // the source fails for the day. This flag cannot carry that alone,
+            // because `request` takes a pooled connection before it reads the
+            // flag; `app.zig` empties the pool itself.
+            .keep_alive = false,
             .headers = .{
                 // Identity only: a compressed transfer encoding would need
                 // `Response.readerDecompressing`, a decompression buffer and a

src/filter/filter_integration_test.zig

@@ -421,7 +421,7 @@ const Env = struct {
 // fixtures: the loopback http server
 // ---------------------------------------------------------------------------
 
-const Route = enum(u8) { body, redirect, not_found, oversize, chunked, stall, changed };
+const Route = enum(u8) { body, redirect, not_found, oversize, chunked, stall, changed, stale_keep_alive };
 
 /// How long the `stall` route holds a reply open when nothing releases it.
 ///
@@ -542,9 +542,10 @@ const HttpFixture = struct {
         }
     }
 
-    /// Every reply closes the connection. A keep-alive reply would leave the
-    /// fetcher holding the connection open while this server waits to accept a
-    /// second one that never comes (milestone-5 spec, S9 note from S8).
+    /// Every reply but `stale_keep_alive` closes the connection. A keep-alive
+    /// reply would leave the fetcher holding the connection open while this
+    /// server waits to accept a second one that never comes (milestone-5 spec,
+    /// S9 note from S8).
     fn respond(self: *HttpFixture, io: std.Io, request: *std.http.Server.Request) !void {
         switch (@as(Route, @enumFromInt(self.route.load(.acquire)))) {
             .body => try request.respond(self.body, .{ .keep_alive = false }),
@@ -566,6 +567,11 @@ const HttpFixture = struct {
                 .transfer_encoding = .none,
                 .extra_headers = &.{.{ .name = "content-length", .value = oversize_length }},
             }),
+            // Advertises keep-alive and then closes anyway, because `serve`
+            // takes one request per connection. That is the Pi's failure in
+            // miniature: a client that pools this connection reuses one the
+            // peer has already closed on its next request to the host.
+            .stale_keep_alive => try request.respond(self.body, .{ .keep_alive = true }),
             .chunked => try self.respondChunked(request),
             .stall => try self.respondStalled(io, request),
         }
@@ -1254,6 +1260,44 @@ fn publishFixtureFiles(env: *Env, id: i64, list_body: []const u8, wild_body: []c
     });
 }
 
+test "8b: a second download to the same host never reuses a dead pooled connection" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    const env = try Env.create(gpa);
+    defer env.destroy();
+    const io = env.io();
+
+    var fixture = try HttpFixture.init(io, http_body);
+    defer fixture.deinit(io);
+    // The server says keep-alive and then drops the socket, which is what the
+    // Pi's upstreams do between one daily pass and the next.
+    fixture.setRoute(.stale_keep_alive);
+    var group: std.Io.Group = .init;
+    defer group.cancel(io);
+    try group.concurrent(io, HttpFixture.serve, .{ &fixture, io });
+
+    var url_buf: [64]u8 = undefined;
+    const url = try fixture.url(&url_buf);
+
+    var sink_buf: [0]u8 = .{};
+    var first: std.Io.Writer.Discarding = .init(&sink_buf);
+    const one = try env.f.fetch(io, url, &first.writer);
+    try testing.expectEqual(@as(u64, http_body.len), one.bytes_read);
+
+    // The whole test is this second call. With a pooling client the request
+    // goes out over a connection the peer has already closed, no response head
+    // byte arrives, and `error.HttpConnectionClosing` comes back with no retry
+    // — the exact record the Pi produced at 0 bytes and under a millisecond.
+    var second: std.Io.Writer.Discarding = .init(&sink_buf);
+    const two = try env.f.fetch(io, url, &second.writer);
+    try testing.expectEqual(@as(u64, http_body.len), two.bytes_read);
+    try testing.expectEqual(@as(?fetcher.Failure, null), env.f.last_failure);
+
+    // Two connections, not one reused: the proof the first was not pooled.
+    try testing.expectEqual(@as(u32, 2), fixture.accepted.load(.monotonic));
+}
+
 test "9: a reload swaps under a held handle and the new generation follows the release" {
     if (!build_options.integration) return error.SkipZigTest;