mokhtar/nxdns
1
0
Fork 0
Code Pull Requests Actions Packages Releases 21 Activity

upstream: overlap attribution test builds its schedule with gates

Browse Source 
The test "overlapping exchanges each report the entry that answered that call" failed once on a loaded release runner and passed on rerun. It started two exchanges back to back and relied on a 50 ms failing stall to keep the first in flight until the second arrived; nothing checked that the two calls were ever in flight together, and the failing schedule was never observed.

Two Fake behaviours, hold and hold_fail, wait on a test-owned std.Io.Semaphore and then reply or fail. The test holds both calls inside the first entry, waits on the entry's in_flight count for one and then two, releases the failing call so it fails over while the other is still held, then releases the other. The fake draws its behaviour before it raises in_flight, so a count of one also fixes which call holds which gate. Each gate has a deferred post registered after its task's deferred await, so any early return releases the held call instead of deadlocking. No sleep remains in the test; every assertion is kept. No pool bug was found. Addendum in specs/milestone-31.md.
This commit is contained in:
mokhtar
2026-09-08 01:14:26 +02:00
parent 85b8be50a0
commit 299d7af99c
2 changed files with 70 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/upstream/pool.zig
+62 -26
View File
@@ -773,6 +773,16 @@ const Fake = struct {
/// Sleep, then fail. Holds the entry's lock long enough for a second
/// task to queue on it before the failure opens a backoff window.
slow_fail: struct { duration: std.Io.Clock.Duration, err: transport.ExchangeError },
/// Wait for the gate, then reply. A test that needs two calls provably
/// in flight together cannot get that from a duration: a loaded
/// scheduler can start the second call after the first one's sleep
/// expired, and no poll can recover an overlap that already passed.
/// The gate makes the test, not the clock, decide when a call returns.
hold: struct { gate: *std.Io.Semaphore, reply: []const u8 },
/// Wait for the gate, then fail. The failing half of `hold`, with the
/// same constraint: the test releases the call once it has observed
/// whatever had to be true while the call was still in flight.
hold_fail: struct { gate: *std.Io.Semaphore, err: transport.ExchangeError },
};
fn exchangeFn(
@@ -787,11 +797,11 @@ const Fake = struct {
// entry, so a test can tell the two apart.
selected.* = "fake://leaf";
const self: *Fake = @ptrCast(@alignCast(ptr));
const entrants = self.in_flight.fetchAdd(1, .acq_rel) + 1;
defer _ = self.in_flight.fetchSub(1, .acq_rel);
_ = self.peak_in_flight.fetchMax(entrants, .acq_rel);
_ = self.calls.fetchAdd(1, .acq_rel);
// The behaviour is taken before `in_flight` rises: a test that waits
// on the count to know a call is inside this fake also needs to know
// which behaviour that call drew, or a second call could still take
// the first behaviour from under it.
const behavior = behavior: {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
@@ -802,6 +812,9 @@ const Fake = struct {
}
break :behavior current;
};
const entrants = self.in_flight.fetchAdd(1, .acq_rel) + 1;
defer _ = self.in_flight.fetchSub(1, .acq_rel);
_ = self.peak_in_flight.fetchMax(entrants, .acq_rel);
switch (behavior) {
.reply => |bytes| return copy(bytes, response_buf),
.fail => |err| return err,
@@ -813,6 +826,14 @@ const Fake = struct {
try slow.duration.sleep(io);
return slow.err;
},
.hold => |held| {
try held.gate.wait(io);
return copy(held.reply, response_buf);
},
.hold_fail => |held| {
try held.gate.wait(io);
return held.err;
},
}
}
@@ -2018,27 +2039,21 @@ test "overlapping exchanges each report the entry that answered that call" {
defer threaded.deinit();
const io = threaded.io();
// Two gates, one per call the first entry takes. Nothing in this test
// sleeps: the test releases each call itself, so the schedule it asserts
// on is the schedule it built.
var gate_a: std.Io.Semaphore = .{};
var gate_b: std.Io.Semaphore = .{};
// The first entry has room for both tasks and treats them differently: it
// fails whichever reaches it first and answers the other, slowly enough
// that the two calls are in flight together. One failure is under
// `test_cfg`'s threshold of two, so no backoff steers the failed-over task
// away and the two calls end on different entries.
// fails whichever reaches it first and answers the other. One failure is
// under `test_cfg`'s threshold of two, so no backoff steers the failed-over
// task away and the two calls end on different entries.
var first_entry: Fake = .{
.behavior = .{ .slow_fail = .{
.duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
.err = error.ConnectFailed,
} },
.then = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(100), .clock = .awake },
.reply = alt_response_bytes,
} },
.behavior = .{ .hold_fail = .{ .gate = &gate_a, .err = error.ConnectFailed } },
.then = .{ .hold = .{ .gate = &gate_b, .reply = alt_response_bytes } },
};
// Slow too, so the failed-over call is still in flight while the other call
// is being answered — a shared identity would be overwritten under it.
var second_entry: Fake = .{ .behavior = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
.reply = response_bytes,
} } };
var second_entry: Fake = .{ .behavior = .{ .reply = response_bytes } };
var entries = [_]Entry{
testEntrySlots(divergent_first_url, &first_entry, 10, 2),
testEntry(divergent_second_url, &second_entry, 20),
@@ -2052,17 +2067,37 @@ test "overlapping exchanges each report the entry that answered that call" {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = first.await(io) catch Attributed.discarded;
// Registered right after the task and after its await, so it runs before
// the await: any early return below (a failed wait, a skipped second
// task, a failed assertion) releases the held call first and the await
// then returns, instead of the test deadlocking on a gate nobody posted.
// Extra permits on an already-released gate are harmless.
defer gate_a.post(io);
// A count of one means the first call is inside the entry with its
// behaviour already drawn, so it holds gate A and the second call will draw
// gate B.
try awaitInFlight(io, &first_entry, 1);
var second = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_b }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = second.await(io) catch Attributed.discarded;
defer gate_b.post(io);
// Both tasks are now inside the first entry, the first on `hold_fail` and
// the second on `hold`. That is the overlap every assertion below rests
// on, and it is established here rather than assumed.
try awaitInFlight(io, &first_entry, 2);
// Release the first call only. It fails, fails over to the second entry
// and is answered there while the second call is still held inside the
// first entry, so a shared identity would be overwritten under it.
gate_a.post(io);
const result_a = try first.await(io);
gate_b.post(io);
const result_b = try second.await(io);
// Which task lands where depends on the order they take the first entry's
// lock, so the claim is over the pair: one identity each, and each one
// matching the bytes that call received.
// The first task reached the first entry first, but the claim stays over
// the pair: one identity each, and each one matching the bytes that call
// received.
try testing.expect(!std.mem.eql(u8, result_a.selected.?, result_b.selected.?));
try expectAnsweredByReporter(result_a, &buf_a);
try expectAnsweredByReporter(result_b, &buf_b);
@@ -2070,7 +2105,8 @@ test "overlapping exchanges each report the entry that answered that call" {
try testing.expectEqual(@as(usize, 2), first_entry.calls.load(.acquire));
try testing.expectEqual(@as(usize, 1), second_entry.calls.load(.acquire));
// Both tasks were admitted to the first entry at once, which is what makes
// them concurrent rather than serialized behind one permit.
// them concurrent rather than serialized behind one permit. The wait above
// already required this; the counter records that it held to the end.
try testing.expectEqual(@as(u32, 2), first_entry.peak_in_flight.load(.acquire));
try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
try testing.expectEqual(@as(u64, 1), entries[0].health.total_successes);