fabric: the pipeline stage link, a TCP ring that carries each round's plan and hidden rows, with a shared key (Linux and macOS, #2005 step 1)

build.zig

@@ -96,6 +96,7 @@ pub fn build(b: *std.Build) void {
     @import("zig/build/kernel_symbols.zig").tests(b, test_step);
     @import("zig/build/checkpoint_security.zig").tests(b, test_step);
     @import("zig/build/quant_geometry_host.zig").steps(b, test_step);
+    @import("zig/build/stage.zig").steps(b, test_step);
     if (b.graph.host.result.os.tag == .macos) {
         const step5_mods = metalEngineModules(b, b.graph.host, .Debug, draft_ids);
         const empty = b.addWriteFiles().add("step5-compile-only.metallib", "");

zig/build/stage.zig

@@ -0,0 +1,22 @@
+//! The pipeline stage link (zig/src/fabric/stage): its module, host tests, and compile checks for Linux and macOS.
+const std = @import("std");
+
+pub fn module(b: *std.Build, target: std.Build.ResolvedTarget, optimize: std.builtin.OptimizeMode) *std.Build.Module {
+    return b.createModule(.{ .root_source_file = b.path("zig/src/fabric/stage/stage.zig"), .target = target, .optimize = optimize, .link_libc = true });
+}
+
+pub fn steps(b: *std.Build, all: *std.Build.Step) void {
+    const host = b.graph.host;
+    if (host.result.os.tag != .linux and host.result.os.tag != .macos) return;
+    const run = b.addRunArtifact(b.addTest(.{ .root_module = module(b, host, .Debug) }));
+    run.has_side_effects = true; // loopback sockets
+    const step = b.step("test-stage", "Pipeline stage link: rings of two and three ranks on loopback, refusals and failures");
+    step.dependOn(&run.step);
+    all.dependOn(step);
+    const compile = b.step("test-stage-compile", "Compile the stage link's tests for Linux and macOS on aarch64");
+    for ([_]std.Target.Os.Tag{ .linux, .macos }) |os| {
+        const target = b.resolveTargetQuery(.{ .cpu_arch = .aarch64, .os_tag = os });
+        compile.dependOn(&b.addTest(.{ .root_module = module(b, target, .Debug) }).step);
+    }
+    all.dependOn(compile);
+}

zig/src/fabric/stage/config.zig

@@ -0,0 +1,128 @@
+//! A stage rank's ring settings from TF_STAGE_RANK, _WORLD, _PEERS, _SECRET_FILE and _TIMEOUT_S.
+const std = @import("std");
+const ring = @import("ring.zig");
+const secret = @import("secret.zig");
+
+/// The most ranks a ring takes.
+pub const most_ranks = 64;
+
+pub const Error = secret.Error || error{ MissingStageSetting, BadStageRank, BadStageWorld, BadStagePeers, BadStageTimeout };
+
+/// Parsed settings; `peers` points into `slots` and into the TF_STAGE_PEERS text.
+pub const Settings = struct {
+    rank: u32,
+    world: u32,
+    timeout_ns: u64,
+    key: secret.Key = undefined,
+    slots: [most_ranks]ring.Peer = undefined,
+    n: usize = 0,
+
+    pub fn peers(s: *const Settings) []const ring.Peer {
+        return s.slots[0..s.n];
+    }
+
+    /// The ring's options for these settings and the checkpoint's and layout's `fingerprint`.
+    pub fn options(s: *const Settings, fingerprint: [32]u8) ring.Options {
+        return .{ .rank = s.rank, .world = s.world, .peers = s.peers(), .fingerprint = fingerprint, .key = s.key, .timeout_ns = s.timeout_ns };
+    }
+};
+
+/// The settings named by `get` (an environment lookup), refused by name when one is missing or malformed.
+pub fn parse(get: anytype, s: *Settings) Error!void {
+    const rank = get.lookup("TF_STAGE_RANK") orelse return error.MissingStageSetting;
+    const world = get.lookup("TF_STAGE_WORLD") orelse return error.MissingStageSetting;
+    const list = get.lookup("TF_STAGE_PEERS") orelse return error.MissingStageSetting;
+    const path = get.lookup("TF_STAGE_SECRET_FILE") orelse return error.MissingStageSetting;
+    s.* = .{ .rank = 0, .world = 0, .timeout_ns = 60 * std.time.ns_per_s };
+    s.world = std.fmt.parseInt(u32, world, 10) catch return error.BadStageWorld;
+    if (s.world < 2 or s.world > most_ranks) return error.BadStageWorld;
+    s.rank = std.fmt.parseInt(u32, rank, 10) catch return error.BadStageRank;
+    if (s.rank >= s.world) return error.BadStageRank;
+    if (get.lookup("TF_STAGE_TIMEOUT_S")) |v| {
+        const secs = std.fmt.parseInt(u32, v, 10) catch return error.BadStageTimeout;
+        if (secs == 0) return error.BadStageTimeout;
+        s.timeout_ns = @as(u64, secs) * std.time.ns_per_s;
+    }
+    var it = std.mem.splitScalar(u8, list, ',');
+    while (it.next()) |entry| {
+        if (s.n == most_ranks) return error.BadStagePeers;
+        const colon = std.mem.lastIndexOfScalar(u8, entry, ':') orelse return error.BadStagePeers;
+        const host = entry[0..colon];
+        const port = std.fmt.parseInt(u16, entry[colon + 1 ..], 10) catch return error.BadStagePeers;
+        if (host.len == 0 or port == 0) return error.BadStagePeers;
+        s.slots[s.n] = .{ .host = host, .port = port };
+        s.n += 1;
+    }
+    if (s.n != s.world) return error.BadStagePeers;
+    var name: [4096]u8 = undefined;
+    if (path.len >= name.len) return error.SecretUnreadable;
+    @memcpy(name[0..path.len], path);
+    name[path.len] = 0;
+    s.key = try secret.fromFile(name[0..path.len :0]);
+}
+
+/// The process environment, for `parse`.
+pub const Env = struct {
+    pub fn lookup(_: Env, name: [:0]const u8) ?[]const u8 {
+        const v = std.c.getenv(name) orelse return null;
+        return std.mem.span(v);
+    }
+};
+
+const Fake = struct {
+    rank: ?[]const u8 = "1",
+    world: ?[]const u8 = "3",
+    peers: ?[]const u8 = "192.0.2.1:7100,stage-b.example:7101,127.0.0.1:7102",
+    timeout: ?[]const u8 = null,
+    secret_file: ?[]const u8 = "/nonexistent/stage-secret",
+    pub fn lookup(f: Fake, name: [:0]const u8) ?[]const u8 {
+        if (std.mem.eql(u8, name, "TF_STAGE_RANK")) return f.rank;
+        if (std.mem.eql(u8, name, "TF_STAGE_WORLD")) return f.world;
+        if (std.mem.eql(u8, name, "TF_STAGE_PEERS")) return f.peers;
+        if (std.mem.eql(u8, name, "TF_STAGE_TIMEOUT_S")) return f.timeout;
+        if (std.mem.eql(u8, name, "TF_STAGE_SECRET_FILE")) return f.secret_file;
+        return null;
+    }
+};
+
+/// An owner-only secret file for one test, at `path`.
+fn writeSecret(path: [:0]const u8) !void {
+    _ = std.c.unlink(path);
+    const fd = std.c.open(path, .{ .ACCMODE = .WRONLY, .CREAT = true, .EXCL = true }, @as(std.c.mode_t, 0o600));
+    if (fd < 0) return error.SecretUnreadable;
+    defer _ = std.c.close(fd);
+    _ = std.c.write(fd, "a stage ring's test secret\n", 27);
+}
+
+test "stage config: a three-rank ring's settings parse, with its timeout and key" {
+    var buf: [64]u8 = undefined;
+    const path = try std.fmt.bufPrintSentinel(&buf, "/tmp/tf-stage-config-{d}", .{std.c.getpid()}, 0);
+    try writeSecret(path);
+    defer _ = std.c.unlink(path);
+    var s: Settings = undefined;
+    try parse(Fake{ .timeout = "5", .secret_file = path }, &s);
+    try std.testing.expectEqual(try secret.fromBytes("a stage ring's test secret"), s.key);
+    try std.testing.expectEqual(@as(u32, 1), s.rank);
+    try std.testing.expectEqual(@as(u32, 3), s.world);
+    try std.testing.expectEqual(5 * std.time.ns_per_s, s.timeout_ns);
+    try std.testing.expectEqualStrings("stage-b.example", s.peers()[1].host);
+    try std.testing.expectEqual(@as(u16, 7102), s.peers()[2].port);
+    try parse(Fake{ .secret_file = path }, &s);
+    try std.testing.expectEqual(60 * std.time.ns_per_s, s.timeout_ns);
+}
+
+test "stage config: a missing or malformed setting is refused by name" {
+    var s: Settings = undefined;
+    try std.testing.expectError(error.MissingStageSetting, parse(Fake{ .peers = null }, &s));
+    try std.testing.expectError(error.MissingStageSetting, parse(Fake{ .secret_file = null }, &s));
+    try std.testing.expectError(error.SecretUnreadable, parse(Fake{}, &s));
+    try std.testing.expectError(error.BadStageWorld, parse(Fake{ .world = "1" }, &s));
+    try std.testing.expectError(error.BadStageWorld, parse(Fake{ .world = "65" }, &s));
+    try std.testing.expectError(error.BadStageRank, parse(Fake{ .rank = "3" }, &s));
+    try std.testing.expectError(error.BadStageRank, parse(Fake{ .rank = "-1" }, &s));
+    try std.testing.expectError(error.BadStageTimeout, parse(Fake{ .timeout = "0" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b:2" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b:0,c:3" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,:2,c:3" }, &s));
+    try std.testing.expectError(error.BadStagePeers, parse(Fake{ .peers = "a:1,b,c:3" }, &s));
+}

zig/src/fabric/stage/frame.zig

@@ -0,0 +1,59 @@
+//! A stage link's frame header: 32 little-endian bytes (magic, version, kind, round, rows, payload bytes).
+const std = @import("std");
+
+pub const magic: u32 = 0x5453_4654; // "TFST" in its little-endian bytes (54 46 53 54)
+pub const version: u16 = 1;
+pub const size = 32;
+
+/// What a frame carries: the link's hello, a round's plan, its hidden rows, or a clean stop.
+pub const Kind = enum(u8) { hello = 1, plan = 2, rows = 3, stop = 4 };
+
+pub const Header = struct {
+    kind: Kind,
+    /// The round this frame belongs to; each link's rounds run 1, 2, 3, ... in order.
+    round: u64,
+    /// The rows a `rows` frame carries (0 for the others).
+    rows: u32,
+    /// Payload bytes after the header.
+    bytes: u64,
+};
+
+pub const Error = error{ BadMagic, BadVersion, BadKind, TooLarge };
+
+pub fn encode(h: Header) [size]u8 {
+    var out: [size]u8 = @splat(0);
+    std.mem.writeInt(u32, out[0..4], magic, .little);
+    std.mem.writeInt(u16, out[4..6], version, .little);
+    out[6] = @backingInt(h.kind);
+    std.mem.writeInt(u64, out[8..16], h.round, .little);
+    std.mem.writeInt(u32, out[16..20], h.rows, .little);
+    std.mem.writeInt(u64, out[24..32], h.bytes, .little);
+    return out;
+}
+
+/// A header from `raw`, refused unless it is this link's magic and version, a known kind, and at most `cap` bytes.
+pub fn decode(raw: *const [size]u8, cap: u64) Error!Header {
+    if (std.mem.readInt(u32, raw[0..4], .little) != magic) return error.BadMagic;
+    if (std.mem.readInt(u16, raw[4..6], .little) != version) return error.BadVersion;
+    const kind = std.enums.fromInt(Kind, raw[6]) orelse return error.BadKind;
+    const bytes = std.mem.readInt(u64, raw[24..32], .little);
+    if (bytes > cap) return error.TooLarge;
+    return .{ .kind = kind, .round = std.mem.readInt(u64, raw[8..16], .little), .rows = std.mem.readInt(u32, raw[16..20], .little), .bytes = bytes };
+}
+
+test "stage frame: a header round-trips with TFST on the wire, and a wrong magic, version, kind or size is refused" {
+    const h: Header = .{ .kind = .rows, .round = 0x0102_0304_0506_0708, .rows = 4096, .bytes = 80 << 20 };
+    const raw = encode(h);
+    try std.testing.expectEqualStrings("TFST", raw[0..4]);
+    try std.testing.expectEqual(h, try decode(&raw, 80 << 20));
+    try std.testing.expectError(error.TooLarge, decode(&raw, (80 << 20) - 1));
+    var bad = raw;
+    bad[0] ^= 1;
+    try std.testing.expectError(error.BadMagic, decode(&bad, 1 << 30));
+    bad = raw;
+    bad[4] = 2;
+    try std.testing.expectError(error.BadVersion, decode(&bad, 1 << 30));
+    bad = raw;
+    bad[6] = 9;
+    try std.testing.expectError(error.BadKind, decode(&bad, 1 << 30));
+}