filter: a failed blocklist download names its phase, cause, status, bytes and elapsed time

src/filter/fetcher.zig

@@ -10,7 +10,9 @@
 //!
 //! There is no timeout parameter and no sleep. `std.http.Client` has no
 //! per-request deadline, so the caller runs `fetch` under `io.concurrent` and
-//! cancels the future; this file only propagates `error.Canceled`.
+//! cancels the future. A cancel during the body copy surfaces as
+//! `error.ReceiveFailed` with `Canceled` in `last_failure`; elsewhere it
+//! propagates as `error.Canceled`.
 
 const std = @import("std");
 const transport = @import("../upstream/transport.zig");
@@ -43,6 +45,30 @@ pub const Result = struct {
     status: std.http.Status,
 };
 
+/// The concrete fault behind an `Error`. `cause` is unwrapped from whatever
+/// `std.http.Client` stashed on the connection or the response, so it names a
+/// record-layer TLS fault or an HTTP framing fault rather than the collapsed
+/// `error.ReadFailed`.
+pub const Failure = struct {
+    phase: Phase,
+    cause: anyerror,
+    /// The response status if a head arrived before the failure.
+    status: ?std.http.Status,
+    /// Body bytes delivered to the caller's writer before the failure, and a
+    /// lower bound rather than an exact count: at least this many reached `w`.
+    /// A single `Reader.stream` call can hand bytes to `w` and then fail, and
+    /// that partial delivery is not counted. Zero on every phase before
+    /// `receive_body`.
+    bytes_read: u64,
+};
+
+/// Which call failed. The phase is what decides the classification, and only
+/// the call site knows it — guessing it from an error name would be wrong the
+/// first time two phases shared an error. `receive_body` is distinct from
+/// `receive`: the head arrived, so the fault is in the transfer, not the
+/// response.
+pub const Phase = enum { connect, send, receive, receive_body };
+
 pub const Fetcher = struct {
     /// Caller-owned; shared across sources, pools connections.
     http: *std.http.Client,
@@ -55,6 +81,12 @@ pub const Fetcher = struct {
     /// one. `error.HttpStatus` carries no `Result`, and the operator's message
     /// needs the number, so it is readable here after a failed `fetch`.
     last_status: ?std.http.Status = null,
+    /// What the most recent `fetch` failed on, or null if it succeeded or has
+    /// not run. The taxonomy `Error` a caller receives names six outcomes; this
+    /// names the one concrete fault behind the outcome, so an operator reading
+    /// the log can tell a TLS alert from a reset connection from a truncated
+    /// chunk.
+    last_failure: ?Failure = null,
 
     /// GETs `url` and streams the body into `w`.
     pub fn fetch(
@@ -71,9 +103,10 @@ pub const Fetcher = struct {
         std.debug.assert(self.transfer_buf.len >= min_transfer_buf);
         std.debug.assert(self.redirect_buf.len >= redirect_buffer_len);
 
-        const uri = try parseUrl(url);
-
         self.last_status = null;
+        self.last_failure = null;
+
+        const uri = parseUrl(url) catch |err| return self.record(err, .connect, err, null, 0);
 
         var req = self.http.request(.GET, uri, .{
             .keep_alive = true,
@@ -83,25 +116,88 @@ pub const Fetcher = struct {
                 // second failure surface, for a download that runs once a day.
                 .accept_encoding = .{ .override = "identity" },
             },
-        }) catch |err| return mapError(err, .connect);
+        }) catch |err| return self.record(mapError(err, .connect), .connect, err, null, 0);
         defer req.deinit();
 
-        req.sendBodiless() catch |err| return mapError(err, .send);
+        req.sendBodiless() catch |err|
+            return self.record(mapError(err, .send), .send, transport.sendCause(&req, err), null, 0);
 
-        var resp = req.receiveHead(self.redirect_buf) catch |err| return mapError(err, .receive);
+        var resp = req.receiveHead(self.redirect_buf) catch |err|
+            return self.record(mapError(err, .receive), .receive, transport.headCause(&req, err), null, 0);
 
         self.last_status = resp.head.status;
-        if (resp.head.status != .ok) return error.HttpStatus;
+        if (resp.head.status != .ok) {
+            return self.record(error.HttpStatus, .receive, error.HttpStatus, resp.head.status, 0);
+        }
 
         // `content-type` is deliberately not checked: blocklists are served as
         // text/plain, application/octet-stream and text/html alike, and the
         // compiler's invalid-line counters are the honest signal about content.
         if (resp.head.content_length) |declared| {
-            if (declared > max_body_bytes) return error.BodyTooLarge;
+            if (declared > max_body_bytes) {
+                return self.record(
+                    error.BodyTooLarge,
+                    .receive,
+                    error.BodyTooLarge,
+                    resp.head.status,
+                    0,
+                );
+            }
         }
 
         const body = resp.reader(self.transfer_buf);
-        return .{ .bytes_read = try pumpBody(body, w, max_body_bytes), .status = resp.head.status };
+        var pump: Pump = .{};
+        const bytes_read = pumpBody(body, w, max_body_bytes, &pump) catch |err| return self.record(
+            err,
+            .receive_body,
+            transport.bodyCause(&resp, pump.cause),
+            resp.head.status,
+            pump.delivered,
+        );
+        return .{ .bytes_read = bytes_read, .status = resp.head.status };
+    }
+
+    /// Stores the failure and hands back `mapped` unchanged.
+    ///
+    /// `mapped` is the taxonomy error the call site already decided, never one
+    /// this function derives. `mapError` answers for a raw std error, and the
+    /// call sites that raise a taxonomy member directly — a non-200 head, a
+    /// declared length over the cap — pass that member: feeding either back
+    /// through `mapError` would find no case for it and fall to the phase
+    /// default, turning `error.HttpStatus` into `error.ReceiveFailed`.
+    ///
+    /// `cause` never reaches the returned error. The unwrap is for the
+    /// operator's line, and a caller that ignores `last_failure` sees exactly
+    /// what it saw before the unwrap existed.
+    fn record(
+        self: *Fetcher,
+        mapped: Error,
+        phase: Phase,
+        cause: anyerror,
+        status: ?std.http.Status,
+        bytes_read: u64,
+    ) Error {
+        self.last_failure = .{
+            .phase = phase,
+            .cause = cause,
+            .status = status,
+            .bytes_read = bytes_read,
+        };
+        return mapped;
+    }
+};
+
+/// What `pumpBody` reports alongside its error: the reader error before
+/// `mapError` collapses it, and the bytes already handed to `w`. `pumpBody`
+/// knows nothing of `std.http`, so the unwrap of `cause` happens in `fetch`,
+/// which holds the response.
+const Pump = struct {
+    cause: anyerror = error.Unexpected,
+    delivered: u64 = 0,
+
+    fn fail(self: *Pump, cause: anyerror, mapped: Error) Error {
+        self.cause = cause;
+        return mapped;
     }
 };
 
@@ -116,17 +212,19 @@ pub const Fetcher = struct {
 /// `self.transfer_buf` — makes `@memcpy` copy the buffer onto itself
 /// (Reader.zig:677) and panics the process on the first read that finds bytes
 /// already buffered.
-fn pumpBody(body: *std.Io.Reader, w: *std.Io.Writer, cap: usize) Error!u64 {
+fn pumpBody(body: *std.Io.Reader, w: *std.Io.Writer, cap: usize, out: *Pump) Error!u64 {
     var total: usize = 0;
     while (total < cap) {
+        out.delivered = total;
         total += body.stream(w, .limited(cap - total)) catch |err| switch (err) {
             error.EndOfStream => return total,
             // The caller owns `w` and can read the concrete failure from its
             // own writer; this taxonomy has no member for a failing sink.
-            error.WriteFailed => return error.Unexpected,
-            else => |e| return mapError(e, .receive),
+            error.WriteFailed => return out.fail(err, error.Unexpected),
+            else => |e| return out.fail(e, mapError(e, .receive_body)),
         };
     }
+    out.delivered = total;
 
     // `cap` written exactly. One more byte separates a body that fits from one
     // that was cut off, and it must not reach `w`.
@@ -134,10 +232,10 @@ fn pumpBody(body: *std.Io.Reader, w: *std.Io.Writer, cap: usize) Error!u64 {
     var probe: std.Io.Writer.Discarding = .init(&probe_buf);
     const extra = body.stream(&probe.writer, .limited(1)) catch |err| switch (err) {
         error.EndOfStream => return total,
-        error.WriteFailed => return error.Unexpected,
-        else => |e| return mapError(e, .receive),
+        error.WriteFailed => return out.fail(err, error.Unexpected),
+        else => |e| return out.fail(e, mapError(e, .receive_body)),
     };
-    return if (extra == 0) total else error.BodyTooLarge;
+    return if (extra == 0) total else out.fail(error.BodyTooLarge, error.BodyTooLarge);
 }
 
 /// A scheme other than `http`/`https`, an unparseable URL and a URL with no
@@ -151,11 +249,6 @@ fn parseUrl(url: []const u8) Error!std.Uri {
     return uri;
 }
 
-/// Which call failed. The phase is what decides the classification, and only
-/// the call site knows it — guessing it from an error name would be wrong the
-/// first time two phases shared an error.
-const Phase = enum { connect, send, receive };
-
 // `error.X` in an expression names a member into existence rather than
 // referring to one, so the switch in `mapError` would keep compiling — and
 // silently stop matching — if std renamed either of these. This is what fails
@@ -179,9 +272,10 @@ fn errorSetHas(comptime Set: type, comptime name: []const u8) bool {
 /// `std.http.Client` collapses every handshake fault into
 /// `error.TlsInitializationFailed` (Client.zig:1470) and every bundle fault into
 /// `error.CertificateBundleLoadFailure`, and this file never unwraps a
-/// connection's stashed read cause — it maps the collapsed `error.ReadFailed` by
-/// phase — so the record-layer members of `std.crypto.tls.Client.ReadError`
-/// cannot arrive here. `doh_client.zig` does unwrap, and names them.
+/// connection's stashed read cause *for classification* — it maps the collapsed
+/// `error.ReadFailed` by phase — so the record-layer members of
+/// `std.crypto.tls.Client.ReadError` cannot arrive here. `fetch` does unwrap
+/// them, into `Failure.cause`, which no classification reads.
 fn mapError(err: anyerror, phase: Phase) Error {
     if (transport.mapLocal(err)) |local| return narrowLocal(local);
     switch (err) {
@@ -198,7 +292,7 @@ fn mapError(err: anyerror, phase: Phase) Error {
     return switch (phase) {
         .connect => error.ConnectFailed,
         .send => error.SendFailed,
-        .receive => error.ReceiveFailed,
+        .receive, .receive_body => error.ReceiveFailed,
     };
 }
 
@@ -230,7 +324,8 @@ test "pumpBody streams a fully buffered body without aliasing its source" {
     var out: [payload.len]u8 = undefined;
     var sink: std.Io.Writer = .fixed(&out);
 
-    const n = try pumpBody(&body, &sink, max_body_bytes);
+    var pump: Pump = .{};
+    const n = try pumpBody(&body, &sink, max_body_bytes, &pump);
     try testing.expectEqual(@as(u64, payload.len), n);
     try testing.expectEqualStrings(payload, out[0..payload.len]);
 }
@@ -241,7 +336,8 @@ test "pumpBody accepts a body of exactly the cap" {
     var out: [payload.len]u8 = undefined;
     var sink: std.Io.Writer = .fixed(&out);
 
-    const n = try pumpBody(&body, &sink, payload.len);
+    var pump: Pump = .{};
+    const n = try pumpBody(&body, &sink, payload.len, &pump);
     try testing.expectEqual(@as(u64, payload.len), n);
     try testing.expectEqualStrings(payload, out[0..payload.len]);
 }
@@ -252,10 +348,12 @@ test "pumpBody refuses a body one byte over the cap" {
     var out: [payload.len]u8 = undefined;
     var sink: std.Io.Writer = .fixed(&out);
 
+    var pump: Pump = .{};
     try testing.expectError(
         error.BodyTooLarge,
-        pumpBody(&body, &sink, payload.len - 1),
+        pumpBody(&body, &sink, payload.len - 1, &pump),
     );
+    try testing.expectEqual(error.BodyTooLarge, pump.cause);
 }
 
 test "pumpBody reports an empty body as zero bytes" {
@@ -263,7 +361,8 @@ test "pumpBody reports an empty body as zero bytes" {
     var sink_buf: [0]u8 = .{};
     var discarding: std.Io.Writer.Discarding = .init(&sink_buf);
 
-    try testing.expectEqual(@as(u64, 0), try pumpBody(&body, &discarding.writer, max_body_bytes));
+    var pump: Pump = .{};
+    try testing.expectEqual(@as(u64, 0), try pumpBody(&body, &discarding.writer, max_body_bytes, &pump));
 }
 
 fn undefinedFetcher(transfer_buf: []u8, redirect_buf: []u8) Fetcher {
@@ -359,3 +458,171 @@ test "caps are the values the memory budget was sized against" {
     try testing.expectEqual(@as(usize, 64 * 1024 * 1024), max_body_bytes);
     try testing.expectEqual(@as(usize, 8192), redirect_buffer_len);
 }
+
+/// Only the fields the unwrap helpers read are set; the rest of a `Connection`
+/// is two buffered streams, a host name and a pool node. `.plain` for the
+/// reason `doh_client.zig` gives: `getReadError` reaches a TLS connection's
+/// cause through `@fieldParentPtr`, which on a stub would read memory that was
+/// never a `Tls`.
+fn stubConnection(read_err: ?std.Io.net.Stream.Reader.Error) std.http.Client.Connection {
+    var connection: std.http.Client.Connection = undefined;
+    connection.protocol = .plain;
+    connection.stream_reader.err = read_err;
+    connection.stream_writer.err = null;
+    return connection;
+}
+
+test "a head failure records the stashed cause, not the collapsed read error" {
+    var connection = stubConnection(error.ConnectionResetByPeer);
+    var req: std.http.Client.Request = undefined;
+    req.connection = &connection;
+
+    var transfer_buf: [min_transfer_buf]u8 = undefined;
+    var redirect_buf: [redirect_buffer_len]u8 = undefined;
+    var f = undefinedFetcher(&transfer_buf, &redirect_buf);
+
+    const err = error.ReadFailed;
+    const mapped = f.record(mapError(err, .receive), .receive, transport.headCause(&req, err), null, 0);
+
+    // The taxonomy the caller sees is what it was before the unwrap existed.
+    try testing.expectEqual(error.ReceiveFailed, mapped);
+    const failure = f.last_failure.?;
+    try testing.expectEqual(Phase.receive, failure.phase);
+    try testing.expectEqual(error.ConnectionResetByPeer, failure.cause);
+    try testing.expectEqual(@as(?std.http.Status, null), failure.status);
+    try testing.expectEqual(@as(u64, 0), failure.bytes_read);
+}
+
+test "a body failure records the bytes delivered before it and the body phase" {
+    const payload = "0.0.0.0 ads.example.com\n";
+    var buffer: [payload.len]u8 = (payload ++ "").*;
+    // Buffered bytes reach `w` first; the vtable is only asked for more once
+    // the buffer is drained, which is the shape of a transfer cut off mid-body.
+    var body: std.Io.Reader = .{
+        .vtable = std.Io.Reader.failing.vtable,
+        .buffer = &buffer,
+        .seek = 0,
+        .end = buffer.len,
+    };
+    var sink_buf: [0]u8 = .{};
+    var discarding: std.Io.Writer.Discarding = .init(&sink_buf);
+
+    var pump: Pump = .{};
+    try testing.expectError(
+        error.ReceiveFailed,
+        pumpBody(&body, &discarding.writer, max_body_bytes, &pump),
+    );
+    try testing.expectEqual(@as(u64, payload.len), pump.delivered);
+
+    var connection = stubConnection(null);
+    var req: std.http.Client.Request = undefined;
+    req.connection = &connection;
+    req.reader.body_err = error.HttpChunkTruncated;
+    const resp: std.http.Client.Response = .{ .request = &req, .head = undefined };
+
+    var transfer_buf: [min_transfer_buf]u8 = undefined;
+    var redirect_buf: [redirect_buffer_len]u8 = undefined;
+    var f = undefinedFetcher(&transfer_buf, &redirect_buf);
+    const mapped = f.record(
+        mapError(error.ReadFailed, .receive_body),
+        .receive_body,
+        transport.bodyCause(&resp, pump.cause),
+        .ok,
+        pump.delivered,
+    );
+
+    try testing.expectEqual(error.ReceiveFailed, mapped);
+    const failure = f.last_failure.?;
+    try testing.expectEqual(Phase.receive_body, failure.phase);
+    try testing.expectEqual(error.HttpChunkTruncated, failure.cause);
+    try testing.expectEqual(@as(?std.http.Status, .ok), failure.status);
+    try testing.expectEqual(@as(u64, payload.len), failure.bytes_read);
+}
+
+test "a rejected url records the phase without claiming a connect failure" {
+    var transfer_buf: [min_transfer_buf]u8 = undefined;
+    var redirect_buf: [redirect_buffer_len]u8 = undefined;
+    var f = undefinedFetcher(&transfer_buf, &redirect_buf);
+    var sink_buf: [0]u8 = .{};
+    var discarding: std.Io.Writer.Discarding = .init(&sink_buf);
+
+    try testing.expectError(
+        error.BadUrl,
+        f.fetch(undefined, "ftp://example.com/list.txt", &discarding.writer),
+    );
+    try testing.expectEqual(error.BadUrl, f.last_failure.?.cause);
+}
+
+test "record returns the taxonomy member the call site raised, not a remapped one" {
+    // `error.HttpStatus` and `error.BodyTooLarge` are raised by `fetch` itself
+    // rather than by std, so `mapError` has no case for either and would fall
+    // to the phase default. `record` must not run them through it.
+    var transfer_buf: [min_transfer_buf]u8 = undefined;
+    var redirect_buf: [redirect_buffer_len]u8 = undefined;
+    var f = undefinedFetcher(&transfer_buf, &redirect_buf);
+
+    try testing.expectEqual(
+        error.HttpStatus,
+        f.record(error.HttpStatus, .receive, error.HttpStatus, .not_found, 0),
+    );
+    try testing.expectEqual(@as(?std.http.Status, .not_found), f.last_failure.?.status);
+    try testing.expectEqual(error.HttpStatus, f.last_failure.?.cause);
+
+    try testing.expectEqual(
+        error.BodyTooLarge,
+        f.record(error.BodyTooLarge, .receive, error.BodyTooLarge, .ok, 0),
+    );
+    try testing.expectEqual(error.BodyTooLarge, f.last_failure.?.cause);
+
+    // This is why: `mapError` is unchanged and does collapse both to the phase
+    // default. `fetch` must never route these two through it.
+    try testing.expectEqual(error.ReceiveFailed, mapError(error.HttpStatus, .receive));
+    try testing.expectEqual(error.ReceiveFailed, mapError(error.BodyTooLarge, .receive));
+}
+
+test "a cancelled body read records the cancellation and the progress so far" {
+    // This is the shape the expiry race leaves behind. `Select.cancelDiscard`
+    // cancels the fetch, the cancellation surfaces from the socket as a
+    // stashed `error.Canceled` under a collapsed `error.ReadFailed`, and the
+    // fetch records it on its way out. The manager then reports its own
+    // `error.Timeout` beside this record.
+    const payload = "0.0.0.0 ads.example.com\n";
+    var buffer: [payload.len]u8 = (payload ++ "").*;
+    var body: std.Io.Reader = .{
+        .vtable = std.Io.Reader.failing.vtable,
+        .buffer = &buffer,
+        .seek = 0,
+        .end = buffer.len,
+    };
+    var sink_buf: [0]u8 = .{};
+    var discarding: std.Io.Writer.Discarding = .init(&sink_buf);
+
+    var pump: Pump = .{};
+    try testing.expectError(
+        error.ReceiveFailed,
+        pumpBody(&body, &discarding.writer, max_body_bytes, &pump),
+    );
+
+    var connection = stubConnection(error.Canceled);
+    var req: std.http.Client.Request = undefined;
+    req.connection = &connection;
+    req.reader.body_err = null;
+    const resp: std.http.Client.Response = .{ .request = &req, .head = undefined };
+
+    var transfer_buf: [min_transfer_buf]u8 = undefined;
+    var redirect_buf: [redirect_buffer_len]u8 = undefined;
+    var f = undefinedFetcher(&transfer_buf, &redirect_buf);
+    // `pumpBody` maps the collapsed `error.ReadFailed` it was handed, not the
+    // unwrapped cause, so the taxonomy stays `ReceiveFailed` here. That is what
+    // this path returned before `Failure` existed, and the addendum keeps it:
+    // only the recorded cause is new. The manager reports `error.Timeout` from
+    // the expiry race anyway, so nothing reads this return value on this path.
+    const cause = transport.bodyCause(&resp, error.ReadFailed);
+    const mapped = f.record(error.ReceiveFailed, .receive_body, cause, .ok, pump.delivered);
+
+    try testing.expectEqual(error.ReceiveFailed, mapped);
+    try testing.expectEqual(Phase.receive_body, f.last_failure.?.phase);
+    try testing.expectEqual(error.Canceled, f.last_failure.?.cause);
+    try testing.expectEqual(@as(?std.http.Status, .ok), f.last_failure.?.status);
+    try testing.expectEqual(@as(u64, payload.len), f.last_failure.?.bytes_read);
+}