milestone 15: make a green run mean a real pass

src/filter/compiler.zig

@@ -466,6 +466,32 @@ test "an over-long line is skipped when the reader buffer is large" {
     try testing.expectEqualStrings("a.example.com\nb.example.com\n", c.list());
 }
 
+test "an over-long final line with no newline ends the stream inside the discard" {
+    const gpa = testing.allocator;
+    var text: std.ArrayList(u8) = .empty;
+    defer text.deinit(gpa);
+
+    try text.appendSlice(gpa, "a.example.com\n");
+    try text.appendNTimes(gpa, 'x', 5_000);
+
+    // A reader buffer smaller than the trailing line makes `takeDelimiter`
+    // report `error.StreamTooLong`, and the discard that follows then runs out
+    // of input because nothing terminates that line. That is the `EndOfStream`
+    // arm: the loop must count the line and stop, not treat the exhausted
+    // stream as a read failure.
+    var backing: std.Io.Reader = .fixed(text.items);
+    var buf: [max_line_len]u8 = undefined;
+    var limited = backing.limited(.unlimited, &buf);
+
+    var c = try compileReader(gpa, &limited.interface, .domains);
+    defer c.deinit();
+
+    try testing.expectEqual(@as(u32, 1), c.result.counts.long_lines);
+    try testing.expectEqual(@as(u32, 1), c.result.counts.domains);
+    try testing.expectEqualStrings("a.example.com\n", c.list());
+    try testing.expect(std.mem.indexOf(u8, c.list(), "xxxx") == null);
+}
+
 test "carriage returns are stripped" {
     var c = try compileText(testing.allocator, "b.example.com\r\na.example.com\r\n", .domains);
     defer c.deinit();

src/filter/filter_integration_test.zig

@@ -351,10 +351,37 @@ const Env = struct {
 // fixtures: the loopback http server
 // ---------------------------------------------------------------------------
 
-const Route = enum(u8) { body, redirect, not_found, oversize };
+const Route = enum(u8) { body, redirect, not_found, oversize, chunked };
 
 const redirect_path = "/redirected.txt";
 
+const chunked_domains = 1_500;
+
+/// The body the `chunked` route serves (milestone-15 ruling 7). Two thresholds
+/// matter here and they are different: a body over `fetcher.min_transfer_buf`
+/// (16 KiB) makes the fetcher's `pumpBody` loop iterate instead of finishing in
+/// one read, and a body over the fixture connection's 8192-byte write buffer
+/// makes the reply reach the wire in parts. 1500 lines of 31 bytes clears both
+/// with room to spare.
+fn chunkedFixtureBody(gpa: std.mem.Allocator) ![]u8 {
+    var out: std.Io.Writer.Allocating = .init(gpa);
+    errdefer out.deinit();
+    for (0..chunked_domains) |i| {
+        try out.writer.print("0.0.0.0 chunk{d:0>5}.example.com\n", .{i});
+    }
+    return out.toOwnedSlice();
+}
+
+/// `body` cut into three equal parts, deliberately not at line boundaries: a
+/// domain name then straddles every part boundary, so a byte the transfer loses
+/// or repeats there corrupts a name and moves the compiled count, instead of
+/// being absorbed by a spare newline.
+fn thirds(body: []const u8) [3][]const u8 {
+    const first = body.len / 3;
+    const second = 2 * (body.len / 3);
+    return .{ body[0..first], body[first..second], body[second..] };
+}
+
 /// Past `fetcher.max_body_bytes`. The cap is a constant, so the only way to
 /// reach it in a test is a declared length: the fetcher refuses on the response
 /// head, before a byte of body streams.
@@ -364,6 +391,11 @@ const HttpFixture = struct {
     server: net.Server,
     body: []const u8,
     route: std.atomic.Value(u8),
+    /// How many parts the `chunked` route has flushed. The test reads it to
+    /// prove the reply really left this server in pieces, because a `Writer`
+    /// reports a buffered part as written and would otherwise hide a fixture
+    /// that sent everything in one go.
+    flushed_parts: std.atomic.Value(u32),
 
     fn init(io: std.Io, body: []const u8) !HttpFixture {
         const local: net.IpAddress = try .parse("127.0.0.1", 0);
@@ -371,6 +403,7 @@ const HttpFixture = struct {
             .server = try local.listen(io, .{ .reuse_address = true }),
             .body = body,
             .route = .init(@intFromEnum(Route.body)),
+            .flushed_parts = .init(0),
         };
     }
 
@@ -429,8 +462,27 @@ const HttpFixture = struct {
                 .transfer_encoding = .none,
                 .extra_headers = &.{.{ .name = "content-length", .value = oversize_length }},
             }),
+            .chunked => try self.respondChunked(request),
         }
     }
+
+    /// Streams the body in three flushed parts instead of one `respond`
+    /// (milestone-15 ruling 7). Every other arm sends ~130 bytes in a single
+    /// write, which the fetcher consumes in one read: the loop that 35f2324
+    /// killed the process in was never driven end to end by this suite.
+    fn respondChunked(self: *HttpFixture, request: *std.http.Server.Request) !void {
+        var send_buf: [4096]u8 = undefined;
+        var stream = try request.respondStreaming(&send_buf, .{
+            .respond_options = .{ .keep_alive = false },
+        });
+        for (thirds(self.body)) |part| {
+            try stream.writer.writeAll(part);
+            // Without this the parts sit in `send_buf` and leave as one write.
+            try stream.flush();
+            _ = self.flushed_parts.fetchAdd(1, .release);
+        }
+        try stream.end();
+    }
 };
 
 // ---------------------------------------------------------------------------
@@ -1478,3 +1530,64 @@ test "17: each blocking mode synthesizes the documented blocked reply" {
         try testing.expect(packet.findOptRecord(nx_packet) != null);
     }
 }
+
+// ---------------------------------------------------------------------------
+// 18: the multi-read download path
+// ---------------------------------------------------------------------------
+
+test "18: a body streamed in flushed parts survives the fetcher's multi-read pump" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    const env = try Env.create(gpa);
+    defer env.destroy();
+    const io = env.io();
+
+    const body = try chunkedFixtureBody(gpa);
+    defer gpa.free(body);
+    try testing.expect(body.len >= 24 * 1024);
+    try testing.expect(body.len > fetcher.min_transfer_buf);
+
+    var fixture = try HttpFixture.init(io, body);
+    defer fixture.deinit(io);
+    fixture.setRoute(.chunked);
+    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);
+    const id = try seedSource(&env.database, url);
+
+    try testing.expect(try refreshOnce(env, url));
+    try env.mgr.reload(io);
+
+    try testing.expect(fixture.flushed_parts.load(.acquire) >= 3);
+
+    var rows = try listRows(&env.database);
+    defer rows.deinit();
+    const row = try rows.byUrl(url);
+    try testing.expectEqual(@as(i64, chunked_domains), row.domain_count);
+    try testing.expectEqual(@as(i64, 0), row.wildcard_count);
+    try testing.expectEqual(manager.State.ok, (try env.status(id)).state);
+
+    // Every name arrived intact, including the two that straddled a flush
+    // boundary: a byte lost or repeated at a boundary corrupts a name, which
+    // moves the compiled count away from the number of lines sent.
+    var dir = try env.blocklistDir();
+    defer dir.close(io);
+    var bodies = try Bodies.read(gpa, io, dir, "1");
+    defer bodies.deinit(gpa);
+    try testing.expectEqual(
+        @as(usize, chunked_domains),
+        std.mem.count(u8, manager.stripHeader(bodies.list), "\n"),
+    );
+
+    var buf: [64]u8 = undefined;
+    for ([_]usize{ 0, chunked_domains / 3, 2 * chunked_domains / 3, chunked_domains - 1 }) |i| {
+        const domain = try std.fmt.bufPrint(&buf, "chunk{d:0>5}.example.com", .{i});
+        const decision, _ = try env.evaluate(domain);
+        try testing.expect(decision.blocked);
+        try testing.expectEqual(matcher.Reason.blocklist_domain, decision.reason);
+    }
+}