From a6dd239777245942b1eb656203910a1e5cc74137 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 05:49:44 -0700
Subject: [PATCH 01/15] dsv41: a rank whose peer dies ends with exit 75 (peer
 link watch)

When rank 1 died while rank 0 was idle, rank 0 stayed up: /health said
200 and the next request hung without an error or exit. When rank 1 died
during a decode, the live streams hung too: the round's GPU waits on the
RDMA ring got no error completion, so the abort word was never set
(release fault F3).

- peer_watch.zig: a thread that polls the ranks' TCP links for the
  peer's end (POLLRDHUP, HUP, ERR; it reads nothing). On a peer that is
  gone without a done frame, it aborts the RDMA ring (Ring.abort: every
  wait gives up, so a live round fails and its requests get the error)
  and, with TF_DS_EXIT_ON_FAULT=1, ends the process with exit 75 after
  TF_DS_EXIT_GRACE plus 2 s (the lane host's own fault exit comes first
  when a round ran). It waits 1 s first, for rank 0's done frame and
  the close that follows it.
- link.zig: a keepalive (10 s idle, 3 probes 3 s apart), so a peer
  whose node vanished shows within about 20 s.
- lanes.zig: rank 0 watches its followers' links from Lanes.init and
  stops the watch before its done frame; a follower watches its link to
  rank 0 while it follows.
---
 zig/src/families/dsv41/dsv41.zig      |   2 +
 zig/src/families/dsv41/lanes.zig      |  23 ++++
 zig/src/families/dsv41/link.zig       |   8 +-
 zig/src/families/dsv41/peer_watch.zig | 149 ++++++++++++++++++++++++++
 4 files changed, 181 insertions(+), 1 deletion(-)
 create mode 100644 zig/src/families/dsv41/peer_watch.zig

diff --git a/zig/src/families/dsv41/dsv41.zig b/zig/src/families/dsv41/dsv41.zig
index 38ba0e44d8..d1439eba28 100644
--- a/zig/src/families/dsv41/dsv41.zig
+++ b/zig/src/families/dsv41/dsv41.zig
@@ -16,6 +16,7 @@ pub const weights = @import("weights.zig");
 pub const engram_io = @import("engram_io.zig");
 pub const engram_aio = @import("engram_aio.zig");
 pub const link = @import("link.zig");
+pub const peer_watch = @import("peer_watch.zig");
 pub const comm = @import("comm.zig");
 pub const tri = @import("tri.zig");
 pub const tri_basic = @import("tri_basic.zig");
@@ -64,6 +65,7 @@ test {
     refAll(engram_io);
     refAll(engram_aio);
     refAll(link);
+    refAll(peer_watch);
     refAll(comm);
     refAll(tri);
     refAll(tri_basic);
diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 3e01638f55..5e035d6e54 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -20,6 +20,7 @@ const round = @import("round.zig");
 const draft = @import("draft.zig");
 const sampling = @import("sampling.zig");
 const link = @import("link.zig");
+const peer_watch = @import("peer_watch.zig");
 const cuda = @import("cuda");
 
 const be = lanes.backend;
@@ -285,6 +286,7 @@ pub const Lanes = struct {
     gpa: std.mem.Allocator,
     m: *Model,
     peer: ?link.Link, // rank 1 (null: one rank); every follower gets each frame (Model.followers)
+    watch: ?*peer_watch.Watch = null, // the links to the followers, watched for a rank that dies (peer_watch.zig)
     vision: ?*@import("vision.zig").Vision = null, // rank 0's tower (a vision checkpoint served with its kit's vision/)
     step: u64 = 0,
     wire: Wire = .{},
@@ -307,6 +309,7 @@ pub const Lanes = struct {
 
     pub fn init(gpa: std.mem.Allocator, m: *Model) Lanes {
         var ln: Lanes = .{ .gpa = gpa, .m = m, .peer = if (m.world > 1) m.peer() else null };
+        if (m.world > 1) ln.watch = watchLinks(gpa, m, m.followers());
         // the round costs by rows: ROUND_MS[min(rows, len) - 1] (the served table, or TF_DS_ROUND_MS)
         const t = m.round_ms;
         for (&ln.wc, 0..) |*c, i| c.* = .{ .width = @intCast(i + 1), .ms = t[@min(i, t.len - 1)] };
@@ -321,6 +324,7 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
+        if (self.watch) |w| w.stop(self.gpa); // before the done frame: the links' close is expected from now on
         if (self.peer != null) self.send(.done) catch {};
         self.streams.deinit(self.gpa);
         for (&self.seqs) |*q| q.deinit(self.gpa);
@@ -876,9 +880,28 @@ fn refused(err: anyerror) void {
 /// A follower's verify times with Model.prof (a profile on ranks 1-3: rounds, issue and wall ns summed).
 pub var follow_prof: struct { rounds: u64 = 0, enqueue: u64 = 0, forward: u64 = 0 } = .{};
 
+/// A watch on `links` (null: it could not start; the ranks then run as before, without it).
+fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*peer_watch.Watch {
+    var fds: [3]std.posix.fd_t = undefined;
+    var n: usize = 0;
+    for (links) |l| if (l) |x| if (n < fds.len) {
+        fds[n] = x.fd;
+        n += 1;
+    };
+    if (n == 0) return null;
+    return peer_watch.Watch.start(gpa, fds[0..n], m.ring, m.rank) catch |err| {
+        std.debug.print("[tensorfold] the peer link watch did not start ({s})
+", .{@errorName(err)});
+        return null;
+    };
+}
+
 /// Rank 1 (each of ranks 1-3 on the four-node split): the primitives rank 0 sends, run in order until it says done.
 pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     const peer = m.peer();
+    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig)
+    const watch = watchLinks(gpa, m, &.{peer});
+    defer if (watch) |w| w.stop(gpa);
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);
diff --git a/zig/src/families/dsv41/link.zig b/zig/src/families/dsv41/link.zig
index e6a75ecae3..15afe0f552 100644
--- a/zig/src/families/dsv41/link.zig
+++ b/zig/src/families/dsv41/link.zig
@@ -15,12 +15,18 @@ pub const Link = struct {
 
     /// TCP_NODELAY, and SO_BUSY_POLL: a read polls the NIC's receive queue itself (up to 50 us) instead of waiting for
     /// its interrupt, whose core may be leaving a deep idle state (node to node, 64-byte frames: round trip 1.0 ms
-    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL).
+    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL). A keepalive (10 s idle, then 3 probes 3 s
+    /// apart) shows a peer whose node vanished (no FIN) within about 20 s, to a read and to peer_watch.zig.
     fn tune(fd: c.fd_t) void {
         const one: c_int = 1;
         const busy_us: c_int = 50;
         _ = c.setsockopt(fd, c.IPPROTO.TCP, std.posix.TCP.NODELAY, std.mem.asBytes(&one), @sizeOf(c_int));
         _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.BUSY_POLL, std.mem.asBytes(&busy_us), @sizeOf(c_int));
+        _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.KEEPALIVE, std.mem.asBytes(&one), @sizeOf(c_int));
+        for ([_][2]u32{ .{ std.posix.TCP.KEEPIDLE, 10 }, .{ std.posix.TCP.KEEPINTVL, 3 }, .{ std.posix.TCP.KEEPCNT, 3 } }) |o| {
+            const v: c_int = @intCast(o[1]);
+            _ = c.setsockopt(fd, c.IPPROTO.TCP, o[0], std.mem.asBytes(&v), @sizeOf(c_int));
+        }
     }
 
     fn addr(ip: [4]u8, port: u16) c.sockaddr.in {
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
new file mode 100644
index 0000000000..b93ab69c9a
--- /dev/null
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -0,0 +1,149 @@
+//! A watch on the ranks' TCP links (release fault F3). A rank waits for its peer's frames and RDMA writes without a
+//! limit, so when the peer died while this rank was idle, or during a round whose waits got no error completion, the
+//! rank stayed up: /health said 200 and every request hung. The watch reads nothing: it polls each link for the
+//! peer's end (its process died: the kernel closed the socket) or a failed keepalive (its node is gone, link.zig).
+//! Then it aborts the RDMA ring (every wait gives up, so a live round fails and its requests get the error) and, with
+//! TF_DS_EXIT_ON_FAULT=1, ends the process with 75 after TF_DS_EXIT_GRACE, so a supervisor restarts both ranks.
+const std = @import("std");
+const rdma = @import("rdma.zig");
+const posix = std.posix;
+const Allocator = std.mem.Allocator;
+
+/// POLLRDHUP (Linux): the peer closed its end.
+const rdhup: i16 = 0x2000;
+/// The peer closed its end, or the connection failed (a keepalive or a send got no answer).
+const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR;
+const max_links = 3;
+
+pub const Watch = struct {
+    fds: [max_links]posix.fd_t = @splat(-1),
+    n: usize = 0,
+    ring: ?*rdma.Ring,
+    rank: u32,
+    running: std.atomic.Value(bool) = .init(true),
+    orderly: std.atomic.Value(bool) = .init(false),
+    /// Set when the watch found a peer gone (before it acts).
+    lost: std.atomic.Value(bool) = .init(false),
+    thread: ?std.Thread = null,
+
+    /// A watch thread on `fds`: rank 0's links to its followers, or a follower's one link to rank 0.
+    pub fn start(gpa: Allocator, fds: []const posix.fd_t, ring: ?*rdma.Ring, rank: u32) !*Watch {
+        const w = try gpa.create(Watch);
+        errdefer gpa.destroy(w);
+        w.* = .{ .ring = ring, .rank = rank, .n = @min(fds.len, max_links) };
+        @memcpy(w.fds[0..w.n], fds[0..w.n]);
+        w.thread = try std.Thread.spawn(.{}, loop, .{w});
+        return w;
+    }
+
+    /// An orderly stop (rank 0 sends its done frame, or a follower takes it): a link that closes after it is no fault.
+    pub fn stop(w: *Watch, gpa: Allocator) void {
+        w.orderly.store(true, .release);
+        w.running.store(false, .release);
+        if (w.thread) |t| t.join();
+        gpa.destroy(w);
+    }
+
+    fn quiet(w: *Watch) bool {
+        return w.orderly.load(.acquire) or !w.running.load(.acquire);
+    }
+
+    /// Waits up to `ms`; true if an orderly stop came in that time.
+    fn stopped(w: *Watch, ms: u64) bool {
+        var left = ms;
+        while (left > 0) : (left -|= 100) {
+            if (w.quiet()) return true;
+            const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
+            _ = std.c.nanosleep(&ts, null);
+        }
+        return w.quiet();
+    }
+
+    fn loop(w: *Watch) void {
+        var pfds: [max_links]posix.pollfd = undefined;
+        for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
+        while (!w.quiet()) {
+            const k = posix.poll(pfds[0..w.n], 200) catch continue;
+            if (k == 0) continue;
+            for (pfds[0..w.n]) |p| if (p.revents & gone != 0) return w.peerGone();
+        }
+    }
+
+    fn peerGone(w: *Watch) void {
+        // rank 0's done frame and the close of its link come together: a second for this rank to take the stop
+        if (w.stopped(1000)) return;
+        w.lost.store(true, .release);
+        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the RDMA ring's waits are aborted\n", .{w.rank});
+        if (w.ring) |r| r.abort("a peer rank is gone (its link closed)");
+        const grace_ms = exitGrace() orelse return;
+        // a live round fails first and the lane host ends the process itself after the same grace
+        if (w.stopped(grace_ms + 2000)) return;
+        std.debug.print("[tensorfold] rank {d}: exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{w.rank});
+        // a stream may hold kernels that wait on the gone rank (NCCL's), which a normal exit would join
+        std.c._exit(75);
+    }
+};
+
+/// TF_DS_EXIT_ON_FAULT=1: the grace in ms (TF_DS_EXIT_GRACE seconds, default 5), as lane_host.faultExit reads it.
+fn exitGrace() ?u64 {
+    const on = std.c.getenv("TF_DS_EXIT_ON_FAULT") orelse return null;
+    if (!std.mem.eql(u8, std.mem.span(on), "1")) return null;
+    const g = std.c.getenv("TF_DS_EXIT_GRACE") orelse return 5000;
+    const s = std.fmt.parseFloat(f64, std.mem.span(g)) catch return 5000;
+    if (!(s >= 0) or s > 600) return 5000;
+    return @intFromFloat(s * 1000);
+}
+
+const Link = @import("link.zig").Link;
+
+fn testPort() u16 {
+    return 29500 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(std.testing.io, .awake).nanoseconds, 400)));
+}
+
+test "a peer that closes its link without a done frame is found within two seconds" {
+    const port = testPort();
+    const Peer = struct {
+        fn run(p: u16) void {
+            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
+            l.send("one frame") catch {};
+            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
+            _ = std.c.nanosleep(&ts, null);
+            l.close(); // the process "dies": its end closes
+        }
+    };
+    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, null, 0);
+    defer w.stop(std.testing.allocator);
+    t.join();
+    // a frame waiting unread is no end: the watch asks for the peer's close alone
+    var waited: u32 = 0;
+    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) {
+        const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
+        _ = std.c.nanosleep(&ts, null);
+    }
+    try std.testing.expect(w.lost.load(.acquire));
+}
+
+test "an orderly stop before the link closes is no fault" {
+    const port = testPort() + 400;
+    const Peer = struct {
+        fn run(p: u16) void {
+            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
+            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
+            _ = std.c.nanosleep(&ts, null);
+            l.close();
+        }
+    };
+    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, null, 1);
+    w.orderly.store(true, .release); // the done frame came: the close that follows is expected
+    t.join();
+    const ts: std.c.timespec = .{ .sec = 1, .nsec = 500 * std.time.ns_per_ms };
+    _ = std.c.nanosleep(&ts, null);
+    try std.testing.expect(!w.lost.load(.acquire));
+    w.stop(std.testing.allocator);
+}

From 31466020f17d8bcedace295e1e1d31aa1e509f9f Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 05:53:44 -0700
Subject: [PATCH 02/15] dsv41: a follower stops its peer watch only on the done
 frame

An error unwind (rank 0's link gone while the follower reads it) left through a deferred stop, which marked the watch orderly and cancelled its exit 75. The watch now stays running on an error unwind, so its exit 75 still ends the process if the normal exit hangs on kernels that wait on the gone rank.
---
 zig/src/families/dsv41/lanes.zig | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 5e035d6e54..fe2f4b34b9 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -899,9 +899,9 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
 /// Rank 1 (each of ranks 1-3 on the four-node split): the primitives rank 0 sends, run in order until it says done.
 pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     const peer = m.peer();
-    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig)
+    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); only the
+    // done frame stops the watch: an error unwinds with it still running, so its exit 75 stays if the exit hangs
     const watch = watchLinks(gpa, m, &.{peer});
-    defer if (watch) |w| w.stop(gpa);
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);
@@ -1045,7 +1045,10 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
                     b.free();
                 }
             },
-            .done => return,
+            .done => {
+                if (watch) |w| w.stop(gpa);
+                return;
+            },
         }
     }
 }

From 84a0e16a2260e7c1cfd29289493171b9f88e2b5e Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 06:22:03 -0700
Subject: [PATCH 03/15] dsv41: the peer watch aborts every RDMA ring and stops
 when a start fails

World 4 keeps its rings in rdma_rings (all4, col, pair) or neighbor_gather (first, second), not in ring, so a dead peer aborted no GPU wait there. The watch now takes every ring the rank's waits read. A start that fails after Lanes.init (a bad TF_DS_FIX_K, a vision load error) closed the model with the watch still polling its links and holding its rings; native.open now stops the watch on that path.
---
 zig/src/families/dsv41/lanes.zig      | 26 ++++++++++++++++++++++++--
 zig/src/families/dsv41/native.zig     |  1 +
 zig/src/families/dsv41/peer_watch.zig | 21 +++++++++++++--------
 3 files changed, 38 insertions(+), 10 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index fe2f4b34b9..8fcd19b6de 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -21,6 +21,7 @@ const draft = @import("draft.zig");
 const sampling = @import("sampling.zig");
 const link = @import("link.zig");
 const peer_watch = @import("peer_watch.zig");
+const rdma = @import("rdma.zig");
 const cuda = @import("cuda");
 
 const be = lanes.backend;
@@ -317,6 +318,12 @@ pub const Lanes = struct {
         return ln;
     }
 
+    /// Stops the peer link watch (an orderly stop, or a start that fails after init: it holds the model's links and rings).
+    pub fn stopWatch(self: *Lanes) void {
+        if (self.watch) |w| w.stop(self.gpa);
+        self.watch = null;
+    }
+
     /// Tells rank 1 to stop, then frees the host state (the model is the caller's).
     pub fn deinit(self: *Lanes) void {
         for (self.snaps.items) |sn| {
@@ -324,7 +331,7 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
-        if (self.watch) |w| w.stop(self.gpa); // before the done frame: the links' close is expected from now on
+        self.stopWatch(); // before the done frame: the links' close is expected from now on
         if (self.peer != null) self.send(.done) catch {};
         self.streams.deinit(self.gpa);
         for (&self.seqs) |*q| q.deinit(self.gpa);
@@ -889,7 +896,22 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
         n += 1;
     };
     if (n == 0) return null;
-    return peer_watch.Watch.start(gpa, fds[0..n], m.ring, m.rank) catch |err| {
+    // every ring a GPU wait of this rank reads: TP2's, world 4's three, or the neighbor gather's two
+    var rings: [6]*rdma.Ring = undefined;
+    var k: usize = 0;
+    if (m.ring) |r| {
+        rings[k] = r;
+        k += 1;
+    }
+    if (m.rdma_rings) |rr| for ([_]*rdma.Ring{ rr.all4, rr.col, rr.pair }) |r| {
+        rings[k] = r;
+        k += 1;
+    };
+    if (m.neighbor_gather) |ng| for ([_]*rdma.Ring{ ng.first, ng.second }) |r| {
+        rings[k] = r;
+        k += 1;
+    };
+    return peer_watch.Watch.start(gpa, fds[0..n], rings[0..k], m.rank) catch |err| {
         std.debug.print("[tensorfold] the peer link watch did not start ({s})
 ", .{@errorName(err)});
         return null;
diff --git a/zig/src/families/dsv41/native.zig b/zig/src/families/dsv41/native.zig
index 07074e2fa4..7494118ab9 100644
--- a/zig/src/families/dsv41/native.zig
+++ b/zig/src/families/dsv41/native.zig
@@ -158,6 +158,7 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [
     });
     errdefer st.m.close();
     st.lanes = lanes_mod.Lanes.init(gpa, st.m);
+    errdefer st.lanes.stopWatch(); // a failed start closes the model: the watch must not keep its links and rings
     if (getenv("TF_DS_FIX_K")) |t| st.lanes.fix_k = std.fmt.parseInt(usize, t, 10) catch return error.BadFixK;
     if (getenv("TF_DS_SERVED_K")) |t| st.lanes.served_k = if (std.mem.eql(u8, t, "1") or std.mem.eql(u8, t, "all")) .all else if (std.mem.eql(u8, t, "solo")) .solo else .off;
     // a vision checkpoint's tower on this rank, from the kit's vision/ (torch's attention cubin, the routing bias)
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index b93ab69c9a..c59e1cd306 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -14,11 +14,14 @@ const rdhup: i16 = 0x2000;
 /// The peer closed its end, or the connection failed (a keepalive or a send got no answer).
 const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR;
 const max_links = 3;
+/// TP2's ring, world 4's three rings (ring2d.zig), or the neighbor gather's two (neighbor.zig).
+const max_rings = 6;
 
 pub const Watch = struct {
     fds: [max_links]posix.fd_t = @splat(-1),
     n: usize = 0,
-    ring: ?*rdma.Ring,
+    rings: [max_rings]*rdma.Ring = undefined,
+    n_rings: usize = 0,
     rank: u32,
     running: std.atomic.Value(bool) = .init(true),
     orderly: std.atomic.Value(bool) = .init(false),
@@ -26,12 +29,14 @@ pub const Watch = struct {
     lost: std.atomic.Value(bool) = .init(false),
     thread: ?std.Thread = null,
 
-    /// A watch thread on `fds`: rank 0's links to its followers, or a follower's one link to rank 0.
-    pub fn start(gpa: Allocator, fds: []const posix.fd_t, ring: ?*rdma.Ring, rank: u32) !*Watch {
+    /// A watch thread on `fds` (rank 0's links to its followers, or a follower's one link to rank 0) that aborts
+    /// `rings` (every RDMA ring this rank's GPU waits on) when a peer is gone.
+    pub fn start(gpa: Allocator, fds: []const posix.fd_t, rings: []const *rdma.Ring, rank: u32) !*Watch {
         const w = try gpa.create(Watch);
         errdefer gpa.destroy(w);
-        w.* = .{ .ring = ring, .rank = rank, .n = @min(fds.len, max_links) };
+        w.* = .{ .rank = rank, .n = @min(fds.len, max_links), .n_rings = @min(rings.len, max_rings) };
         @memcpy(w.fds[0..w.n], fds[0..w.n]);
+        @memcpy(w.rings[0..w.n_rings], rings[0..w.n_rings]);
         w.thread = try std.Thread.spawn(.{}, loop, .{w});
         return w;
     }
@@ -73,8 +78,8 @@ pub const Watch = struct {
         // rank 0's done frame and the close of its link come together: a second for this rank to take the stop
         if (w.stopped(1000)) return;
         w.lost.store(true, .release);
-        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the RDMA ring's waits are aborted\n", .{w.rank});
-        if (w.ring) |r| r.abort("a peer rank is gone (its link closed)");
+        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
+        for (w.rings[0..w.n_rings]) |r| r.abort("a peer rank is gone (its link closed)");
         const grace_ms = exitGrace() orelse return;
         // a live round fails first and the lane host ends the process itself after the same grace
         if (w.stopped(grace_ms + 2000)) return;
@@ -114,7 +119,7 @@ test "a peer that closes its link without a done frame is found within two secon
     const t = try std.Thread.spawn(.{}, Peer.run, .{port});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, null, 0);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
     defer w.stop(std.testing.allocator);
     t.join();
     // a frame waiting unread is no end: the watch asks for the peer's close alone
@@ -139,7 +144,7 @@ test "an orderly stop before the link closes is no fault" {
     const t = try std.Thread.spawn(.{}, Peer.run, .{port});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, null, 1);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
     w.orderly.store(true, .release); // the done frame came: the close that follows is expected
     t.join();
     const ts: std.c.timespec = .{ .sec = 1, .nsec = 500 * std.time.ns_per_ms };

From 56b1a68d68c4954263a30d2c758709e34808c6a8 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 06:55:55 -0700
Subject: [PATCH 04/15] dsv41: a follower that ends on an error releases its
 rings from the peer watch

Model.close frees the RDMA rings after follow() returns an error, while the
watch can still abort them. The follower now calls Watch.release on the
error path: it aborts the rings (so close does not wait on a gone peer) and
forgets them under a lock, and keeps the exit 75 armed.
---
 zig/src/families/dsv41/lanes.zig      |  2 ++
 zig/src/families/dsv41/peer_watch.zig | 50 +++++++++++++++++++++++++--
 2 files changed, 50 insertions(+), 2 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 8fcd19b6de..6a92809ff3 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -924,6 +924,8 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); only the
     // done frame stops the watch: an error unwinds with it still running, so its exit 75 stays if the exit hangs
     const watch = watchLinks(gpa, m, &.{peer});
+    // Model.close frees the rings next: the watch lets them go first (and aborts their waits, so close does not hang)
+    errdefer if (watch) |w| w.release("this rank ends on an error");
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index c59e1cd306..def7a8aa02 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -27,6 +27,8 @@ pub const Watch = struct {
     orderly: std.atomic.Value(bool) = .init(false),
     /// Set when the watch found a peer gone (before it acts).
     lost: std.atomic.Value(bool) = .init(false),
+    /// Held while the watch aborts the rings, so release() never frees them under it (a spin: it is taken at most twice).
+    rings_busy: std.atomic.Value(bool) = .init(false),
     thread: ?std.Thread = null,
 
     /// A watch thread on `fds` (rank 0's links to its followers, or a follower's one link to rank 0) that aborts
@@ -49,6 +51,19 @@ pub const Watch = struct {
         gpa.destroy(w);
     }
 
+    /// A rank that ends on an error (its rings are freed next): aborts the rings and forgets them, so the watch never
+    /// reads freed rings. It keeps the links watched and the exit 75 armed: an error is no orderly stop.
+    pub fn release(w: *Watch, why: []const u8) void {
+        w.lockRings();
+        defer w.rings_busy.store(false, .release);
+        for (w.rings[0..w.n_rings]) |r| r.abort(why);
+        w.n_rings = 0;
+    }
+
+    fn lockRings(w: *Watch) void {
+        while (w.rings_busy.swap(true, .acquire)) std.atomic.spinLoopHint();
+    }
+
     fn quiet(w: *Watch) bool {
         return w.orderly.load(.acquire) or !w.running.load(.acquire);
     }
@@ -78,8 +93,12 @@ pub const Watch = struct {
         // rank 0's done frame and the close of its link come together: a second for this rank to take the stop
         if (w.stopped(1000)) return;
         w.lost.store(true, .release);
-        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
-        for (w.rings[0..w.n_rings]) |r| r.abort("a peer rank is gone (its link closed)");
+        {
+            w.lockRings();
+            defer w.rings_busy.store(false, .release);
+            std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
+            for (w.rings[0..w.n_rings]) |r| r.abort("a peer rank is gone (its link closed)");
+        }
         const grace_ms = exitGrace() orelse return;
         // a live round fails first and the lane host ends the process itself after the same grace
         if (w.stopped(grace_ms + 2000)) return;
@@ -131,6 +150,33 @@ test "a peer that closes its link without a done frame is found within two secon
     try std.testing.expect(w.lost.load(.acquire));
 }
 
+test "a released watch forgets its rings and still finds the peer gone" {
+    const port = testPort() + 800;
+    const Peer = struct {
+        fn run(p: u16) void {
+            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
+            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
+            _ = std.c.nanosleep(&ts, null);
+            l.close();
+        }
+    };
+    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
+    defer w.stop(std.testing.allocator);
+    w.release("the test ends on an error");
+    try std.testing.expectEqual(@as(usize, 0), w.n_rings);
+    t.join();
+    var waited: u32 = 0;
+    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) {
+        const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
+        _ = std.c.nanosleep(&ts, null);
+    }
+    // an error is no orderly stop: the watch still acts on the peer's end
+    try std.testing.expect(w.lost.load(.acquire));
+}
+
 test "an orderly stop before the link closes is no fault" {
     const port = testPort() + 400;
     const Peer = struct {

From 0e92350a06f1bfc7d29b23c0bc957bb49653d01a Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 07:00:26 -0700
Subject: [PATCH 05/15] dsv41: fix a line break inside a string literal in
 watchLinks

---
 zig/src/families/dsv41/lanes.zig | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 6a92809ff3..4c14f960c7 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -912,8 +912,7 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
         k += 1;
     };
     return peer_watch.Watch.start(gpa, fds[0..n], rings[0..k], m.rank) catch |err| {
-        std.debug.print("[tensorfold] the peer link watch did not start ({s})
-", .{@errorName(err)});
+        std.debug.print("[tensorfold] the peer link watch did not start ({s})\n", .{@errorName(err)});
         return null;
     };
 }

From bbb861b7af24080c5ebc62292808168b3de9928a Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 07:23:25 -0700
Subject: [PATCH 06/15] dsv41: rank 0 keeps its peer watch armed when it ends
 after a peer fault

Lanes.deinit stopped the watch before the done frame on every teardown, so
an error teardown after a follower died marked the loss orderly and cancelled
the exit 75. Now the watch stops only after a done frame that goes out with no
link end seen. Otherwise rank 0 releases the rings from the watch and holds
(at most the grace plus 4 s) for the watch's exit 75.
---
 zig/src/families/dsv41/lanes.zig      | 16 ++++++++++++++--
 zig/src/families/dsv41/peer_watch.zig | 27 ++++++++++++++++++++++++++-
 2 files changed, 40 insertions(+), 3 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 4c14f960c7..19b8d35ddc 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -331,8 +331,20 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
-        self.stopWatch(); // before the done frame: the links' close is expected from now on
-        if (self.peer != null) self.send(.done) catch {};
+        // a peer that the watch found gone before the done frame is a fault, and so is a done frame that cannot go out
+        const w = self.watch;
+        self.watch = null;
+        const lost = if (w) |x| x.faulted() else false;
+        const sent = if (self.peer != null) (if (self.send(.done)) |_| true else |_| false) else true;
+        if (w) |x| {
+            if (!lost and sent) {
+                x.stop(self.gpa); // within the watch's one-second wait: the followers' close after the done frame is expected
+            } else {
+                // the model frees the rings next; the exit 75 stays armed and comes before this rank's own return
+                x.release("rank 0 ends after a peer fault");
+                x.hold();
+            }
+        }
         self.streams.deinit(self.gpa);
         for (&self.seqs) |*q| q.deinit(self.gpa);
         self.wire.deinit(self.gpa);
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index def7a8aa02..834f7d63cf 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -25,8 +25,12 @@ pub const Watch = struct {
     rank: u32,
     running: std.atomic.Value(bool) = .init(true),
     orderly: std.atomic.Value(bool) = .init(false),
+    /// Set when the poll found a link's end (before the one-second wait for an orderly stop).
+    seen: std.atomic.Value(bool) = .init(false),
     /// Set when the watch found a peer gone (before it acts).
     lost: std.atomic.Value(bool) = .init(false),
+    /// Set when the watch thread ends.
+    finished: std.atomic.Value(bool) = .init(false),
     /// Held while the watch aborts the rings, so release() never frees them under it (a spin: it is taken at most twice).
     rings_busy: std.atomic.Value(bool) = .init(false),
     thread: ?std.Thread = null,
@@ -60,6 +64,22 @@ pub const Watch = struct {
         w.n_rings = 0;
     }
 
+    /// True when the watch found a link's end: a stop after it is no orderly stop.
+    pub fn faulted(w: *Watch) bool {
+        return w.seen.load(.acquire);
+    }
+
+    /// A released watch on a rank that ends after a peer fault: with TF_DS_EXIT_ON_FAULT=1, waits (at most the grace
+    /// plus 4 s) for the watch's exit 75, so the rank's normal return does not end the process with another code first.
+    pub fn hold(w: *Watch) void {
+        const grace_ms = exitGrace() orelse return;
+        var left = grace_ms + 4000;
+        while (left > 0 and !w.finished.load(.acquire)) : (left -|= 100) {
+            const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
+            _ = std.c.nanosleep(&ts, null);
+        }
+    }
+
     fn lockRings(w: *Watch) void {
         while (w.rings_busy.swap(true, .acquire)) std.atomic.spinLoopHint();
     }
@@ -80,12 +100,16 @@ pub const Watch = struct {
     }
 
     fn loop(w: *Watch) void {
+        defer w.finished.store(true, .release);
         var pfds: [max_links]posix.pollfd = undefined;
         for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
         while (!w.quiet()) {
             const k = posix.poll(pfds[0..w.n], 200) catch continue;
             if (k == 0) continue;
-            for (pfds[0..w.n]) |p| if (p.revents & gone != 0) return w.peerGone();
+            for (pfds[0..w.n]) |p| if (p.revents & gone != 0) {
+                w.seen.store(true, .release);
+                return w.peerGone();
+            };
         }
     }
 
@@ -148,6 +172,7 @@ test "a peer that closes its link without a done frame is found within two secon
         _ = std.c.nanosleep(&ts, null);
     }
     try std.testing.expect(w.lost.load(.acquire));
+    try std.testing.expect(w.faulted()); // a stop after it (rank 0's teardown) keeps the fault
 }
 
 test "a released watch forgets its rings and still finds the peer gone" {

From c41445ff952bac2b0185bc61d16e5132f0db01a9 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 07:31:54 -0700
Subject: [PATCH 07/15] dsv41: one teardown for the peer watch (Watch.end)

Every way a rank ends now calls Watch.end before the model frees its links
and rings: rank 0's Lanes.deinit, a follower's done frame or error, and a
start that fails after Lanes.init. On an error it aborts every RDMA ring
(TP=2 and TP=4), stops and joins the watch thread, frees the watch, and,
when a peer's link is gone and TF_DS_EXIT_ON_FAULT=1, exits 75 before
Model.close. Watch.stop, release and hold are gone.

Link sends use MSG_NOSIGNAL: a send to a gone follower fails with PeerDown
instead of a SIGPIPE that ends rank 0 with no exit 75.

Tests: a failed start with the peer up, an idle follower whose rank 0 dies,
a sending leader whose follower dies, a follower that gets PeerDown before
the watch acts, and a done frame then the peer's close.
---
 zig/src/families/dsv41/lanes.zig      |  33 ++--
 zig/src/families/dsv41/link.zig       |   3 +-
 zig/src/families/dsv41/native.zig     |   2 +-
 zig/src/families/dsv41/peer_watch.zig | 236 +++++++++++++++-----------
 4 files changed, 154 insertions(+), 120 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 19b8d35ddc..08c683e6f7 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -318,9 +318,10 @@ pub const Lanes = struct {
         return ln;
     }
 
-    /// Stops the peer link watch (an orderly stop, or a start that fails after init: it holds the model's links and rings).
-    pub fn stopWatch(self: *Lanes) void {
-        if (self.watch) |w| w.stop(self.gpa);
+    /// Ends the peer link watch (Watch.end, the one teardown: before the model frees its links and rings). `failed`:
+    /// with a follower gone and TF_DS_EXIT_ON_FAULT=1, it does not return (exit 75).
+    pub fn endWatch(self: *Lanes, how: peer_watch.How) void {
+        if (self.watch) |w| _ = w.end(self.gpa, how);
         self.watch = null;
     }
 
@@ -331,20 +332,11 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
-        // a peer that the watch found gone before the done frame is a fault, and so is a done frame that cannot go out
-        const w = self.watch;
-        self.watch = null;
-        const lost = if (w) |x| x.faulted() else false;
+        // a follower that the watch found gone before the done frame is a fault, and so is a done frame that cannot go
+        // out; the followers' close after a done frame that went out is expected (within the watch's one-second wait)
+        const lost = if (self.watch) |w| w.faulted() else false;
         const sent = if (self.peer != null) (if (self.send(.done)) |_| true else |_| false) else true;
-        if (w) |x| {
-            if (!lost and sent) {
-                x.stop(self.gpa); // within the watch's one-second wait: the followers' close after the done frame is expected
-            } else {
-                // the model frees the rings next; the exit 75 stays armed and comes before this rank's own return
-                x.release("rank 0 ends after a peer fault");
-                x.hold();
-            }
-        }
+        self.endWatch(if (!lost and sent) .done else .failed);
         self.streams.deinit(self.gpa);
         for (&self.seqs) |*q| q.deinit(self.gpa);
         self.wire.deinit(self.gpa);
@@ -932,11 +924,10 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
 /// Rank 1 (each of ranks 1-3 on the four-node split): the primitives rank 0 sends, run in order until it says done.
 pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     const peer = m.peer();
-    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); only the
-    // done frame stops the watch: an error unwinds with it still running, so its exit 75 stays if the exit hangs
+    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); the done
+    // frame ends the watch orderly, an error ends it failed (exit 75 when rank 0 is gone), before Model.close
     const watch = watchLinks(gpa, m, &.{peer});
-    // Model.close frees the rings next: the watch lets them go first (and aborts their waits, so close does not hang)
-    errdefer if (watch) |w| w.release("this rank ends on an error");
+    errdefer if (watch) |w| _ = w.end(gpa, .failed);
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);
@@ -1081,7 +1072,7 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
                 }
             },
             .done => {
-                if (watch) |w| w.stop(gpa);
+                if (watch) |w| _ = w.end(gpa, .done);
                 return;
             },
         }
diff --git a/zig/src/families/dsv41/link.zig b/zig/src/families/dsv41/link.zig
index 15afe0f552..28d5794779 100644
--- a/zig/src/families/dsv41/link.zig
+++ b/zig/src/families/dsv41/link.zig
@@ -70,7 +70,8 @@ pub const Link = struct {
     fn writeAll(l: Link, bytes: []const u8) !void {
         var done: usize = 0;
         while (done < bytes.len) {
-            const n = c.send(l.fd, bytes.ptr + done, bytes.len - done, 0);
+            // NOSIGNAL: a send to a gone peer fails with PeerDown (no SIGPIPE, which would end the rank with no exit 75)
+            const n = c.send(l.fd, bytes.ptr + done, bytes.len - done, c.MSG.NOSIGNAL);
             if (n <= 0) {
                 if (n < 0 and c.errno(n) == .INTR) continue;
                 return error.PeerDown;
diff --git a/zig/src/families/dsv41/native.zig b/zig/src/families/dsv41/native.zig
index 7494118ab9..db1ba482a3 100644
--- a/zig/src/families/dsv41/native.zig
+++ b/zig/src/families/dsv41/native.zig
@@ -158,7 +158,7 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [
     });
     errdefer st.m.close();
     st.lanes = lanes_mod.Lanes.init(gpa, st.m);
-    errdefer st.lanes.stopWatch(); // a failed start closes the model: the watch must not keep its links and rings
+    errdefer st.lanes.endWatch(.failed); // a failed start closes the model next: the watch ends first
     if (getenv("TF_DS_FIX_K")) |t| st.lanes.fix_k = std.fmt.parseInt(usize, t, 10) catch return error.BadFixK;
     if (getenv("TF_DS_SERVED_K")) |t| st.lanes.served_k = if (std.mem.eql(u8, t, "1") or std.mem.eql(u8, t, "all")) .all else if (std.mem.eql(u8, t, "solo")) .solo else .off;
     // a vision checkpoint's tower on this rank, from the kit's vision/ (torch's attention cubin, the routing bias)
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index 834f7d63cf..3080a47fe1 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -2,8 +2,9 @@
 //! limit, so when the peer died while this rank was idle, or during a round whose waits got no error completion, the
 //! rank stayed up: /health said 200 and every request hung. The watch reads nothing: it polls each link for the
 //! peer's end (its process died: the kernel closed the socket) or a failed keepalive (its node is gone, link.zig).
-//! Then it aborts the RDMA ring (every wait gives up, so a live round fails and its requests get the error) and, with
+//! Then it aborts the RDMA rings (every wait gives up, so a live round fails and its requests get the error) and, with
 //! TF_DS_EXIT_ON_FAULT=1, ends the process with 75 after TF_DS_EXIT_GRACE, so a supervisor restarts both ranks.
+//! Watch.end is the one teardown of a watch: every way a rank ends calls it before the rings are freed.
 const std = @import("std");
 const rdma = @import("rdma.zig");
 const posix = std.posix;
@@ -17,6 +18,24 @@ const max_links = 3;
 /// TP2's ring, world 4's three rings (ring2d.zig), or the neighbor gather's two (neighbor.zig).
 const max_rings = 6;
 
+/// How a rank's work with its peers ended, as its caller knows it.
+pub const How = enum {
+    /// rank 0 sent its done frame, or a follower took it: a link that closes after it is expected
+    done,
+    /// an error (a failed start, a failed send or receive, a failed round)
+    failed,
+};
+
+/// What Watch.end found.
+pub const Outcome = enum {
+    /// the done frame: no fault
+    orderly,
+    /// an error while every peer link is still up: this rank's own fault (its caller reports it)
+    failed,
+    /// an error with a peer's link gone: with TF_DS_EXIT_ON_FAULT=1, end does not return (exit 75)
+    peer_fault,
+};
+
 pub const Watch = struct {
     fds: [max_links]posix.fd_t = @splat(-1),
     n: usize = 0,
@@ -24,14 +43,11 @@ pub const Watch = struct {
     n_rings: usize = 0,
     rank: u32,
     running: std.atomic.Value(bool) = .init(true),
-    orderly: std.atomic.Value(bool) = .init(false),
     /// Set when the poll found a link's end (before the one-second wait for an orderly stop).
     seen: std.atomic.Value(bool) = .init(false),
-    /// Set when the watch found a peer gone (before it acts).
+    /// Set when the watch thread acted on a peer gone (it aborted the rings).
     lost: std.atomic.Value(bool) = .init(false),
-    /// Set when the watch thread ends.
-    finished: std.atomic.Value(bool) = .init(false),
-    /// Held while the watch aborts the rings, so release() never frees them under it (a spin: it is taken at most twice).
+    /// Held while a thread aborts the rings (a spin: it is taken at most twice).
     rings_busy: std.atomic.Value(bool) = .init(false),
     thread: ?std.Thread = null,
 
@@ -47,48 +63,58 @@ pub const Watch = struct {
         return w;
     }
 
-    /// An orderly stop (rank 0 sends its done frame, or a follower takes it): a link that closes after it is no fault.
-    pub fn stop(w: *Watch, gpa: Allocator) void {
-        w.orderly.store(true, .release);
+    /// True when the poll found a link's end: rank 0 reads it before its done frame (a done frame after it is no
+    /// orderly stop).
+    pub fn faulted(w: *Watch) bool {
+        return w.seen.load(.acquire);
+    }
+
+    /// The one teardown, for every way a rank ends (rank 0's Lanes.deinit, a follower's done frame or error, a start
+    /// that fails after the watch started), always before the caller frees the rings (Model.close):
+    /// 1. on an error, it aborts every RDMA ring, so no GPU wait of this rank waits on a gone peer;
+    /// 2. it stops the watch thread and joins it, so no thread reads the rings after end returns;
+    /// 3. it frees the watch (the caller frees the rings and links next);
+    /// 4. on an error with a peer's link gone and TF_DS_EXIT_ON_FAULT=1, it ends the process with 75 (a normal exit
+    ///    could wait on the gone rank in Model.close).
+    pub fn end(w: *Watch, gpa: Allocator, how: How) Outcome {
+        if (how == .failed) w.abortRings("this rank ends on an error");
         w.running.store(false, .release);
         if (w.thread) |t| t.join();
+        const out: Outcome = switch (how) {
+            .done => .orderly,
+            .failed => if (w.seen.load(.acquire) or w.linkGone()) .peer_fault else .failed,
+        };
+        const rank = w.rank;
         gpa.destroy(w);
+        if (out == .peer_fault and exitGrace() != null) {
+            std.debug.print("[tensorfold] rank {d}: a peer rank is gone; exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{rank});
+            std.c._exit(75);
+        }
+        return out;
     }
 
-    /// A rank that ends on an error (its rings are freed next): aborts the rings and forgets them, so the watch never
-    /// reads freed rings. It keeps the links watched and the exit 75 armed: an error is no orderly stop.
-    pub fn release(w: *Watch, why: []const u8) void {
-        w.lockRings();
+    fn abortRings(w: *Watch, why: []const u8) void {
+        while (w.rings_busy.swap(true, .acquire)) std.atomic.spinLoopHint();
         defer w.rings_busy.store(false, .release);
         for (w.rings[0..w.n_rings]) |r| r.abort(why);
-        w.n_rings = 0;
-    }
-
-    /// True when the watch found a link's end: a stop after it is no orderly stop.
-    pub fn faulted(w: *Watch) bool {
-        return w.seen.load(.acquire);
-    }
-
-    /// A released watch on a rank that ends after a peer fault: with TF_DS_EXIT_ON_FAULT=1, waits (at most the grace
-    /// plus 4 s) for the watch's exit 75, so the rank's normal return does not end the process with another code first.
-    pub fn hold(w: *Watch) void {
-        const grace_ms = exitGrace() orelse return;
-        var left = grace_ms + 4000;
-        while (left > 0 and !w.finished.load(.acquire)) : (left -|= 100) {
-            const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
-            _ = std.c.nanosleep(&ts, null);
-        }
     }
 
-    fn lockRings(w: *Watch) void {
-        while (w.rings_busy.swap(true, .acquire)) std.atomic.spinLoopHint();
+    /// A peer's link has ended now (a poll that does not wait): the error that the caller has may have come from it
+    /// before the watch thread saw it.
+    fn linkGone(w: *Watch) bool {
+        var pfds: [max_links]posix.pollfd = undefined;
+        for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
+        const k = posix.poll(pfds[0..w.n], 0) catch return false;
+        if (k == 0) return false;
+        for (pfds[0..w.n]) |p| if (p.revents & gone != 0) return true;
+        return false;
     }
 
     fn quiet(w: *Watch) bool {
-        return w.orderly.load(.acquire) or !w.running.load(.acquire);
+        return !w.running.load(.acquire);
     }
 
-    /// Waits up to `ms`; true if an orderly stop came in that time.
+    /// Waits up to `ms`; true if end came in that time.
     fn stopped(w: *Watch, ms: u64) bool {
         var left = ms;
         while (left > 0) : (left -|= 100) {
@@ -100,7 +126,6 @@ pub const Watch = struct {
     }
 
     fn loop(w: *Watch) void {
-        defer w.finished.store(true, .release);
         var pfds: [max_links]posix.pollfd = undefined;
         for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
         while (!w.quiet()) {
@@ -113,18 +138,15 @@ pub const Watch = struct {
         }
     }
 
+    /// A peer gone while this rank does not end (an idle rank, or a round whose waits got no error completion).
     fn peerGone(w: *Watch) void {
-        // rank 0's done frame and the close of its link come together: a second for this rank to take the stop
+        // rank 0's done frame and the close of its link come together: a second for this rank to call end
         if (w.stopped(1000)) return;
         w.lost.store(true, .release);
-        {
-            w.lockRings();
-            defer w.rings_busy.store(false, .release);
-            std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
-            for (w.rings[0..w.n_rings]) |r| r.abort("a peer rank is gone (its link closed)");
-        }
+        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
+        w.abortRings("a peer rank is gone (its link closed)");
         const grace_ms = exitGrace() orelse return;
-        // a live round fails first and the lane host ends the process itself after the same grace
+        // a live round fails first: its error goes out, then this rank calls end, or the lane host ends the process
         if (w.stopped(grace_ms + 2000)) return;
         std.debug.print("[tensorfold] rank {d}: exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{w.rank});
         // a stream may hold kernels that wait on the gone rank (NCCL's), which a normal exit would join
@@ -148,78 +170,98 @@ fn testPort() u16 {
     return 29500 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(std.testing.io, .awake).nanoseconds, 400)));
 }
 
-test "a peer that closes its link without a done frame is found within two seconds" {
-    const port = testPort();
-    const Peer = struct {
-        fn run(p: u16) void {
-            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
-            l.send("one frame") catch {};
-            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
-            _ = std.c.nanosleep(&ts, null);
-            l.close(); // the process "dies": its end closes
+fn sleepMs(ms: u64) void {
+    const ts: std.c.timespec = .{ .sec = @intCast(ms / 1000), .nsec = @intCast((ms % 1000) * std.time.ns_per_ms) };
+    _ = std.c.nanosleep(&ts, null);
+}
+
+/// The test's peer rank: it connects, then reads frames until a "done" frame or its link ends (keep), or closes its
+/// end after `ms` (dies).
+const Peer = struct {
+    port: u16,
+    die_ms: ?u64,
+
+    fn run(p: Peer) void {
+        var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p.port, 10) catch return;
+        defer l.close(); // the process ends: its end closes
+        if (p.die_ms) |ms| return sleepMs(ms);
+        var buf: [64]u8 = undefined;
+        while (true) {
+            const f = l.recv(&buf) catch return;
+            if (std.mem.eql(u8, f, "done")) return;
         }
-    };
-    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+    }
+};
+
+test "a start that fails ends the watch with no fault while the peer is up" {
+    const port = testPort();
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
-    defer l.close();
     const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
-    defer w.stop(std.testing.allocator);
+    // native.open fails after Lanes.init: end runs before Model.close frees the rings
+    try std.testing.expectEqual(Outcome.failed, w.end(std.testing.allocator, .failed));
+    l.close();
     t.join();
-    // a frame waiting unread is no end: the watch asks for the peer's close alone
+}
+
+test "an idle follower whose rank 0 dies: the watch acts, then end finds the peer fault" {
+    const port = testPort() + 400;
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 300 }});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
+    t.join();
+    // no receive runs: the watch alone finds the end (and aborts the rings) after its one-second wait
     var waited: u32 = 0;
-    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) {
-        const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
-        _ = std.c.nanosleep(&ts, null);
-    }
+    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) sleepMs(100);
     try std.testing.expect(w.lost.load(.acquire));
-    try std.testing.expect(w.faulted()); // a stop after it (rank 0's teardown) keeps the fault
+    try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
 }
 
-test "a released watch forgets its rings and still finds the peer gone" {
+test "a sending leader whose follower dies gets PeerDown, and end finds the peer fault" {
     const port = testPort() + 800;
-    const Peer = struct {
-        fn run(p: u16) void {
-            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
-            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
-            _ = std.c.nanosleep(&ts, null);
-            l.close();
-        }
-    };
-    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
-    defer w.stop(std.testing.allocator);
-    w.release("the test ends on an error");
-    try std.testing.expectEqual(@as(usize, 0), w.n_rings);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
     t.join();
-    var waited: u32 = 0;
-    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) {
-        const ts: std.c.timespec = .{ .sec = 0, .nsec = 100 * std.time.ns_per_ms };
-        _ = std.c.nanosleep(&ts, null);
+    // the first send after the close can still go out; a later one fails (no SIGPIPE: link.zig sends with NOSIGNAL)
+    var err: ?anyerror = null;
+    for (0..200) |_| {
+        l.send("a round's frame") catch |e| {
+            err = e;
+            break;
+        };
+        sleepMs(10);
     }
-    // an error is no orderly stop: the watch still acts on the peer's end
-    try std.testing.expect(w.lost.load(.acquire));
+    try std.testing.expectEqual(@as(?anyerror, error.PeerDown), err);
+    try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
 }
 
-test "an orderly stop before the link closes is no fault" {
-    const port = testPort() + 400;
-    const Peer = struct {
-        fn run(p: u16) void {
-            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
-            const ts: std.c.timespec = .{ .sec = 0, .nsec = 300 * std.time.ns_per_ms };
-            _ = std.c.nanosleep(&ts, null);
-            l.close();
-        }
-    };
-    const t = try std.Thread.spawn(.{}, Peer.run, .{port});
+test "a follower that gets PeerDown calls end before the watch acts, and end still finds the peer fault" {
+    const port = testPort() + 1200;
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 200 }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
     const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
-    w.orderly.store(true, .release); // the done frame came: the close that follows is expected
+    var buf: [64]u8 = undefined;
+    try std.testing.expectError(error.PeerDown, l.recv(&buf));
+    // within the watch's one-second wait: the watch has not acted, end acts for it
+    try std.testing.expect(!w.lost.load(.acquire));
+    try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
     t.join();
-    const ts: std.c.timespec = .{ .sec = 1, .nsec = 500 * std.time.ns_per_ms };
-    _ = std.c.nanosleep(&ts, null);
+}
+
+test "a done frame, then the peer's close: end is orderly" {
+    const port = testPort() + 1600;
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
+    try std.testing.expect(!w.faulted());
+    try l.send("done");
+    t.join(); // the peer took done and closed its end: the watch may have seen it
+    sleepMs(300);
     try std.testing.expect(!w.lost.load(.acquire));
-    w.stop(std.testing.allocator);
+    try std.testing.expectEqual(Outcome.orderly, w.end(std.testing.allocator, .done));
 }

From 5a339278f94f2662b2102719b6deecc9f8db7cc6 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 09:58:14 -0700
Subject: [PATCH 08/15] dsv41: the follower's errdefer ends the watch in a
 block

---
 zig/src/families/dsv41/lanes.zig | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 08c683e6f7..795e88184f 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -927,7 +927,9 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); the done
     // frame ends the watch orderly, an error ends it failed (exit 75 when rank 0 is gone), before Model.close
     const watch = watchLinks(gpa, m, &.{peer});
-    errdefer if (watch) |w| _ = w.end(gpa, .failed);
+    errdefer {
+        if (watch) |w| _ = w.end(gpa, .failed);
+    }
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);

From 6749974bb14047f2fd970a9b6200367b4dd82a6a Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 10:34:05 -0700
Subject: [PATCH 09/15] dsv41: TCP_USER_TIMEOUT on the rank links (20 s)

A keepalive only covers an idle link. When a peer's node loses power while
this rank has sent data that is not acknowledged, the kernel retransmits for
many minutes and the peer watch gets no POLLERR. TCP_USER_TIMEOUT at 20 s
(the keepalive's bound) ends such a link with ETIMEDOUT. Test: both ends of
a loopback link have the timeout and the keepalive.
---
 zig/src/families/dsv41/link.zig | 46 ++++++++++++++++++++++++++++++---
 1 file changed, 43 insertions(+), 3 deletions(-)

diff --git a/zig/src/families/dsv41/link.zig b/zig/src/families/dsv41/link.zig
index 28d5794779..7ebab304a1 100644
--- a/zig/src/families/dsv41/link.zig
+++ b/zig/src/families/dsv41/link.zig
@@ -3,6 +3,13 @@
 const std = @import("std");
 const c = std.c;
 
+/// TCP_USER_TIMEOUT (Linux): the longest time sent data can stay not acknowledged before the kernel ends the link
+/// (ETIMEDOUT: POLLERR to peer_watch.zig, PeerDown to a read or send). Without it, retransmissions run for many minutes.
+const tcp_user_timeout: u32 = 18;
+/// The same bound as the keepalive (10 s + 3 x 3 s, rounded up). With a keepalive on, Linux also ends an idle link
+/// whose probes get no answer at this time.
+const user_timeout_ms: u32 = 20_000;
+
 pub const Link = struct {
     fd: c.fd_t,
 
@@ -15,15 +22,16 @@ pub const Link = struct {
 
     /// TCP_NODELAY, and SO_BUSY_POLL: a read polls the NIC's receive queue itself (up to 50 us) instead of waiting for
     /// its interrupt, whose core may be leaving a deep idle state (node to node, 64-byte frames: round trip 1.0 ms
-    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL). A keepalive (10 s idle, then 3 probes 3 s
-    /// apart) shows a peer whose node vanished (no FIN) within about 20 s, to a read and to peer_watch.zig.
+    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL). A peer whose node vanished (no FIN) shows
+    /// within about 20 s, to a read and to peer_watch.zig: on an idle link by the keepalive (10 s idle, then 3 probes
+    /// 3 s apart), and with sent data not acknowledged (the node lost power during a round) by TCP_USER_TIMEOUT.
     fn tune(fd: c.fd_t) void {
         const one: c_int = 1;
         const busy_us: c_int = 50;
         _ = c.setsockopt(fd, c.IPPROTO.TCP, std.posix.TCP.NODELAY, std.mem.asBytes(&one), @sizeOf(c_int));
         _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.BUSY_POLL, std.mem.asBytes(&busy_us), @sizeOf(c_int));
         _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.KEEPALIVE, std.mem.asBytes(&one), @sizeOf(c_int));
-        for ([_][2]u32{ .{ std.posix.TCP.KEEPIDLE, 10 }, .{ std.posix.TCP.KEEPINTVL, 3 }, .{ std.posix.TCP.KEEPCNT, 3 } }) |o| {
+        for ([_][2]u32{ .{ std.posix.TCP.KEEPIDLE, 10 }, .{ std.posix.TCP.KEEPINTVL, 3 }, .{ std.posix.TCP.KEEPCNT, 3 }, .{ tcp_user_timeout, user_timeout_ms } }) |o| {
             const v: c_int = @intCast(o[1]);
             _ = c.setsockopt(fd, c.IPPROTO.TCP, o[0], std.mem.asBytes(&v), @sizeOf(c_int));
         }
@@ -162,6 +170,38 @@ test "addresses parse as four bytes" {
     try std.testing.expectError(error.BadAddress, parseIp("10.200.30"));
 }
 
+fn tcpOption(fd: c.fd_t, level: i32, name: u32) !c_int {
+    var v: c_int = 0;
+    var len: c.socklen_t = @sizeOf(c_int);
+    if (c.getsockopt(fd, level, name, &v, &len) != 0) return error.GetSockOptFailed;
+    return v;
+}
+
+test "both ends of a link find a gone peer within 20 s: keepalive and TCP_USER_TIMEOUT" {
+    const io = std.testing.io;
+    const port: u16 = 28400 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(io, .awake).nanoseconds, 500)));
+    const Peer = struct {
+        fn run(p: u16, out: *[2]c_int) void {
+            var l = Link.connect(std.testing.io, .{ 127, 0, 0, 1 }, p, 10) catch return;
+            defer l.close();
+            out[0] = tcpOption(l.fd, c.IPPROTO.TCP, tcp_user_timeout) catch -1;
+            out[1] = tcpOption(l.fd, c.SOL.SOCKET, c.SO.KEEPALIVE) catch -1;
+            var buf: [8]u8 = undefined;
+            _ = l.recv(&buf) catch {}; // until rank 0 closes its end
+        }
+    };
+    var follower: [2]c_int = .{ 0, 0 };
+    const t = try std.Thread.spawn(.{}, Peer.run, .{ port, &follower });
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    // rank 0's end (the accepted socket) and rank 1's end (connect)
+    try std.testing.expectEqual(@as(c_int, user_timeout_ms), try tcpOption(l.fd, c.IPPROTO.TCP, tcp_user_timeout));
+    try std.testing.expectEqual(@as(c_int, 1), try tcpOption(l.fd, c.SOL.SOCKET, c.SO.KEEPALIVE));
+    try std.testing.expectEqual(@as(c_int, 10), try tcpOption(l.fd, c.IPPROTO.TCP, std.posix.TCP.KEEPIDLE));
+    l.close();
+    t.join();
+    try std.testing.expectEqual([2]c_int{ user_timeout_ms, 1 }, follower);
+}
+
 test "frames cross a loopback link in order" {
     const io = std.testing.io;
     const port: u16 = 29000 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(io, .awake).nanoseconds, 500)));

From b862c70d80b95c85c008b800751e3754f23c83ad Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 11:05:30 -0700
Subject: [PATCH 10/15] dsv41: synchronize checks every RDMA ring, not only the
 neighbor gather's

When the peer watch aborts a ring while a round waits in synchronize, the
collect kernels leave their waits and can copy stale peer slots. synchronize
checked only the neighbor gather, so a TP=2 or TP=4 round could return its
results. It now checks every ring (Model.rdmaRings, the same list the watch
aborts) and fails with RdmaTransportFailed. Test: rdma.anyFailed finds an
aborted ring.
---
 zig/src/families/dsv41/lanes.zig | 17 ++---------------
 zig/src/families/dsv41/model.zig | 23 +++++++++++++++++++++++
 zig/src/families/dsv41/rdma.zig  | 28 ++++++++++++++++++++++++++++
 3 files changed, 53 insertions(+), 15 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 795e88184f..a7084d7c8c 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -900,22 +900,9 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
         n += 1;
     };
     if (n == 0) return null;
-    // every ring a GPU wait of this rank reads: TP2's, world 4's three, or the neighbor gather's two
+    // every ring a GPU wait of this rank reads (the same list that Model.synchronize checks)
     var rings: [6]*rdma.Ring = undefined;
-    var k: usize = 0;
-    if (m.ring) |r| {
-        rings[k] = r;
-        k += 1;
-    }
-    if (m.rdma_rings) |rr| for ([_]*rdma.Ring{ rr.all4, rr.col, rr.pair }) |r| {
-        rings[k] = r;
-        k += 1;
-    };
-    if (m.neighbor_gather) |ng| for ([_]*rdma.Ring{ ng.first, ng.second }) |r| {
-        rings[k] = r;
-        k += 1;
-    };
-    return peer_watch.Watch.start(gpa, fds[0..n], rings[0..k], m.rank) catch |err| {
+    return peer_watch.Watch.start(gpa, fds[0..n], m.rdmaRings(&rings), m.rank) catch |err| {
         std.debug.print("[tensorfold] the peer link watch did not start ({s})\n", .{@errorName(err)});
         return null;
     };
diff --git a/zig/src/families/dsv41/model.zig b/zig/src/families/dsv41/model.zig
index 6442e43d81..3072cbef45 100644
--- a/zig/src/families/dsv41/model.zig
+++ b/zig/src/families/dsv41/model.zig
@@ -448,6 +448,29 @@ pub const Model = struct {
     fn synchronize(m: *Model) !void {
         try m.stream.synchronize();
         if (m.neighbor_gather) |*ng| try ng.check();
+        // a ring aborted while this work ran (a peer gone, peer_watch.zig): its collects left their waits and may have
+        // copied stale peer slots, so the results of this work must not be used
+        var buf: [6]*rdma.Ring = undefined;
+        if (rdma.anyFailed(m.rdmaRings(&buf))) return error.RdmaTransportFailed;
+    }
+
+    /// Every RDMA ring that a GPU wait of this rank reads: TP2's ring, world 4's three rings (ring2d.zig), and the
+    /// neighbor gather's two (neighbor.zig). The peer watch aborts these; synchronize checks them.
+    pub fn rdmaRings(m: *Model, buf: *[6]*rdma.Ring) []*rdma.Ring {
+        var k: usize = 0;
+        if (m.ring) |r| {
+            buf[k] = r;
+            k += 1;
+        }
+        if (m.rdma_rings) |rr| for ([_]*rdma.Ring{ rr.all4, rr.col, rr.pair }) |r| {
+            buf[k] = r;
+            k += 1;
+        };
+        if (m.neighbor_gather) |ng| for ([_]*rdma.Ring{ ng.first, ng.second }) |r| {
+            buf[k] = r;
+            k += 1;
+        };
+        return buf[0..k];
     }
 
     /// TP2's RDMA ring for the decode gathers: created on both ranks, the queue-pair infos exchanged over NCCL,
diff --git a/zig/src/families/dsv41/rdma.zig b/zig/src/families/dsv41/rdma.zig
index 73ff3ee63b..900cd087f5 100644
--- a/zig/src/families/dsv41/rdma.zig
+++ b/zig/src/families/dsv41/rdma.zig
@@ -502,6 +502,12 @@ pub const Ring = struct {
     }
 };
 
+/// True when one of `rings` failed or was aborted: the work that waited on it may hold stale peer data.
+pub fn anyFailed(rings: []const *Ring) bool {
+    for (rings) |r| if (r.failure() != null) return true;
+    return false;
+}
+
 fn postRecv(ctx: *vabi.Context, qp: *vabi.Qp) Error!void {
     var wr: vabi.RecvWr = .{};
     var bad: ?*vabi.RecvWr = null;
@@ -546,3 +552,25 @@ test "a failed neighbor stage releases both local stages' device waits" {
     try std.testing.expectEqual(@as(u64, 1), first.abortWord().*);
     try std.testing.expectEqual(@as(u64, 1), second.abortWord().*);
 }
+
+test "anyFailed finds a ring that the peer watch aborted" {
+    var rings: [3]Ring = undefined;
+    var meta0: [32]u64 align(std.heap.page_size_min) = @splat(0);
+    var meta1: [32]u64 align(std.heap.page_size_min) = @splat(0);
+    var meta2: [32]u64 align(std.heap.page_size_min) = @splat(0);
+    var ptrs: [3]*Ring = undefined;
+    for (&rings, [_][]u64{ &meta0, &meta1, &meta2 }, &ptrs) |*r, meta, *p| {
+        r.s.slots = 4;
+        r.world = 2;
+        r.meta.mem = @alignCast(std.mem.sliceAsBytes(meta));
+        r.failed = .init(false);
+        r.why = @splat(0);
+        r.failure_peer = null;
+        p.* = r;
+    }
+    try std.testing.expect(!anyFailed(&ptrs));
+    try std.testing.expect(!anyFailed(&.{}));
+    rings[2].abort("a peer rank is gone (its link closed)");
+    try std.testing.expect(anyFailed(&ptrs));
+    try std.testing.expect(!anyFailed(ptrs[0..2]));
+}

From f4012d8c656d9ff262d5241e3fad64ff70c6dc53 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 15:43:07 -0700
Subject: [PATCH 11/15] dsv41: peer watch review fixes (requested stop,
 injectable exit, logs, POLLNVAL)

- A requested stop (the server's SIGTERM) ends rank 0's watch orderly even
  when a follower went away during it, so a clean stop exits 0, not 75.
- Watch.start takes the exit grace (faultGrace() in the engine, null in
  tests), so an exported TF_DS_EXIT_ON_FAULT cannot end the test runner.
- The abort comes before any log, and every watch log is written only when
  stderr takes it at once, so a blocked stderr cannot hold the abort or the
  exit 75.
- faulted() also polls the links now: a FIN that the watch thread has not
  seen yet still makes rank 0's teardown a peer fault.
- POLLNVAL counts as a link end, and a failed poll sleeps 200 ms (no spin).
  With TF_DS_EXIT_ON_FAULT=1 a watch that cannot start fails the start.
- The follower starts its watch after its kept states, so on an error the
  rings are aborted (and exit 75 comes) before their device memory is freed.

Tests: POLLNVAL on a closed pipe end; the exit comes from Watch.start's grace
alone, with TF_DS_EXIT_ON_FAULT=1 exported.
---
 zig/src/families/dsv41/lanes.zig      | 32 ++++++----
 zig/src/families/dsv41/lanes_main.zig |  2 +-
 zig/src/families/dsv41/native.zig     |  3 +-
 zig/src/families/dsv41/peer_watch.zig | 90 ++++++++++++++++++++-------
 4 files changed, 90 insertions(+), 37 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index a7084d7c8c..9cd02f8a90 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -288,6 +288,7 @@ pub const Lanes = struct {
     m: *Model,
     peer: ?link.Link, // rank 1 (null: one rank); every follower gets each frame (Model.followers)
     watch: ?*peer_watch.Watch = null, // the links to the followers, watched for a rank that dies (peer_watch.zig)
+    stop_requested: bool = false, // set by the server's stop (native.zig deinitFn): deinit then ends the watch orderly
     vision: ?*@import("vision.zig").Vision = null, // rank 0's tower (a vision checkpoint served with its kit's vision/)
     step: u64 = 0,
     wire: Wire = .{},
@@ -308,9 +309,9 @@ pub const Lanes = struct {
     snaps: std.ArrayList(*Snap) = .empty, // kept prompt states, oldest first
     next_snap: u64 = 1,
 
-    pub fn init(gpa: std.mem.Allocator, m: *Model) Lanes {
+    pub fn init(gpa: std.mem.Allocator, m: *Model) !Lanes {
         var ln: Lanes = .{ .gpa = gpa, .m = m, .peer = if (m.world > 1) m.peer() else null };
-        if (m.world > 1) ln.watch = watchLinks(gpa, m, m.followers());
+        if (m.world > 1) ln.watch = try watchLinks(gpa, m, m.followers());
         // the round costs by rows: ROUND_MS[min(rows, len) - 1] (the served table, or TF_DS_ROUND_MS)
         const t = m.round_ms;
         for (&ln.wc, 0..) |*c, i| c.* = .{ .width = @intCast(i + 1), .ms = t[@min(i, t.len - 1)] };
@@ -332,11 +333,12 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
-        // a follower that the watch found gone before the done frame is a fault, and so is a done frame that cannot go
+        // a requested stop (the server's SIGTERM) is orderly even when a follower went away during it (both ranks got
+        // the signal). Else a follower gone before the done frame is a fault, and so is a done frame that cannot go
         // out; the followers' close after a done frame that went out is expected (within the watch's one-second wait)
         const lost = if (self.watch) |w| w.faulted() else false;
         const sent = if (self.peer != null) (if (self.send(.done)) |_| true else |_| false) else true;
-        self.endWatch(if (!lost and sent) .done else .failed);
+        self.endWatch(if (self.stop_requested or (!lost and sent)) .done else .failed);
         self.streams.deinit(self.gpa);
         for (&self.seqs) |*q| q.deinit(self.gpa);
         self.wire.deinit(self.gpa);
@@ -891,8 +893,9 @@ fn refused(err: anyerror) void {
 /// A follower's verify times with Model.prof (a profile on ranks 1-3: rounds, issue and wall ns summed).
 pub var follow_prof: struct { rounds: u64 = 0, enqueue: u64 = 0, forward: u64 = 0 } = .{};
 
-/// A watch on `links` (null: it could not start; the ranks then run as before, without it).
-fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*peer_watch.Watch {
+/// A watch on `links`. If it cannot start: with TF_DS_EXIT_ON_FAULT=1 the error (the exit 75 contract needs the watch,
+/// so the start fails); else null, and the ranks run as before, without it.
+fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) !?*peer_watch.Watch {
     var fds: [3]std.posix.fd_t = undefined;
     var n: usize = 0;
     for (links) |l| if (l) |x| if (n < fds.len) {
@@ -902,8 +905,10 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
     if (n == 0) return null;
     // every ring a GPU wait of this rank reads (the same list that Model.synchronize checks)
     var rings: [6]*rdma.Ring = undefined;
-    return peer_watch.Watch.start(gpa, fds[0..n], m.rdmaRings(&rings), m.rank) catch |err| {
+    const grace = peer_watch.faultGrace();
+    return peer_watch.Watch.start(gpa, fds[0..n], m.rdmaRings(&rings), m.rank, grace) catch |err| {
         std.debug.print("[tensorfold] the peer link watch did not start ({s})\n", .{@errorName(err)});
+        if (grace != null) return err;
         return null;
     };
 }
@@ -911,12 +916,6 @@ fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) ?*pe
 /// Rank 1 (each of ranks 1-3 on the four-node split): the primitives rank 0 sends, run in order until it says done.
 pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
     const peer = m.peer();
-    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); the done
-    // frame ends the watch orderly, an error ends it failed (exit 75 when rank 0 is gone), before Model.close
-    const watch = watchLinks(gpa, m, &.{peer});
-    errdefer {
-        if (watch) |w| _ = w.end(gpa, .failed);
-    }
     const buf = try gpa.alloc(u8, (m.pool_cap + 1) * 4 + (1 << 16));
     defer gpa.free(buf);
     var seqs: [max_streams]std.ArrayList(i32) = @splat(.empty);
@@ -934,6 +933,13 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
         while (ki.next()) |b| b.free();
         kept.deinit(gpa);
     }
+    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); the done
+    // frame ends the watch orderly, an error ends it failed (exit 75 when rank 0 is gone). The watch starts after the
+    // kept states, so on an error it ends (rings aborted, exit 75) before their device memory is freed
+    const watch = try watchLinks(gpa, m, &.{peer});
+    errdefer {
+        if (watch) |w| _ = w.end(gpa, .failed);
+    }
     while (true) {
         // the next primitive, polled for 20 ms before the read blocks (rank 0 sends it a fraction of a millisecond
         // after this rank's GPU goes idle in a decode; a sleeping thread would add its core's wake-up to every frame)
diff --git a/zig/src/families/dsv41/lanes_main.zig b/zig/src/families/dsv41/lanes_main.zig
index e91cff92c2..7ee73538ca 100644
--- a/zig/src/families/dsv41/lanes_main.zig
+++ b/zig/src/families/dsv41/lanes_main.zig
@@ -184,7 +184,7 @@ pub fn main(init: std.process.Init) !u8 {
     const text = try std.Io.Dir.cwd().readFileAlloc(io, requests_file orelse return error.NoRequests, a, .limited(1 << 30));
     const Requests = struct { requests: []Request };
     const reqs = (try std.json.parseFromSliceLeaky(Requests, a, text, .{ .ignore_unknown_fields = true })).requests;
-    var ln = dsv41.lanes.Lanes.init(gpa, m);
+    var ln = try dsv41.lanes.Lanes.init(gpa, m);
     defer ln.deinit();
     ln.fix_k = fix_k;
     if (profile) {
diff --git a/zig/src/families/dsv41/native.zig b/zig/src/families/dsv41/native.zig
index db1ba482a3..003925c97c 100644
--- a/zig/src/families/dsv41/native.zig
+++ b/zig/src/families/dsv41/native.zig
@@ -157,7 +157,7 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [
         .streams = if (getenv("TF_DS_STREAMS")) |v| std.fmt.parseInt(usize, v, 10) catch return error.BadStreams else model.max_streams_default,
     });
     errdefer st.m.close();
-    st.lanes = lanes_mod.Lanes.init(gpa, st.m);
+    st.lanes = try lanes_mod.Lanes.init(gpa, st.m);
     errdefer st.lanes.endWatch(.failed); // a failed start closes the model next: the watch ends first
     if (getenv("TF_DS_FIX_K")) |t| st.lanes.fix_k = std.fmt.parseInt(usize, t, 10) catch return error.BadFixK;
     if (getenv("TF_DS_SERVED_K")) |t| st.lanes.served_k = if (std.mem.eql(u8, t, "1") or std.mem.eql(u8, t, "all")) .all else if (std.mem.eql(u8, t, "solo")) .solo else .off;
@@ -174,6 +174,7 @@ pub fn open(gpa: std.mem.Allocator, io: std.Io, ctx: *const cuda.Context, dir: [
 
 fn deinitFn(ptr: *anyopaque) void {
     const st: *State = @ptrCast(@alignCast(ptr));
+    st.lanes.stop_requested = true; // the server stops (its SIGTERM, or its start failed after open): an orderly end
     st.lanes.deinit(); // rank 1 hears done
     if (st.vision) |v| v.close();
     st.m.close();
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index 3080a47fe1..1122da7874 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -12,8 +12,9 @@ const Allocator = std.mem.Allocator;
 
 /// POLLRDHUP (Linux): the peer closed its end.
 const rdhup: i16 = 0x2000;
-/// The peer closed its end, or the connection failed (a keepalive or a send got no answer).
-const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR;
+/// The peer closed its end, or the connection failed (a keepalive or a send got no answer), or the descriptor is no
+/// longer open (POLLNVAL: a poll on it would return at once, again and again).
+const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR | posix.POLL.NVAL;
 const max_links = 3;
 /// TP2's ring, world 4's three rings (ring2d.zig), or the neighbor gather's two (neighbor.zig).
 const max_rings = 6;
@@ -49,24 +50,26 @@ pub const Watch = struct {
     lost: std.atomic.Value(bool) = .init(false),
     /// Held while a thread aborts the rings (a spin: it is taken at most twice).
     rings_busy: std.atomic.Value(bool) = .init(false),
+    /// The exit 75 on a peer fault, after this grace in ms (faultGrace: TF_DS_EXIT_ON_FAULT=1); null: no exit.
+    exit_grace: ?u64,
     thread: ?std.Thread = null,
 
     /// A watch thread on `fds` (rank 0's links to its followers, or a follower's one link to rank 0) that aborts
-    /// `rings` (every RDMA ring this rank's GPU waits on) when a peer is gone.
-    pub fn start(gpa: Allocator, fds: []const posix.fd_t, rings: []const *rdma.Ring, rank: u32) !*Watch {
+    /// `rings` (every RDMA ring this rank's GPU waits on) when a peer is gone. `exit_grace`: faultGrace() (tests: null).
+    pub fn start(gpa: Allocator, fds: []const posix.fd_t, rings: []const *rdma.Ring, rank: u32, exit_grace: ?u64) !*Watch {
         const w = try gpa.create(Watch);
         errdefer gpa.destroy(w);
-        w.* = .{ .rank = rank, .n = @min(fds.len, max_links), .n_rings = @min(rings.len, max_rings) };
+        w.* = .{ .rank = rank, .n = @min(fds.len, max_links), .n_rings = @min(rings.len, max_rings), .exit_grace = exit_grace };
         @memcpy(w.fds[0..w.n], fds[0..w.n]);
         @memcpy(w.rings[0..w.n_rings], rings[0..w.n_rings]);
         w.thread = try std.Thread.spawn(.{}, loop, .{w});
         return w;
     }
 
-    /// True when the poll found a link's end: rank 0 reads it before its done frame (a done frame after it is no
-    /// orderly stop).
+    /// True when a peer's link has ended (the watch thread saw it, or a poll that does not wait finds it now, before
+    /// the thread sees it): rank 0 reads it before its done frame (a done frame after it is no orderly stop).
     pub fn faulted(w: *Watch) bool {
-        return w.seen.load(.acquire);
+        return w.seen.load(.acquire) or w.linkGone();
     }
 
     /// The one teardown, for every way a rank ends (rank 0's Lanes.deinit, a follower's done frame or error, a start
@@ -82,12 +85,13 @@ pub const Watch = struct {
         if (w.thread) |t| t.join();
         const out: Outcome = switch (how) {
             .done => .orderly,
-            .failed => if (w.seen.load(.acquire) or w.linkGone()) .peer_fault else .failed,
+            .failed => if (w.faulted()) .peer_fault else .failed,
         };
         const rank = w.rank;
+        const exit_on_fault = w.exit_grace != null;
         gpa.destroy(w);
-        if (out == .peer_fault and exitGrace() != null) {
-            std.debug.print("[tensorfold] rank {d}: a peer rank is gone; exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{rank});
+        if (out == .peer_fault and exit_on_fault) {
+            note("[tensorfold] rank {d}: a peer rank is gone; exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{rank});
             std.c._exit(75);
         }
         return out;
@@ -129,7 +133,10 @@ pub const Watch = struct {
         var pfds: [max_links]posix.pollfd = undefined;
         for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
         while (!w.quiet()) {
-            const k = posix.poll(pfds[0..w.n], 200) catch continue;
+            const k = posix.poll(pfds[0..w.n], 200) catch {
+                sleepMs(200); // no spin on a poll that fails at once
+                continue;
+            };
             if (k == 0) continue;
             for (pfds[0..w.n]) |p| if (p.revents & gone != 0) {
                 w.seen.store(true, .release);
@@ -143,19 +150,32 @@ pub const Watch = struct {
         // rank 0's done frame and the close of its link come together: a second for this rank to call end
         if (w.stopped(1000)) return;
         w.lost.store(true, .release);
-        std.debug.print("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
+        // the abort first: no log comes before it
         w.abortRings("a peer rank is gone (its link closed)");
-        const grace_ms = exitGrace() orelse return;
+        note("[tensorfold] rank {d}: a peer rank's link closed (its process or node is gone); the waits of its {d} RDMA ring(s) are aborted\n", .{ w.rank, w.n_rings });
+        const grace_ms = w.exit_grace orelse return;
         // a live round fails first: its error goes out, then this rank calls end, or the lane host ends the process
         if (w.stopped(grace_ms + 2000)) return;
-        std.debug.print("[tensorfold] rank {d}: exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{w.rank});
+        note("[tensorfold] rank {d}: exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{w.rank});
         // a stream may hold kernels that wait on the gone rank (NCCL's), which a normal exit would join
         std.c._exit(75);
     }
 };
 
-/// TF_DS_EXIT_ON_FAULT=1: the grace in ms (TF_DS_EXIT_GRACE seconds, default 5), as lane_host.faultExit reads it.
-fn exitGrace() ?u64 {
+/// A log line to stderr that never blocks: it is written only when stderr can take it now (a write up to PIPE_BUF to
+/// a pipe that poll reports writable goes out whole), else it is dropped. The abort and the exit never wait on a log.
+fn note(comptime fmt: []const u8, args: anytype) void {
+    var buf: [512]u8 = undefined;
+    const line = std.fmt.bufPrint(&buf, fmt, args) catch return;
+    var p = [1]posix.pollfd{.{ .fd = 2, .events = posix.POLL.OUT, .revents = 0 }};
+    const k = posix.poll(&p, 0) catch return;
+    if (k == 0 or p[0].revents & posix.POLL.OUT == 0) return;
+    _ = std.c.write(2, line.ptr, line.len);
+}
+
+/// TF_DS_EXIT_ON_FAULT=1: the grace in ms (TF_DS_EXIT_GRACE seconds, default 5), as lane_host.faultExit reads it. The
+/// exit 75 of a watch (Watch.start's exit_grace); null when the variable is not 1.
+pub fn faultGrace() ?u64 {
     const on = std.c.getenv("TF_DS_EXIT_ON_FAULT") orelse return null;
     if (!std.mem.eql(u8, std.mem.span(on), "1")) return null;
     const g = std.c.getenv("TF_DS_EXIT_GRACE") orelse return 5000;
@@ -197,7 +217,7 @@ test "a start that fails ends the watch with no fault while the peer is up" {
     const port = testPort();
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null);
     // native.open fails after Lanes.init: end runs before Model.close frees the rings
     try std.testing.expectEqual(Outcome.failed, w.end(std.testing.allocator, .failed));
     l.close();
@@ -209,7 +229,7 @@ test "an idle follower whose rank 0 dies: the watch acts, then end finds the pee
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 300 }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null);
     t.join();
     // no receive runs: the watch alone finds the end (and aborts the rings) after its one-second wait
     var waited: u32 = 0;
@@ -223,7 +243,7 @@ test "a sending leader whose follower dies gets PeerDown, and end finds the peer
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null);
     t.join();
     // the first send after the close can still go out; a later one fails (no SIGPIPE: link.zig sends with NOSIGNAL)
     var err: ?anyerror = null;
@@ -243,7 +263,7 @@ test "a follower that gets PeerDown calls end before the watch acts, and end sti
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 200 }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null);
     var buf: [64]u8 = undefined;
     try std.testing.expectError(error.PeerDown, l.recv(&buf));
     // within the watch's one-second wait: the watch has not acted, end acts for it
@@ -257,7 +277,7 @@ test "a done frame, then the peer's close: end is orderly" {
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = null }});
     var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
     defer l.close();
-    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0);
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 0, null);
     try std.testing.expect(!w.faulted());
     try l.send("done");
     t.join(); // the peer took done and closed its end: the watch may have seen it
@@ -265,3 +285,29 @@ test "a done frame, then the peer's close: end is orderly" {
     try std.testing.expect(!w.lost.load(.acquire));
     try std.testing.expectEqual(Outcome.orderly, w.end(std.testing.allocator, .done));
 }
+
+test "a descriptor that is no longer open counts as a link end (POLLNVAL), not as a poll to repeat" {
+    const fds = try posix.pipe();
+    defer posix.close(fds[1]);
+    const w = try Watch.start(std.testing.allocator, &.{fds[0]}, &.{}, 1, null);
+    posix.close(fds[0]);
+    var waited: u32 = 0;
+    while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) sleepMs(100);
+    try std.testing.expect(w.lost.load(.acquire));
+    try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
+}
+
+test "the exit 75 comes from Watch.start's grace alone: TF_DS_EXIT_ON_FAULT in the test's environment has no effect" {
+    try std.testing.expectEqual(@as(c_int, 0), std.c.setenv("TF_DS_EXIT_ON_FAULT", "1", 1));
+    defer _ = std.c.unsetenv("TF_DS_EXIT_ON_FAULT");
+    try std.testing.expectEqual(@as(?u64, 5000), faultGrace());
+    const port = testPort() + 2000;
+    const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }});
+    var l = try Link.listen(.{ 127, 0, 0, 1 }, port);
+    defer l.close();
+    const w = try Watch.start(std.testing.allocator, &.{l.fd}, &.{}, 1, null);
+    t.join();
+    sleepMs(100);
+    // a peer fault with no grace: end returns (the test runner goes on) instead of exit 75
+    try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
+}

From cd5970c467fb57db8be4f5303b8630aa41eccae9 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:03:17 -0700
Subject: [PATCH 12/15] dsv41: the draft pass gets the ring check too

Model.pass returns the drafter's drafts after the drafter syncs its own
stream (draft.zig), so the ring check in synchronize never ran for it. A
ring aborted during a draft pass now fails the pass (RdmaTransportFailed)
before its drafts are used. The check is one function, ringsOk, for both.
---
 zig/src/families/dsv41/model.zig | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/zig/src/families/dsv41/model.zig b/zig/src/families/dsv41/model.zig
index 3072cbef45..1560580ec9 100644
--- a/zig/src/families/dsv41/model.zig
+++ b/zig/src/families/dsv41/model.zig
@@ -448,8 +448,12 @@ pub const Model = struct {
     fn synchronize(m: *Model) !void {
         try m.stream.synchronize();
         if (m.neighbor_gather) |*ng| try ng.check();
-        // a ring aborted while this work ran (a peer gone, peer_watch.zig): its collects left their waits and may have
-        // copied stale peer slots, so the results of this work must not be used
+        try m.ringsOk();
+    }
+
+    /// After a stream sync, before its results are used: a ring aborted while the work ran (a peer gone,
+    /// peer_watch.zig) made its collects leave their waits, maybe with stale peer slots, so the work fails.
+    fn ringsOk(m: *Model) !void {
         var buf: [6]*rdma.Ring = undefined;
         if (rdma.anyFailed(m.rdmaRings(&buf))) return error.RdmaTransportFailed;
     }
@@ -854,5 +858,7 @@ pub const Model = struct {
         const pool = &m.dpool.?;
         for (q0, slots) |q, s| pool.absorbed[@intCast(s)] = @intCast(q);
         try dr.pass(&m.eng, &m.ch, pool, tokens, q0, slots, steps, null);
+        // the drafter syncs its stream itself (draft.zig), so its drafts get the same ring check as synchronize
+        try m.ringsOk();
     }
 };

From 8e7d31f516af4cafc7885a186c48a13451794f1c Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:26:54 -0700
Subject: [PATCH 13/15] dsv41: peer_watch tests declare pipe, setenv and
 unsetenv (std.c has none)

---
 zig/src/families/dsv41/peer_watch.zig | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index 1122da7874..abe05cc277 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -186,6 +186,11 @@ pub fn faultGrace() ?u64 {
 
 const Link = @import("link.zig").Link;
 
+// libc calls that the tests need and std.c does not declare
+extern "c" fn pipe(fds: *[2]c_int) c_int;
+extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
+extern "c" fn unsetenv(name: [*:0]const u8) c_int;
+
 fn testPort() u16 {
     return 29500 + @as(u16, @intCast(@mod(std.Io.Timestamp.now(std.testing.io, .awake).nanoseconds, 400)));
 }
@@ -287,10 +292,11 @@ test "a done frame, then the peer's close: end is orderly" {
 }
 
 test "a descriptor that is no longer open counts as a link end (POLLNVAL), not as a poll to repeat" {
-    const fds = try posix.pipe();
-    defer posix.close(fds[1]);
+    var fds: [2]c_int = undefined;
+    try std.testing.expectEqual(@as(c_int, 0), pipe(&fds));
+    defer _ = std.c.close(fds[1]);
     const w = try Watch.start(std.testing.allocator, &.{fds[0]}, &.{}, 1, null);
-    posix.close(fds[0]);
+    _ = std.c.close(fds[0]);
     var waited: u32 = 0;
     while (!w.lost.load(.acquire) and waited < 40) : (waited += 1) sleepMs(100);
     try std.testing.expect(w.lost.load(.acquire));
@@ -298,8 +304,8 @@ test "a descriptor that is no longer open counts as a link end (POLLNVAL), not a
 }
 
 test "the exit 75 comes from Watch.start's grace alone: TF_DS_EXIT_ON_FAULT in the test's environment has no effect" {
-    try std.testing.expectEqual(@as(c_int, 0), std.c.setenv("TF_DS_EXIT_ON_FAULT", "1", 1));
-    defer _ = std.c.unsetenv("TF_DS_EXIT_ON_FAULT");
+    try std.testing.expectEqual(@as(c_int, 0), setenv("TF_DS_EXIT_ON_FAULT", "1", 1));
+    defer _ = unsetenv("TF_DS_EXIT_ON_FAULT");
     try std.testing.expectEqual(@as(?u64, 5000), faultGrace());
     const port = testPort() + 2000;
     const t = try std.Thread.spawn(.{}, Peer.run, .{Peer{ .port = port, .die_ms = 0 }});

From b4109452b4661135b66995851695e7a40ecf55e1 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Sat, 24 Oct 2026 11:36:57 -0700
Subject: [PATCH 14/15] dsv41: the peer watch logs through a nonblocking copy
 of stderr (no poll snapshot)

note() wrote to fd 2 after a poll(POLLOUT) snapshot. Another thread can fill
the pipe between the poll and the write, so the write could block after the
ring abort and before the exit 75. Watch.start now opens /proc/self/fd/2
again with O_NONBLOCK (a new open file description, so fd 2 stays blocking
for the other threads); note writes only through it, and a socket stderr
gets send with MSG_DONTWAIT. A new test fills a pipe through the copy and
checks that the pipe's own descriptor stays blocking.
---
 zig/src/families/dsv41/peer_watch.zig | 52 +++++++++++++++++++++++----
 1 file changed, 46 insertions(+), 6 deletions(-)

diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index abe05cc277..d96093ca3f 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -62,6 +62,7 @@ pub const Watch = struct {
         w.* = .{ .rank = rank, .n = @min(fds.len, max_links), .n_rings = @min(rings.len, max_rings), .exit_grace = exit_grace };
         @memcpy(w.fds[0..w.n], fds[0..w.n]);
         @memcpy(w.rings[0..w.n_rings], rings[0..w.n_rings]);
+        openLog();
         w.thread = try std.Thread.spawn(.{}, loop, .{w});
         return w;
     }
@@ -162,15 +163,34 @@ pub const Watch = struct {
     }
 };
 
-/// A log line to stderr that never blocks: it is written only when stderr can take it now (a write up to PIPE_BUF to
-/// a pipe that poll reports writable goes out whole), else it is dropped. The abort and the exit never wait on a log.
+/// note's descriptor: stderr's file opened again with O_NONBLOCK (openLog); -1: none; -2: not opened yet.
+var log_fd: std.atomic.Value(c_int) = .init(-2);
+
+/// A new open file description of `fd` with O_NONBLOCK, so fd itself (shared with other threads) stays blocking.
+fn nonblockingCopy(fd: c_int) c_int {
+    var buf: [32]u8 = undefined;
+    const path = std.fmt.bufPrintSentinel(&buf, "/proc/self/fd/{d}", .{fd}, 0) catch return -1;
+    return std.c.open(path, .{ .ACCMODE = .WRONLY, .NONBLOCK = true, .APPEND = true, .CLOEXEC = true });
+}
+
+/// Opens note's descriptor once per process (Watch.start), before any fault path can need it.
+fn openLog() void {
+    if (log_fd.load(.acquire) != -2) return;
+    const fd = nonblockingCopy(2);
+    if (log_fd.cmpxchgStrong(-2, if (fd >= 0) fd else -1, .acq_rel, .acquire) != null and fd >= 0) _ = std.c.close(fd);
+}
+
+/// A log line to stderr that never blocks: a full pipe or terminal drops it; the abort and exit never wait on it.
 fn note(comptime fmt: []const u8, args: anytype) void {
     var buf: [512]u8 = undefined;
     const line = std.fmt.bufPrint(&buf, fmt, args) catch return;
-    var p = [1]posix.pollfd{.{ .fd = 2, .events = posix.POLL.OUT, .revents = 0 }};
-    const k = posix.poll(&p, 0) catch return;
-    if (k == 0 or p[0].revents & posix.POLL.OUT == 0) return;
-    _ = std.c.write(2, line.ptr, line.len);
+    const fd = log_fd.load(.acquire);
+    if (fd >= 0) {
+        _ = std.c.write(fd, line.ptr, line.len);
+        return;
+    }
+    // a socket stderr cannot open through /proc: MSG_DONTWAIT; on any other descriptor send fails and the line drops
+    _ = std.c.send(2, line.ptr, line.len, std.c.MSG.DONTWAIT | std.c.MSG.NOSIGNAL);
 }
 
 /// TF_DS_EXIT_ON_FAULT=1: the grace in ms (TF_DS_EXIT_GRACE seconds, default 5), as lane_host.faultExit reads it. The
@@ -317,3 +337,23 @@ test "the exit 75 comes from Watch.start's grace alone: TF_DS_EXIT_ON_FAULT in t
     // a peer fault with no grace: end returns (the test runner goes on) instead of exit 75
     try std.testing.expectEqual(Outcome.peer_fault, w.end(std.testing.allocator, .failed));
 }
+
+test "note's descriptor never blocks on a full pipe, and the pipe's own descriptor stays blocking" {
+    var fds: [2]c_int = undefined;
+    try std.testing.expectEqual(@as(c_int, 0), pipe(&fds));
+    defer _ = std.c.close(fds[0]);
+    defer _ = std.c.close(fds[1]);
+    const w = nonblockingCopy(fds[1]);
+    try std.testing.expect(w >= 0);
+    defer _ = std.c.close(w);
+    const chunk: [4096]u8 = @splat('x');
+    var full = false;
+    // 4 MiB at most, more than a pipe holds: a write that would wait fails at once (EAGAIN)
+    for (0..1024) |_| if (std.c.write(w, &chunk, chunk.len) < 0) {
+        full = true;
+        break;
+    };
+    try std.testing.expect(full);
+    const flags: std.c.O = @bitCast(@as(u32, @intCast(std.c.fcntl(fds[1], std.c.F.GETFL))));
+    try std.testing.expect(!flags.NONBLOCK);
+}

From cf72ae6693cd17079ef86de2d6418c62e75b5109 Mon Sep 17 00:00:00 2001
From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com>
Date: Sat, 24 Oct 2026 11:39:20 -0700
Subject: [PATCH 15/15] dsv41: peer watch, link and ring check comments are one
 line (lean check)

---
 zig/src/families/dsv41/lanes.zig      | 14 +++-------
 zig/src/families/dsv41/link.zig       | 13 ++++------
 zig/src/families/dsv41/model.zig      |  6 ++---
 zig/src/families/dsv41/peer_watch.zig | 37 ++++++++-------------------
 4 files changed, 22 insertions(+), 48 deletions(-)

diff --git a/zig/src/families/dsv41/lanes.zig b/zig/src/families/dsv41/lanes.zig
index 9cd02f8a90..6dc6f6577a 100644
--- a/zig/src/families/dsv41/lanes.zig
+++ b/zig/src/families/dsv41/lanes.zig
@@ -319,8 +319,7 @@ pub const Lanes = struct {
         return ln;
     }
 
-    /// Ends the peer link watch (Watch.end, the one teardown: before the model frees its links and rings). `failed`:
-    /// with a follower gone and TF_DS_EXIT_ON_FAULT=1, it does not return (exit 75).
+    /// Ends the peer link watch (Watch.end) before the model frees its links and rings; may exit 75 on `failed`.
     pub fn endWatch(self: *Lanes, how: peer_watch.How) void {
         if (self.watch) |w| _ = w.end(self.gpa, how);
         self.watch = null;
@@ -333,9 +332,7 @@ pub const Lanes = struct {
             self.gpa.destroy(sn);
         }
         self.snaps.deinit(self.gpa);
-        // a requested stop (the server's SIGTERM) is orderly even when a follower went away during it (both ranks got
-        // the signal). Else a follower gone before the done frame is a fault, and so is a done frame that cannot go
-        // out; the followers' close after a done frame that went out is expected (within the watch's one-second wait)
+        // a SIGTERM stop is orderly; else a follower gone before the done frame, or a done frame that fails, is a fault
         const lost = if (self.watch) |w| w.faulted() else false;
         const sent = if (self.peer != null) (if (self.send(.done)) |_| true else |_| false) else true;
         self.endWatch(if (self.stop_requested or (!lost and sent)) .done else .failed);
@@ -893,8 +890,7 @@ fn refused(err: anyerror) void {
 /// A follower's verify times with Model.prof (a profile on ranks 1-3: rounds, issue and wall ns summed).
 pub var follow_prof: struct { rounds: u64 = 0, enqueue: u64 = 0, forward: u64 = 0 } = .{};
 
-/// A watch on `links`. If it cannot start: with TF_DS_EXIT_ON_FAULT=1 the error (the exit 75 contract needs the watch,
-/// so the start fails); else null, and the ranks run as before, without it.
+/// A watch on `links`; if it cannot start, the error with TF_DS_EXIT_ON_FAULT=1 (exit 75 needs it), else null.
 fn watchLinks(gpa: std.mem.Allocator, m: *Model, links: []const ?link.Link) !?*peer_watch.Watch {
     var fds: [3]std.posix.fd_t = undefined;
     var n: usize = 0;
@@ -933,9 +929,7 @@ pub fn follow(gpa: std.mem.Allocator, m: *Model) !void {
         while (ki.next()) |b| b.free();
         kept.deinit(gpa);
     }
-    // a rank 0 that dies while this rank waits on the GPU for its writes ends this rank too (peer_watch.zig); the done
-    // frame ends the watch orderly, an error ends it failed (exit 75 when rank 0 is gone). The watch starts after the
-    // kept states, so on an error it ends (rings aborted, exit 75) before their device memory is freed
+    // a dead rank 0 ends this rank too (peer_watch.zig); started after the kept states, so it ends before they free
     const watch = try watchLinks(gpa, m, &.{peer});
     errdefer {
         if (watch) |w| _ = w.end(gpa, .failed);
diff --git a/zig/src/families/dsv41/link.zig b/zig/src/families/dsv41/link.zig
index 7ebab304a1..9556b2a1b6 100644
--- a/zig/src/families/dsv41/link.zig
+++ b/zig/src/families/dsv41/link.zig
@@ -3,11 +3,9 @@
 const std = @import("std");
 const c = std.c;
 
-/// TCP_USER_TIMEOUT (Linux): the longest time sent data can stay not acknowledged before the kernel ends the link
-/// (ETIMEDOUT: POLLERR to peer_watch.zig, PeerDown to a read or send). Without it, retransmissions run for many minutes.
+/// TCP_USER_TIMEOUT (Linux): the longest time sent data stays not acknowledged before the kernel ends the link.
 const tcp_user_timeout: u32 = 18;
-/// The same bound as the keepalive (10 s + 3 x 3 s, rounded up). With a keepalive on, Linux also ends an idle link
-/// whose probes get no answer at this time.
+/// The same bound as the keepalive (10 s + 3 x 3 s, rounded up); also the end of an idle link's unanswered probes.
 const user_timeout_ms: u32 = 20_000;
 
 pub const Link = struct {
@@ -22,14 +20,13 @@ pub const Link = struct {
 
     /// TCP_NODELAY, and SO_BUSY_POLL: a read polls the NIC's receive queue itself (up to 50 us) instead of waiting for
     /// its interrupt, whose core may be leaving a deep idle state (node to node, 64-byte frames: round trip 1.0 ms
-    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL). A peer whose node vanished (no FIN) shows
-    /// within about 20 s, to a read and to peer_watch.zig: on an idle link by the keepalive (10 s idle, then 3 probes
-    /// 3 s apart), and with sent data not acknowledged (the node lost power during a round) by TCP_USER_TIMEOUT.
+    /// blocking, 0.72 ms polling the socket, 11 us with SO_BUSY_POLL).
     fn tune(fd: c.fd_t) void {
         const one: c_int = 1;
         const busy_us: c_int = 50;
         _ = c.setsockopt(fd, c.IPPROTO.TCP, std.posix.TCP.NODELAY, std.mem.asBytes(&one), @sizeOf(c_int));
         _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.BUSY_POLL, std.mem.asBytes(&busy_us), @sizeOf(c_int));
+        // a vanished node (no FIN) shows in about 20 s to a read and to peer_watch.zig: keepalive, TCP_USER_TIMEOUT
         _ = c.setsockopt(fd, c.SOL.SOCKET, c.SO.KEEPALIVE, std.mem.asBytes(&one), @sizeOf(c_int));
         for ([_][2]u32{ .{ std.posix.TCP.KEEPIDLE, 10 }, .{ std.posix.TCP.KEEPINTVL, 3 }, .{ std.posix.TCP.KEEPCNT, 3 }, .{ tcp_user_timeout, user_timeout_ms } }) |o| {
             const v: c_int = @intCast(o[1]);
@@ -78,7 +75,7 @@ pub const Link = struct {
     fn writeAll(l: Link, bytes: []const u8) !void {
         var done: usize = 0;
         while (done < bytes.len) {
-            // NOSIGNAL: a send to a gone peer fails with PeerDown (no SIGPIPE, which would end the rank with no exit 75)
+            // NOSIGNAL: a send to a gone peer fails with PeerDown (a SIGPIPE would end the rank with no exit 75)
             const n = c.send(l.fd, bytes.ptr + done, bytes.len - done, c.MSG.NOSIGNAL);
             if (n <= 0) {
                 if (n < 0 and c.errno(n) == .INTR) continue;
diff --git a/zig/src/families/dsv41/model.zig b/zig/src/families/dsv41/model.zig
index 1560580ec9..68914aa112 100644
--- a/zig/src/families/dsv41/model.zig
+++ b/zig/src/families/dsv41/model.zig
@@ -451,15 +451,13 @@ pub const Model = struct {
         try m.ringsOk();
     }
 
-    /// After a stream sync, before its results are used: a ring aborted while the work ran (a peer gone,
-    /// peer_watch.zig) made its collects leave their waits, maybe with stale peer slots, so the work fails.
+    /// After a stream sync: a ring the peer watch aborted during the work may leave stale peer slots, so it fails.
     fn ringsOk(m: *Model) !void {
         var buf: [6]*rdma.Ring = undefined;
         if (rdma.anyFailed(m.rdmaRings(&buf))) return error.RdmaTransportFailed;
     }
 
-    /// Every RDMA ring that a GPU wait of this rank reads: TP2's ring, world 4's three rings (ring2d.zig), and the
-    /// neighbor gather's two (neighbor.zig). The peer watch aborts these; synchronize checks them.
+    /// Every RDMA ring a GPU wait of this rank reads (TP2, ring2d.zig, neighbor.zig): the peer watch aborts these.
     pub fn rdmaRings(m: *Model, buf: *[6]*rdma.Ring) []*rdma.Ring {
         var k: usize = 0;
         if (m.ring) |r| {
diff --git a/zig/src/families/dsv41/peer_watch.zig b/zig/src/families/dsv41/peer_watch.zig
index d96093ca3f..f33b6729a9 100644
--- a/zig/src/families/dsv41/peer_watch.zig
+++ b/zig/src/families/dsv41/peer_watch.zig
@@ -1,10 +1,4 @@
-//! A watch on the ranks' TCP links (release fault F3). A rank waits for its peer's frames and RDMA writes without a
-//! limit, so when the peer died while this rank was idle, or during a round whose waits got no error completion, the
-//! rank stayed up: /health said 200 and every request hung. The watch reads nothing: it polls each link for the
-//! peer's end (its process died: the kernel closed the socket) or a failed keepalive (its node is gone, link.zig).
-//! Then it aborts the RDMA rings (every wait gives up, so a live round fails and its requests get the error) and, with
-//! TF_DS_EXIT_ON_FAULT=1, ends the process with 75 after TF_DS_EXIT_GRACE, so a supervisor restarts both ranks.
-//! Watch.end is the one teardown of a watch: every way a rank ends calls it before the rings are freed.
+//! A watch on the ranks' TCP links: a gone peer aborts the RDMA rings, so no request hangs, then optionally exits 75.
 const std = @import("std");
 const rdma = @import("rdma.zig");
 const posix = std.posix;
@@ -12,8 +6,7 @@ const Allocator = std.mem.Allocator;
 
 /// POLLRDHUP (Linux): the peer closed its end.
 const rdhup: i16 = 0x2000;
-/// The peer closed its end, or the connection failed (a keepalive or a send got no answer), or the descriptor is no
-/// longer open (POLLNVAL: a poll on it would return at once, again and again).
+/// The peer closed its end, the link failed (keepalive or send), or the descriptor closed (POLLNVAL would repeat).
 const gone: i16 = rdhup | posix.POLL.HUP | posix.POLL.ERR | posix.POLL.NVAL;
 const max_links = 3;
 /// TP2's ring, world 4's three rings (ring2d.zig), or the neighbor gather's two (neighbor.zig).
@@ -54,8 +47,7 @@ pub const Watch = struct {
     exit_grace: ?u64,
     thread: ?std.Thread = null,
 
-    /// A watch thread on `fds` (rank 0's links to its followers, or a follower's one link to rank 0) that aborts
-    /// `rings` (every RDMA ring this rank's GPU waits on) when a peer is gone. `exit_grace`: faultGrace() (tests: null).
+    /// A thread that polls `fds` (this rank's peer links) and aborts `rings` when a peer is gone; see faultGrace.
     pub fn start(gpa: Allocator, fds: []const posix.fd_t, rings: []const *rdma.Ring, rank: u32, exit_grace: ?u64) !*Watch {
         const w = try gpa.create(Watch);
         errdefer gpa.destroy(w);
@@ -67,22 +59,17 @@ pub const Watch = struct {
         return w;
     }
 
-    /// True when a peer's link has ended (the watch thread saw it, or a poll that does not wait finds it now, before
-    /// the thread sees it): rank 0 reads it before its done frame (a done frame after it is no orderly stop).
+    /// True when a peer's link has ended (seen by the thread or by a poll now); rank 0 reads it before its done frame.
     pub fn faulted(w: *Watch) bool {
         return w.seen.load(.acquire) or w.linkGone();
     }
 
-    /// The one teardown, for every way a rank ends (rank 0's Lanes.deinit, a follower's done frame or error, a start
-    /// that fails after the watch started), always before the caller frees the rings (Model.close):
-    /// 1. on an error, it aborts every RDMA ring, so no GPU wait of this rank waits on a gone peer;
-    /// 2. it stops the watch thread and joins it, so no thread reads the rings after end returns;
-    /// 3. it frees the watch (the caller frees the rings and links next);
-    /// 4. on an error with a peer's link gone and TF_DS_EXIT_ON_FAULT=1, it ends the process with 75 (a normal exit
-    ///    could wait on the gone rank in Model.close).
+    /// The one teardown, for every way a rank ends, always before the caller frees the rings (Model.close).
     pub fn end(w: *Watch, gpa: Allocator, how: How) Outcome {
+        // on an error, no GPU wait of this rank waits on a gone peer
         if (how == .failed) w.abortRings("this rank ends on an error");
         w.running.store(false, .release);
+        // no thread reads the rings after end returns
         if (w.thread) |t| t.join();
         const out: Outcome = switch (how) {
             .done => .orderly,
@@ -91,6 +78,7 @@ pub const Watch = struct {
         const rank = w.rank;
         const exit_on_fault = w.exit_grace != null;
         gpa.destroy(w);
+        // a normal exit could wait on the gone rank in Model.close
         if (out == .peer_fault and exit_on_fault) {
             note("[tensorfold] rank {d}: a peer rank is gone; exit 75 (TF_DS_EXIT_ON_FAULT): a supervisor restarts both ranks\n", .{rank});
             std.c._exit(75);
@@ -104,8 +92,7 @@ pub const Watch = struct {
         for (w.rings[0..w.n_rings]) |r| r.abort(why);
     }
 
-    /// A peer's link has ended now (a poll that does not wait): the error that the caller has may have come from it
-    /// before the watch thread saw it.
+    /// A peer's link has ended now (a poll that does not wait), maybe before the watch thread saw it.
     fn linkGone(w: *Watch) bool {
         var pfds: [max_links]posix.pollfd = undefined;
         for (pfds[0..w.n], w.fds[0..w.n]) |*p, fd| p.* = .{ .fd = fd, .events = rdhup, .revents = 0 };
@@ -193,8 +180,7 @@ fn note(comptime fmt: []const u8, args: anytype) void {
     _ = std.c.send(2, line.ptr, line.len, std.c.MSG.DONTWAIT | std.c.MSG.NOSIGNAL);
 }
 
-/// TF_DS_EXIT_ON_FAULT=1: the grace in ms (TF_DS_EXIT_GRACE seconds, default 5), as lane_host.faultExit reads it. The
-/// exit 75 of a watch (Watch.start's exit_grace); null when the variable is not 1.
+/// TF_DS_EXIT_ON_FAULT=1: the exit 75 grace in ms (TF_DS_EXIT_GRACE s, default 5, as lane_host.faultExit); else null.
 pub fn faultGrace() ?u64 {
     const on = std.c.getenv("TF_DS_EXIT_ON_FAULT") orelse return null;
     if (!std.mem.eql(u8, std.mem.span(on), "1")) return null;
@@ -220,8 +206,7 @@ fn sleepMs(ms: u64) void {
     _ = std.c.nanosleep(&ts, null);
 }
 
-/// The test's peer rank: it connects, then reads frames until a "done" frame or its link ends (keep), or closes its
-/// end after `ms` (dies).
+/// The test's peer rank: it connects, then reads frames until "done" or its link ends, or closes after `die_ms`.
 const Peer = struct {
     port: u16,
     die_ms: ?u64,