milestone 17: real deadlines, validator holes, upstream editor, trusted proxies, contract samples, badvers

src/upstream/pool.zig

@@ -1,9 +1,17 @@
-//! Priority-ordered sequential failover across upstream endpoints, with a
-//! per-attempt deadline, health tracking and backoff (PLAN §9).
+//! Priority-ordered sequential failover across upstream endpoints, with two
+//! deadlines, health tracking and backoff (PLAN §9).
 //!
 //! The pool is itself a `transport.Client`, so the handler above it sees one
 //! interface and knows nothing about how many upstreams exist.
 //!
+//! Two deadlines, nested. `timeouts.attempt` bounds one exchange against one
+//! entry; `timeouts.total` bounds the whole failover loop, every attempt
+//! together. The outer one is what the client asking the question actually
+//! waits for: without it, N unreachable upstreams cost N × attempt, and the
+//! resolver above has already given up. The outer expiry is `error.Timeout`
+//! unconditionally — the fast-failure paths (no entry enabled) return long
+//! before the budget, so an expiry always means an attempt was in flight.
+//!
 //! Two passes, not one. Pass one walks the enabled entries that health says are
 //! available. Pass two runs only when pass one attempted nothing, and skips the
 //! backoff check: every endpoint being in backoff must not turn into SERVFAIL
@@ -81,7 +89,7 @@ pub const Entry = struct {
 };
 
 /// A copy of one entry's health, taken under the mutex. Feeds
-/// `GET /api/upstream/health` in Phase 8.
+/// `GET /api/upstream/health`.
 pub const Snapshot = struct {
     /// Whole, not redacted. `GET /api/upstream/health` returns this to a session
     /// that `GET /api/upstreams` already serves the same url to in full, so
@@ -102,19 +110,30 @@ pub const Snapshot = struct {
     backoff_until: ?std.Io.Timestamp,
 };
 
+/// The two budgets, named rather than positional: they are the same type, so
+/// two parameters in a row could be swapped at a call site and still compile,
+/// and the swap would be invisible until an operator watched a query take five
+/// attempts of two and a half seconds each.
+pub const Timeouts = struct {
+    /// Bounds one exchange against one entry.
+    attempt: std.Io.Clock.Duration,
+    /// Bounds the whole failover loop.
+    total: std.Io.Clock.Duration,
+};
+
 pub const Pool = struct {
     /// Caller-owned, sorted ascending by priority in `init`.
     entries: []Entry,
     cfg: health.Config,
-    /// On the `.awake` clock, so a suspended Pi does not burn the budget.
-    attempt_timeout: std.Io.Clock.Duration,
+    /// Both on the `.awake` clock, so a suspended Pi does not burn a budget.
+    timeouts: Timeouts,
     mutex: std.Io.Mutex,
     rng: std.Random.DefaultPrng,
 
     pub fn init(
         entries: []Entry,
         cfg: health.Config,
-        attempt_timeout: std.Io.Clock.Duration,
+        timeouts: Timeouts,
         seed: u64,
     ) Pool {
         std.debug.assert(entries.len > 0);
@@ -123,7 +142,7 @@ pub const Pool = struct {
         return .{
             .entries = entries,
             .cfg = cfg,
-            .attempt_timeout = attempt_timeout,
+            .timeouts = timeouts,
             .mutex = .init,
             .rng = .init(seed),
         };
@@ -150,12 +169,52 @@ pub const Pool = struct {
     /// `response_buf` is handed to each attempt in turn, so a failed attempt
     /// may have written into it. The returned slice is only meaningful on
     /// success; on error the buffer's contents are undefined.
+    ///
+    /// The failover loop runs raced against `timeouts.total`. Losing that race
+    /// cancels the loop, which unwinds whatever attempt was in flight: the
+    /// `Entry.busy` lock is taken cancelably on purpose and released by defer,
+    /// and the health bookkeeping around a completed exchange is uncancelable,
+    /// so a canceled attempt leaves no lock held and no counter half-written.
     pub fn exchange(
         self: *Pool,
         io: std.Io,
         query: []const u8,
         response_buf: []u8,
     ) transport.ExchangeError![]u8 {
+        var outcomes: [2]LoopOutcome = undefined;
+        var race: std.Io.Select(LoopOutcome) = .init(io, &outcomes);
+        defer race.cancelDiscard();
+
+        race.concurrent(.loop, exchangeLoopLen, .{
+            self, io, query, response_buf,
+        }) catch |err| switch (err) {
+            error.ConcurrencyUnavailable => return error.SystemResources,
+        };
+        race.concurrent(.expiry, expire, .{ io, self.timeouts.total }) catch |err| switch (err) {
+            error.ConcurrencyUnavailable => return error.SystemResources,
+        };
+
+        switch (try race.await()) {
+            .loop => |result| return response_buf[0..try result],
+            .expiry => |result| {
+                // A canceled sleep means this whole task is being torn down,
+                // not that the budget ran out.
+                try result;
+                return error.Timeout;
+            },
+        }
+    }
+
+    /// The two-pass failover loop, as a raceable task. It returns the reply's
+    /// length rather than its slice for the reason `exchangeLen` in the tests
+    /// below does: `Io.concurrent` stores the future's return value, so the
+    /// bytes are read back out of the caller's `response_buf` by `exchange`.
+    fn exchangeLoopLen(
+        self: *Pool,
+        io: std.Io,
+        query: []const u8,
+        response_buf: []u8,
+    ) transport.ExchangeError!usize {
         const now = std.Io.Clock.awake.now(io);
         var last_fault: ?transport.ExchangeError = null;
         var attempted = false;
@@ -170,8 +229,9 @@ pub const Pool = struct {
                 // waiting its turn on a busy upstream has done nothing that a
                 // cancellation could corrupt, so it gives up here rather than
                 // queueing behind an exchange it will not use. The wait itself
-                // is bounded by the holder's `attempt_timeout`; the waiter's own
-                // budget only starts once it has the lock.
+                // is bounded by the holder's `timeouts.attempt`; the waiter's
+                // own budget only starts once it has the lock, and the whole
+                // loop is bounded by `timeouts.total` regardless.
                 try entry.busy.lock(io);
                 defer entry.busy.unlock(io);
 
@@ -204,7 +264,7 @@ pub const Pool = struct {
                 };
 
                 self.recordSuccess(io, entry, completed_at);
-                return response;
+                return response.len;
             }
             if (attempted) break;
         }
@@ -259,7 +319,7 @@ pub const Pool = struct {
         }) catch |err| switch (err) {
             error.ConcurrencyUnavailable => return error.SystemResources,
         };
-        race.concurrent(.expiry, expire, .{ io, self.attempt_timeout }) catch |err| switch (err) {
+        race.concurrent(.expiry, expire, .{ io, self.timeouts.attempt }) catch |err| switch (err) {
             error.ConcurrencyUnavailable => return error.SystemResources,
         };
 
@@ -312,6 +372,11 @@ const Outcome = union(enum) {
     expiry: std.Io.Cancelable!void,
 };
 
+const LoopOutcome = union(enum) {
+    loop: transport.ExchangeError!usize,
+    expiry: std.Io.Cancelable!void,
+};
+
 fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void {
     return duration.sleep(io);
 }
@@ -410,7 +475,12 @@ const test_cfg: health.Config = .{
     .max_backoff_ms = 60_000,
 };
 
-const test_timeout: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake };
+/// Long enough that nothing in a test reaches either budget unless the test is
+/// about a budget, and short enough that a stuck test still ends.
+const test_timeouts: Timeouts = .{
+    .attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
+    .total = .{ .raw = .fromSeconds(30), .clock = .awake },
+};
 
 fn expectFailureLine(expected: []const u8, url: []const u8, err: transport.ExchangeError) !void {
     var buf: [8 * safe_url.max_len]u8 = undefined;
@@ -451,7 +521,7 @@ test "Pool satisfies the Client interface" {
 
     var fake: Fake = .{ .behavior = .{ .reply = response_bytes } };
     var entries = [_]Entry{testEntry("https://a.example/dns-query", &fake, 10)};
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     const reply = try pool.client().exchange(io, query_bytes, &buf);
@@ -470,7 +540,7 @@ test "entries are tried in ascending priority order" {
         testEntry("https://high.example/dns-query", &high, 100),
         testEntry("https://low.example/dns-query", &low, 10),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     try testing.expectEqual(@as(i32, 10), entries[0].priority);
 
@@ -491,7 +561,7 @@ test "a peer fault fails over to the next entry and is recorded" {
         testEntry("https://bad.example/dns-query", &bad, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     const reply = try pool.exchange(io, query_bytes, &buf);
@@ -514,7 +584,7 @@ test "an entry in backoff is skipped while another is available" {
         testEntry("https://bad.example/dns-query", &bad, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     // Two failures reach `failure_threshold` and open a backoff window.
@@ -539,7 +609,7 @@ test "every entry in backoff is still probed" {
         testEntry("https://first.example/dns-query", &first, 10),
         testEntry("https://second.example/dns-query", &second, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     try testing.expectError(error.BadResponse, pool.exchange(io, query_bytes, &buf));
@@ -564,7 +634,7 @@ test "a local resource error short-circuits and records nothing" {
         testEntry("https://broke.example/dns-query", &broke, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     try testing.expectError(error.OutOfMemory, pool.exchange(io, query_bytes, &buf));
@@ -584,7 +654,7 @@ test "a cancellation short-circuits and records nothing" {
         testEntry("https://canceled.example/dns-query", &canceled, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     try testing.expectError(error.Canceled, pool.exchange(io, query_bytes, &buf));
@@ -606,8 +676,12 @@ test "an attempt that outruns the budget is a recorded Timeout" {
         testEntry("https://slow.example/dns-query", &slow, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    const budget: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
-    var pool: Pool = .init(&entries, test_cfg, budget, 1);
+    // A tight attempt budget under a loose total one, so the attempt deadline
+    // is the only one that can fire and the failover after it has room to run.
+    var pool: Pool = .init(&entries, test_cfg, .{
+        .attempt = .{ .raw = .fromMilliseconds(20), .clock = .awake },
+        .total = .{ .raw = .fromSeconds(30), .clock = .awake },
+    }, 1);
 
     var buf: [512]u8 = undefined;
     const reply = try pool.exchange(io, query_bytes, &buf);
@@ -619,7 +693,47 @@ test "an attempt that outruns the budget is a recorded Timeout" {
     try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes);
 }
 
-test "every entry disabled yields ConnectFailed" {
+test "two stalling upstreams cost the total budget, not one budget each" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    // Both entries stall past their own attempt budget, so without the outer
+    // race the exchange would take two attempt budgets and then some. The
+    // total is set below two attempts, which is what makes the assertion mean
+    // something: only the outer deadline can end this call in time.
+    var first: Fake = .{ .behavior = .{ .slow = .{
+        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
+        .reply = response_bytes,
+    } } };
+    var second: Fake = .{ .behavior = .{ .slow = .{
+        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
+        .reply = response_bytes,
+    } } };
+    var entries = [_]Entry{
+        testEntry("https://first.example/dns-query", &first, 10),
+        testEntry("https://second.example/dns-query", &second, 20),
+    };
+    var pool: Pool = .init(&entries, test_cfg, .{
+        .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake },
+        .total = .{ .raw = .fromMilliseconds(60), .clock = .awake },
+    }, 1);
+
+    var buf: [512]u8 = undefined;
+    const started = std.Io.Clock.awake.now(io);
+    try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf));
+    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
+
+    // Under one attempt budget, so the outer deadline is provably what fired.
+    // The bound is generous against a loaded CI box; the failure it catches is
+    // a whole extra attempt, not a scheduling hiccup.
+    try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms);
+    // The second entry was never reached: the loop was canceled inside the
+    // first attempt.
+    try testing.expectEqual(@as(usize, 0), second.calls);
+}
+
+test "every entry disabled yields ConnectFailed without waiting out the total budget" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
     const io = threaded.io();
@@ -631,10 +745,20 @@ test "every entry disabled yields ConnectFailed" {
         testEntry("https://two.example/dns-query", &two, 20),
     };
     for (&entries) |*entry| entry.enabled = false;
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    // A total budget long enough that waiting it out would be unmistakable.
+    // The outer expiry returns `error.Timeout` unconditionally, so this pins
+    // that the fast-failure path still wins its own race.
+    var pool: Pool = .init(&entries, test_cfg, .{
+        .attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
+        .total = .{ .raw = .fromSeconds(30), .clock = .awake },
+    }, 1);
 
     var buf: [512]u8 = undefined;
+    const started = std.Io.Clock.awake.now(io);
     try testing.expectError(error.ConnectFailed, pool.exchange(io, query_bytes, &buf));
+    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
+
+    try testing.expect(elapsed_ns < @as(i96, 5) * std.time.ns_per_s);
     try testing.expectEqual(@as(usize, 0), one.calls);
     try testing.expectEqual(@as(usize, 0), two.calls);
 }
@@ -650,7 +774,7 @@ test "snapshot reports the counters in pool order" {
         testEntry("https://bad.example/dns-query", &bad, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf: [512]u8 = undefined;
     _ = try pool.exchange(io, query_bytes, &buf);
@@ -705,7 +829,7 @@ test "concurrent exchanges through one entry do not overlap" {
         .reply = response_bytes,
     } } };
     var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)};
-    var pool: Pool = .init(&entries, test_cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
 
     var buf_a: [512]u8 = undefined;
     var buf_b: [512]u8 = undefined;
@@ -752,7 +876,7 @@ test "an entry that enters backoff while a task waits on it is not attempted" {
         testEntry("https://slow-bad.example/dns-query", &slow_bad, 10),
         testEntry("https://good.example/dns-query", &good, 20),
     };
-    var pool: Pool = .init(&entries, cfg, test_timeout, 1);
+    var pool: Pool = .init(&entries, cfg, test_timeouts, 1);
 
     var buf_a: [512]u8 = undefined;
     var buf_b: [512]u8 = undefined;