test/delaypipe.zig

Ref: main worktree-web-client-spec   Size: 6.7 KiB   History

//! A byte pump that adds a fixed one-way delay, so a round trip can be made
//! slow without root.
//!
//! Every latency measurement so far has needed `tc netem` on a real box,
//! which means a machine, an interface, and sudo. Prediction's whole claim
//! is that echo latency stops depending on the round trip, and demonstrating
//! that needs a round trip long enough to see — but not a real one.
//!
//! `--via` runs its argument through `/bin/sh -c`, so two of these compose
//! into a symmetric delay with a shell pipeline and nothing else:
//!
//!     mux --via "delaypipe | mux d proxy --sock S | delaypipe"
//!
//! The client's bytes go through the first before reaching the daemon, the
//! daemon's answers through the second on the way back: DELAY_MS each way,
//! 2*DELAY_MS round trip, no privileges anywhere.
//!
//! The delay is applied per chunk and serially — a chunk is read, held, and
//! written before the next read — so throughput is capped at one chunk per
//! DELAY_MS. That is fine for what this is for (a person typing, and a shell
//! answering) and would be wrong for a throughput benchmark. Said plainly
//! because the cap is invisible until something measures against it: a burst
//! arriving while a chunk is being held is coalesced into the next one, so
//! bulk output is delayed once rather than once per byte.
const std = @import("std");

pub const default_delay_ms: u64 = 150;

/// Pump `in_fd` to `out_fd`, holding every chunk for `delay_ms` first.
/// Returns at EOF or on any error. Closes neither descriptor.
pub fn pump(in_fd: std.posix.fd_t, out_fd: std.posix.fd_t, delay_ms: u64) void {
    var buf: [64 * 1024]u8 = undefined;
    while (true) {
        const n = std.posix.read(in_fd, &buf) catch return;
        if (n == 0) return;
        // Held after arrival and before delivery, which is what makes the
        // delay a property of the path rather than of the sender.
        std.Thread.sleep(delay_ms * std.time.ns_per_ms);
        writeAll(out_fd, buf[0..n]) catch return;
    }
}

fn writeAll(fd: std.posix.fd_t, data: []const u8) !void {
    var i: usize = 0;
    while (i < data.len) i += try std.posix.write(fd, data[i..]);
}

/// Interpret a `DELAY_MS` value. Absent or unparseable is the default: this
/// is a test helper, and failing to start would look like a broken transport
/// rather than a typo.
///
/// Split from the lookup so the interpretation is testable on its own —
/// std has no portable setenv, so a test that went through the environment
/// could only ever exercise the unset case, which is the one branch that
/// returns the default no matter how badly the lookup is broken.
pub fn parseDelay(raw: ?[]const u8) u64 {
    const s = raw orelse return default_delay_ms;
    return std.fmt.parseInt(u64, s, 10) catch default_delay_ms;
}

pub fn delayFromEnv() u64 {
    return parseDelay(std.posix.getenv("DELAY_MS"));
}

pub fn main() void {
    pump(std.posix.STDIN_FILENO, std.posix.STDOUT_FILENO, delayFromEnv());
}

const Harness = struct {
    to: [2]std.posix.fd_t,
    from: [2]std.posix.fd_t,
    thread: std.Thread,

    fn start(delay_ms: u64) !Harness {
        const to = try std.posix.pipe();
        const from = try std.posix.pipe();
        return .{
            .to = to,
            .from = from,
            .thread = try std.Thread.spawn(.{}, pump, .{ to[0], from[1], delay_ms }),
        };
    }

    fn finish(self: *Harness) void {
        std.posix.close(self.to[1]); // EOF ends the pump
        self.thread.join();
        std.posix.close(self.to[0]);
        std.posix.close(self.from[0]);
        std.posix.close(self.from[1]);
    }
};

test "a byte arrives no earlier than the delay, and arrives intact" {
    var h = try Harness.start(120);
    defer h.finish();

    const start = std.time.milliTimestamp();
    try writeAll(h.to[1], "x");

    var buf: [16]u8 = undefined;
    const n = try std.posix.read(h.from[0], &buf);
    const elapsed = std.time.milliTimestamp() - start;

    try std.testing.expectEqualStrings("x", buf[0..n]);
    // The lower bound is the whole point, and it is asserted with a little
    // room for clock granularity rather than at exactly 120: sleep promises
    // at least its duration, and a millisecond of rounding either way must
    // not make this flaky. A pump that forgot to wait lands near zero and
    // fails this by a hundred milliseconds, not by one.
    try std.testing.expect(elapsed >= 100);
}

test "a payload larger than one write survives the crossing in order" {
    var h = try Harness.start(5);
    defer h.finish();

    // Written in pieces, so the pump sees several chunks and each one is
    // delayed independently. Order is the property under test: a pump that
    // held chunks concurrently could deliver them out of sequence, and a
    // transport that reorders bytes is not a transport.
    try writeAll(h.to[1], "alpha-");
    try writeAll(h.to[1], "beta-");
    try writeAll(h.to[1], "gamma");

    const alloc = std.testing.allocator;
    var got: std.ArrayList(u8) = .empty;
    defer got.deinit(alloc);
    var buf: [64]u8 = undefined;
    while (got.items.len < "alpha-beta-gamma".len) {
        const n = try std.posix.read(h.from[0], &buf);
        if (n == 0) break;
        try got.appendSlice(alloc, buf[0..n]);
    }
    try std.testing.expectEqualStrings("alpha-beta-gamma", got.items);
}

test "DELAY_MS is honoured when set, and never fatal when it is nonsense" {
    // The case that matters: a value that is NOT the default, because a
    // lookup that silently failed would fall back to the default and every
    // measurement taken through this pipe would quietly be of the wrong
    // path. e2e types into a 400ms path and asserts the round trip is
    // inconsistent with 150, which is the other half of this.
    try std.testing.expectEqual(@as(u64, 250), parseDelay("250"));
    try std.testing.expectEqual(@as(u64, 0), parseDelay("0"));

    // Nonsense is the default rather than a failure to start: a helper that
    // refused to run would read as a broken transport in the suite that
    // used it, which is a much longer way round to finding a typo.
    try std.testing.expectEqual(default_delay_ms, parseDelay("abc"));
    try std.testing.expectEqual(default_delay_ms, parseDelay(""));
    try std.testing.expectEqual(default_delay_ms, parseDelay("-5"));

    // Unset, which is what a bare `delaypipe` in a pipeline takes.
    try std.testing.expectEqual(default_delay_ms, parseDelay(null));
    try std.testing.expectEqual(default_delay_ms, delayFromEnv());
}

// Forces semantic analysis of every pub decl under `zig build test`, so an
// unreferenced decl must at least compile (the silent-module-loss hazard,
// decisions.md). Pub decls only: std.meta.declarations sees nothing private.
test {
    std.testing.refAllDeclsRecursive(@This());
}