//! Deterministic two-endpoint network simulator (transport-simnet-plan.md
//! §A1). A tick-stepped priority queue standing between endpoints A and B:
//! each directed link injects an exponential delivery delay plus drop, dup,
//! clog, and payload-derived corruption faults, every choice a draw from the
//! run's single injected `std.Random`. The seed is the whole universe, so a
//! sync sweep replays any fault schedule exactly from its seed.
//!
//! Native harness design (no RFC clause — the sync-protocol tests carry the
//! §-citations). The faults mirror sim/faulty_file.zig's discipline: a
//! corruption is a pure function of the packet's own bytes, so a re-delivered
//! (duplicated / retried) packet corrupts identically and a fault never
//! self-heals on retry.
//!
//! Fault model is deliberately message-level, not TCP-segment-level: the
//! sans-IO boundary is a whole request/reply body. Drop = a lost round; dup =
//! the receiver processes a body twice (a §3.8 idempotence probe); corrupt =
//! a body bitflip (exercises the zlib-reject / fatal-card paths downstream).
const std = @import("std");
const assert = std.debug.assert;
/// The two ends of the wire. A directed link is named by its destination:
/// `.b` is the A->B link, `.a` is the B->A link — a bijection because there
/// are exactly two endpoints.
pub const Endpoint = enum(u1) { a, b };
/// Per-link fault knobs. Each `*_pct` is a percentage in `0..=100` rolled
/// against fresh bits from the run PRNG; all-zero (the default) is a perfect
/// wire, which is how DST phase 2 "heals" the network for the liveness check.
pub const LinkCfg = struct {
/// The whole message is lost — never scheduled (the sender sees a lost
/// round).
drop_pct: u8 = 0,
/// The message is scheduled twice, so the receiver processes it twice.
dup_pct: u8 = 0,
/// The delivered body is bitflipped (payload-derived, see `corruptInPlace`).
corrupt_pct: u8 = 0,
/// The link stalls for a bounded span; messages scheduled during the
/// stall deliver no earlier than it clears.
clog_pct: u8 = 0,
/// Mean of the exponential delivery-delay distribution, in ticks.
mean_delay_ticks: u32 = 1,
/// Explicit upper bound on any one message's delivery delay, in ticks.
max_delay_ticks: u32 = 64,
/// Explicit upper bound on one clog's duration, in ticks.
max_clog_ticks: u32 = 32,
};
/// One message handed to the receiver. `payload` is owned by the SimNet and
/// valid only until the next `deliver`/`deinit`; a caller that must retain it
/// copies it out. `corrupted` lets the driver mimic the downstream reject
/// path without re-hashing the body.
pub const Delivery = struct {
dst: Endpoint,
payload: []const u8,
corrupted: bool,
};
/// A closed error set so the CLI/driver can switch exhaustively.
pub const Error = error{ OutOfMemory, PayloadTooLarge, QueueFull };
pub const SimNet = struct {
gpa: std.mem.Allocator,
random: std.Random,
/// Simulated clock; advances only through `tick`.
now_tick: u64,
/// Monotonic enqueue counter — the FIFO tie-break among equal delivery
/// ticks, so ordering is total and replay-stable.
seq: u64,
queue: Queue,
/// Link config indexed by destination endpoint (`links[@intFromEnum(dst)]`).
links: [2]LinkCfg,
/// Per-link clog expiry tick, indexed the same way; 0 means clear.
clog_until: [2]u64,
/// Reusable delivery buffer, sized to `max_payload` at init so `deliver`
/// never allocates and stays infallible.
scratch: []u8,
/// Explicit bound on one message body — a larger payload is rejected, so
/// the simulator never allocates without limit.
pub const max_payload: usize = 1 << 20;
/// Explicit bound on messages in flight; a further enqueue is rejected.
pub const max_inflight: usize = 4096;
/// Explicit bound on the clock — `tick` past it is a programming error.
pub const max_tick: u64 = 1 << 40;
/// Bytes altered by a corruption, bounded so the flip is cheap and never
/// scans an unbounded payload.
pub const corrupt_sites: usize = 4;
const Scheduled = struct {
deliver_tick: u64,
seq: u64,
dst: Endpoint,
corrupt: bool,
payload: []u8,
};
const Queue = std.PriorityQueue(Scheduled, void, earlier);
/// Earliest `deliver_tick` first; `seq` breaks ties so the order is total.
fn earlier(_: void, x: Scheduled, y: Scheduled) std.math.Order {
if (x.deliver_tick != y.deliver_tick) return std.math.order(x.deliver_tick, y.deliver_tick);
return std.math.order(x.seq, y.seq);
}
pub fn init(gpa: std.mem.Allocator, random: std.Random) error{OutOfMemory}!SimNet {
const scratch = try gpa.alloc(u8, max_payload);
return .{
.gpa = gpa,
.random = random,
.now_tick = 0,
.seq = 0,
.queue = .empty,
.links = .{ .{}, .{} },
.clog_until = .{ 0, 0 },
.scratch = scratch,
};
}
pub fn deinit(self: *SimNet) void {
while (self.queue.pop()) |entry| self.gpa.free(entry.payload);
self.queue.deinit(self.gpa);
self.gpa.free(self.scratch);
self.* = undefined;
}
/// The current simulated tick.
pub fn clock(self: *const SimNet) u64 {
return self.now_tick;
}
/// Messages still scheduled for a future (or current) delivery.
pub fn inFlight(self: *const SimNet) usize {
return self.queue.count();
}
/// Reset every link to a perfect wire and clear all clogs — DST phase 2
/// heals the network this way before asserting convergence. In-flight
/// messages keep their already-scheduled ticks.
pub fn heal(self: *SimNet) void {
self.links = .{ .{}, .{} };
self.clog_until = .{ 0, 0 };
}
/// Advance the clock one tick. Delivery is separate (`deliver`), so a
/// caller controls exactly when it drains the messages now due.
pub fn tick(self: *SimNet) void {
assert(self.now_tick < max_tick);
self.now_tick += 1;
}
// --- core (implemented after the red run) -------------------------------
/// Offer `payload` to the link toward `dst`. Applies drop/dup and rolls
/// the corruption + clog faults now; the delay (and thus the delivery
/// tick) is drawn per scheduled copy. A dropped message returns cleanly
/// with nothing scheduled.
pub fn enqueue(self: *SimNet, dst: Endpoint, payload: []const u8) Error!void {
if (payload.len > max_payload) return error.PayloadTooLarge;
const i = @intFromEnum(dst);
const link = self.links[i];
if (drawPct(self.random, link.drop_pct)) return; // lost round, nothing scheduled
const copies: usize = if (drawPct(self.random, link.dup_pct)) 2 else 1;
// One corruption roll shared by both copies: a dup is a retry, and a
// retry must corrupt identically — so the decision cannot be re-rolled.
const corrupt = drawPct(self.random, link.corrupt_pct);
self.maybeClog(i, link);
assert(copies >= 1 and copies <= 2);
if (self.queue.count() + copies > max_inflight) return error.QueueFull;
try self.queue.ensureUnusedCapacity(self.gpa, copies);
var n: usize = 0;
while (n < copies) : (n += 1) {
const at = self.deliverTick(i, link);
const buf = try self.gpa.dupe(u8, payload);
errdefer self.gpa.free(buf);
assert(self.seq < std.math.maxInt(u64));
self.seq += 1;
self.queue.push(self.gpa, .{
.deliver_tick = at,
.seq = self.seq,
.dst = dst,
.corrupt = corrupt,
.payload = buf,
}) catch |e| {
self.gpa.free(buf);
return e;
};
}
}
/// Pop the earliest message that has come due (`deliver_tick <= now`),
/// applying its payload-derived corruption. Returns null when nothing is
/// due at the current tick.
pub fn deliver(self: *SimNet) ?Delivery {
const head = self.queue.peek() orelse return null;
if (head.deliver_tick > self.now_tick) return null;
const entry = self.queue.pop().?;
defer self.gpa.free(entry.payload);
assert(entry.payload.len <= max_payload);
const out = self.scratch[0..entry.payload.len];
@memcpy(out, entry.payload);
if (entry.corrupt) corruptInPlace(out);
return .{ .dst = entry.dst, .payload = out, .corrupted = entry.corrupt };
}
/// The absolute delivery tick for one copy on link `i`: an exponential
/// delay clamped to the link bound, floored by any active clog.
fn deliverTick(self: *SimNet, i: usize, link: LinkCfg) u64 {
assert(i < self.links.len);
const mean: f64 = @floatFromInt(link.mean_delay_ticks);
const cap: f64 = @floatFromInt(link.max_delay_ticks);
const scaled = @min(self.random.floatExp(f64) * mean, cap);
const delay: u64 = @intFromFloat(@floor(scaled));
assert(delay <= link.max_delay_ticks);
return @max(self.now_tick + delay, self.clog_until[i]);
}
/// Roll the clog fault for link `i`, extending its stall by a bounded span.
fn maybeClog(self: *SimNet, i: usize, link: LinkCfg) void {
assert(i < self.links.len);
if (link.max_clog_ticks == 0) return;
if (!drawPct(self.random, link.clog_pct)) return;
const dur = 1 + self.random.uintLessThan(u64, link.max_clog_ticks);
assert(dur >= 1 and dur <= link.max_clog_ticks);
self.clog_until[i] = @max(self.clog_until[i], self.now_tick + dur);
}
};
/// True with probability `pct/100` against the run PRNG. `0`/`100` short so
/// a perfect or certain link draws no bits (keeps replay streams aligned).
fn drawPct(random: std.Random, pct: u8) bool {
assert(pct <= 100);
if (pct == 0) return false;
if (pct == 100) return true;
return random.uintLessThan(u8, 100) < pct;
}
/// Payload-derived corruption: a pure function of the bytes, so a
/// re-delivered (duplicated) packet with the same payload corrupts to exactly
/// the same result — a fault never self-heals on retry. Flips one bit in each
/// of up to `corrupt_sites` distinct byte positions, so a non-empty body
/// always changes.
fn corruptInPlace(buf: []u8) void {
if (buf.len == 0) return;
const h = std.hash.Wyhash.hash(0xF0551_1_A1, buf);
const sites = @min(buf.len, SimNet.corrupt_sites);
const start: usize = @intCast(h % buf.len);
var n: usize = 0;
// Distinct consecutive indices (count <= len), each flipped exactly once,
// so no flip cancels another: the body is guaranteed to change.
while (n < sites) : (n += 1) {
const idx = (start + n) % buf.len;
const bit: u3 = @truncate(h >> @intCast(n * 3));
buf[idx] ^= (@as(u8, 1) << bit);
}
}
// ----------------------------------------------------------------------------
// Tests: every fault schedule is a property of the simulator's own
// determinism (seeds are literals here — the enclosing seed reproduces the
// run), never an RFC oracle. Harness code carries no §-citation.
// ----------------------------------------------------------------------------
const testing = std.testing;
/// Drive the net to quiescence, folding every delivery into a stable digest
/// (its dst, corrupted flag, and bytes) plus a count. The tick guard is a
/// hard bound: all test enqueues happen at tick 0, so every delivery is due
/// by `max_delay_ticks`/`max_clog_ticks` (both <= 64).
fn drainDigest(net: *SimNet) struct { count: usize, hash: u64 } {
var hasher = std.hash.Wyhash.init(0);
var count: usize = 0;
var guard: usize = 0;
while (net.inFlight() > 0) {
while (net.deliver()) |d| {
hasher.update(&.{ @intFromEnum(d.dst), @intFromBool(d.corrupted) });
hasher.update(d.payload);
count += 1;
}
net.tick();
guard += 1;
assert(guard <= 4096);
}
while (net.deliver()) |d| {
hasher.update(&.{ @intFromEnum(d.dst), @intFromBool(d.corrupted) });
hasher.update(d.payload);
count += 1;
}
return .{ .count = count, .hash = hasher.final() };
}
test "net §A1: the same seed replays an identical delivery/drop/dup/corrupt sequence" {
const seed: u64 = 0xA1;
var digests: [2]u64 = undefined;
var counts: [2]usize = undefined;
for (0..2) |run| {
var prng = std.Random.DefaultPrng.init(seed);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .drop_pct = 30, .dup_pct = 30, .corrupt_pct = 30, .clog_pct = 20, .mean_delay_ticks = 6 };
net.links[@intFromEnum(Endpoint.a)] = .{ .drop_pct = 10, .dup_pct = 40, .corrupt_pct = 20, .mean_delay_ticks = 3 };
for (0..64) |k| {
var body: [16]u8 = undefined;
@memset(&body, @intCast(k & 0xff));
const dst: Endpoint = if (k % 2 == 0) .b else .a;
try net.enqueue(dst, &body);
}
const d = drainDigest(&net);
digests[run] = d.hash;
counts[run] = d.count;
}
try testing.expectEqual(counts[0], counts[1]);
try testing.expectEqual(digests[0], digests[1]);
}
test "net §A1: payload-derived corruption reproduces identically on a duplicated packet" {
var prng = std.Random.DefaultPrng.init(0xC0);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
// A certain dup and certain corruption, no drop: exactly two deliveries,
// both corrupted from the same body.
net.links[@intFromEnum(Endpoint.b)] = .{ .dup_pct = 100, .corrupt_pct = 100, .mean_delay_ticks = 0 };
const original = "the-quick-brown-fox-0123456789";
try net.enqueue(.b, original);
// Drain the first copy and stash its (corrupted) bytes before the shared
// scratch buffer is overwritten by the second.
var first: [original.len]u8 = undefined;
var got_first = false;
var got_second = false;
var second_matches = false;
var guard: usize = 0;
while (net.inFlight() > 0 or !got_second) {
while (net.deliver()) |d| {
try testing.expect(d.corrupted);
try testing.expect(!std.mem.eql(u8, d.payload, original)); // it really changed
if (!got_first) {
@memcpy(&first, d.payload);
got_first = true;
} else {
second_matches = std.mem.eql(u8, d.payload, &first);
got_second = true;
}
}
if (got_second) break;
net.tick();
guard += 1;
assert(guard <= 4096);
}
try testing.expect(got_first and got_second);
try testing.expect(second_matches); // retry corrupts identically, never self-heals
}
test "net §A1: drop probability is honored — none pass at 100%, all pass at 0%" {
const n = 100;
// drop everything
{
var prng = std.Random.DefaultPrng.init(1);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .drop_pct = 100 };
for (0..n) |_| try net.enqueue(.b, "x");
try testing.expectEqual(@as(usize, 0), net.inFlight());
try testing.expectEqual(@as(usize, 0), drainDigest(&net).count);
}
// drop nothing
{
var prng = std.Random.DefaultPrng.init(2);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .mean_delay_ticks = 2 };
for (0..n) |_| try net.enqueue(.b, "x");
try testing.expectEqual(@as(usize, n), drainDigest(&net).count);
}
// a middling rate loses some but not all
{
var prng = std.Random.DefaultPrng.init(3);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .drop_pct = 50 };
for (0..n) |_| try net.enqueue(.b, "x");
const delivered = drainDigest(&net).count;
try testing.expect(delivered > 0 and delivered < n);
}
}
test "net §A1: duplication delivers a message twice" {
var prng = std.Random.DefaultPrng.init(7);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .dup_pct = 100, .mean_delay_ticks = 3 };
for (0..20) |_| try net.enqueue(.b, "dup");
try testing.expectEqual(@as(usize, 40), drainDigest(&net).count);
}
test "net §A1: delivery delay orders messages through the priority queue" {
var prng = std.Random.DefaultPrng.init(9);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .mean_delay_ticks = 8 };
for (0..50) |_| try net.enqueue(.b, "ordered");
var last: u64 = 0;
var count: usize = 0;
var guard: usize = 0;
while (net.inFlight() > 0) {
while (net.deliver()) |_| {
try testing.expect(net.clock() >= last); // non-decreasing delivery tick
last = net.clock();
count += 1;
}
net.tick();
guard += 1;
assert(guard <= 4096);
}
try testing.expectEqual(@as(usize, 50), count);
}
test "net §A1: the in-flight queue is bounded" {
var prng = std.Random.DefaultPrng.init(11);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
// Large delay so nothing drains; no drop/dup so each enqueue adds one.
net.links[@intFromEnum(Endpoint.b)] = .{ .mean_delay_ticks = 1000 };
for (0..SimNet.max_inflight) |_| try net.enqueue(.b, "q");
try testing.expectError(error.QueueFull, net.enqueue(.b, "q"));
}
test "net §A1: an oversized payload is rejected" {
var prng = std.Random.DefaultPrng.init(13);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
const big = try testing.allocator.alloc(u8, SimNet.max_payload + 1);
defer testing.allocator.free(big);
try testing.expectError(error.PayloadTooLarge, net.enqueue(.b, big));
}
test "net §A1: a healed link delivers every message intact" {
var prng = std.Random.DefaultPrng.init(17);
var net = try SimNet.init(testing.allocator, prng.random());
defer net.deinit();
net.links[@intFromEnum(Endpoint.b)] = .{ .drop_pct = 100 };
net.heal(); // phase-2 heal: faults off
for (0..30) |_| try net.enqueue(.b, "clean");
var count: usize = 0;
var guard: usize = 0;
while (net.inFlight() > 0) {
while (net.deliver()) |d| {
try testing.expect(!d.corrupted);
try testing.expectEqualStrings("clean", d.payload);
count += 1;
}
net.tick();
guard += 1;
assert(guard <= 4096);
}
try testing.expectEqual(@as(usize, 30), count);
}
test "net §A1: corruptInPlace is a deterministic, altering pure function" {
const a = "payload-under-corruption";
var x: [a.len]u8 = a.*;
var y: [a.len]u8 = a.*;
corruptInPlace(&x);
corruptInPlace(&y);
try testing.expectEqualSlices(u8, &x, &y); // same input -> same output
try testing.expect(!std.mem.eql(u8, &x, a)); // actually changed something
}