fix(v2): exit the validator when a service fails in threaded mode

build.zig

@@ -50,7 +50,14 @@ pub fn build(b: *Build) !void {
 
     // bb-test
     const bb_test_step = b.step("bb-test", "Run black box tests.");
-    for (tools.black_box_tests) |bbt| bbt.addToStep(bb_test_step);
+    for (tools.black_box_tests, Tools.black_box_test_descriptions) |bbt, description| {
+        bbt.addToStep(bb_test_step);
+        const test_step = b.step(
+            b.fmt("bb-test-{s}", .{description.name}),
+            b.fmt("Run the {s} black box test.", .{description.name}),
+        );
+        bbt.addToStep(test_step);
+    }
 
     // shred-stream
     const shred_stream_step = b.step("shred-stream", "Stream shreds from an Agave ledger");
@@ -617,6 +624,11 @@ const Tools = struct {
             .root_source_file = "v2/tests/replay/main.zig",
             .services = &.{ "shred_receiver", "replay", "telemetry" },
         },
+        .{
+            .name = "threaded-exit",
+            .root_source_file = "v2/tests/threaded_exit/main.zig",
+            .services = &.{ "failing", "healthy" },
+        },
     };
 
     pub fn init(
@@ -738,9 +750,34 @@ const Tools = struct {
             }, .{ .dest_dir = test_install_dir });
 
             for (description.services) |service_name| {
-                exe.compile.linkLibrary(for (sig.service_libs) |entry| {
+                if (for (sig.service_libs) |entry| {
                     if (std.mem.eql(u8, entry.name, service_name)) break entry.lib;
-                } else std.debug.panic("unknown service '{s}'", .{service_name}));
+                } else null) |service_lib| {
+                    exe.compile.linkLibrary(service_lib);
+                    continue;
+                }
+
+                const service_mod = b.createModule(.{
+                    .root_source_file = b.path(b.fmt(
+                        "v2/tests/threaded_exit/{s}.zig",
+                        .{service_name},
+                    )),
+                    .target = config.target,
+                    .optimize = config.optimize,
+                    .single_threaded = true,
+                    .omit_frame_pointer = false,
+                    .error_tracing = true,
+                    .imports = &.{
+                        .{ .name = "lib", .module = sig.lib },
+                        .{ .name = "start_service", .module = sig.start_service },
+                        .{ .name = "tracy", .module = deps.tracy },
+                    },
+                });
+                exe.compile.linkLibrary(b.addLibrary(.{
+                    .name = b.fmt("test-{s}", .{service_name}),
+                    .root_module = service_mod,
+                    .use_llvm = config.use_llvm,
+                }));
             }
         }
 

v2/init/start_service.zig

@@ -303,8 +303,12 @@ fn notifyThreadCrash() void {
     crash_fn(panic_state.thread_crash_ctx, service_idx);
 }
 
+/// Threaded-mode callers must notify the parent before calling this function (#1721).
 fn abort() noreturn {
-    std.os.linux.exit(255);
+    if (panic_state.thread_crash_fn != null) {
+        std.os.linux.exit(255);
+    }
+    std.os.linux.exit_group(255);
 }
 
 /// Assumes x86-64, and built with frame pointers

v2/init/topology.zig

@@ -44,6 +44,8 @@ pub const max_regions_per_service = lib.ipc.ResolvedArgs.max_regions;
 
 pub const Mode = enum { sandboxed, threaded };
 
+pub const ExitStatus = enum { clean, failed };
+
 /// Pair of structs declaring what regions a service consumes. `ReadOnly` and
 /// `ReadWrite` are each a struct of typed pointers (e.g. `*const Config`, `*Pair`).
 /// `ServiceRegions` mirrors them with corresponding `Region(T).Initialized` fields.
@@ -334,16 +336,32 @@ pub fn Children(Topo: type) type {
             for (self.slice()) |*svc| svc.activity_view.cancel();
         }
 
-        /// Block until the first service exits, then dump its diagnostics.
+        /// Block until the first service exits, dump its diagnostics, and report
+        /// whether it failed.
         /// If `timeout_ns_opt` is non-null, returns `error.Timeout` if no service exits in time.
-        pub fn wait(self: *Children(Topo), timeout_ns_opt: ?u64) error{Timeout}!void {
-            switch (self.mode) {
+        pub fn wait(
+            self: *Children(Topo),
+            timeout_ns_opt: ?u64,
+        ) error{Timeout}!ExitStatus {
+            return switch (self.mode) {
                 .sandboxed => try self.waitSandboxed(timeout_ns_opt),
                 .threaded => try self.waitThreaded(timeout_ns_opt),
-            }
+            };
+        }
+
+        /// Cooperatively cancel every service, allow a bounded grace period,
+        /// then terminate the entire process without running libc destructors.
+        /// Must not be called after `cancel`.
+        pub fn shutdown(self: *Children(Topo), code: u8) noreturn {
+            self.cancel();
+            std.Thread.sleep(std.time.ns_per_s);
+            linux.exit_group(code);
         }
 
-        fn waitSandboxed(self: *Children(Topo), timeout_ns_opt: ?u64) error{Timeout}!void {
+        fn waitSandboxed(
+            self: *Children(Topo),
+            timeout_ns_opt: ?u64,
+        ) error{Timeout}!ExitStatus {
             const timeout_pid_opt = if (timeout_ns_opt) |timeout_ns|
                 spawnSandboxedTimeout(timeout_ns)
             else
@@ -361,13 +379,15 @@ pub fn Children(Topo: type) type {
 
             for (self.slice()) |*svc| {
                 if (svc.slot.sandboxed != exited_pid) continue;
-                dumpOnExit(&svc.runner.exit, svc.label, exited_pid, status);
-                return;
+                return dumpOnExit(&svc.runner.exit, svc.label, exited_pid, status);
             }
             std.debug.panic("Unknown child pid {} exited", .{exited_pid});
         }
 
-        fn waitThreaded(self: *Children(Topo), timeout_ns_opt: ?u64) error{Timeout}!void {
+        fn waitThreaded(
+            self: *Children(Topo),
+            timeout_ns_opt: ?u64,
+        ) error{Timeout}!ExitStatus {
             // Wait for first service to exit
             if (timeout_ns_opt) |timeout_ns| {
                 self.thread_exit.reset_event.timedWait(timeout_ns) catch {};
@@ -379,7 +399,7 @@ pub fn Children(Topo: type) type {
             if (exited_idx == std.math.maxInt(u16)) return error.Timeout;
 
             const svc = &self.slice()[exited_idx];
-            dumpOnExit(&svc.runner.exit, svc.label, 0, 0);
+            return dumpOnExit(&svc.runner.exit, svc.label, 0, 0);
         }
 
         fn slice(self: *Children(Topo)) []Service {
@@ -654,7 +674,20 @@ fn dumpOnExit(
     label: [:0]const u8,
     pid: linux.pid_t,
     status: u32,
-) void {
+) ExitStatus {
+    // Every sandboxed service exits with 255 after completing or aborting the
+    // service protocol. Any other exit code means it died before recording metadata.
+    const service_exit_code = 255;
+    // A pid of zero identifies threaded mode, whose synthetic status must stay excluded.
+    const died_outside_protocol = pid != 0 and
+        linux.W.IFEXITED(status) and
+        linux.W.EXITSTATUS(status) != service_exit_code;
+    const failed = meta.panicMsg() != null or
+        meta.errorName() != null or
+        meta.faultMsg() != null or
+        linux.W.TERMSIG(status) != 0 or
+        died_outside_protocol;
+
     if (meta.panicMsg()) |panic_msg| {
         std.log.err(
             "Service `{s}` (pid: {}) panicked with message: {s}",
@@ -679,6 +712,12 @@ fn dumpOnExit(
             .{ label, pid, linux.W.TERMSIG(status) },
         );
     }
+    if (died_outside_protocol) {
+        std.log.err(
+            "Service `{s}` (pid: {}) exited outside the service protocol with code {}",
+            .{ label, pid, linux.W.EXITSTATUS(status) },
+        );
+    }
     if (meta.errorReturnStackTrace()) |trace| {
         std.log.err("Error trace:", .{});
         std.debug.dumpStackTrace(trace);
@@ -691,4 +730,6 @@ fn dumpOnExit(
         std.log.err("Fault trace:", .{});
         std.debug.dumpStackTrace(trace);
     }
+
+    return if (failed) .failed else .clean;
 }

v2/main.zig

@@ -382,9 +382,10 @@ pub fn main() !void {
             .rw = .{ .exec_req_response = exec_req_response.finish() },
         },
     });
-    try children.wait(null);
+    const exit_status = try children.wait(null);
 
     tracy.message("exiting");
+    children.shutdown(if (exit_status == .failed) 1 else 0);
 }
 
 fn populateSnapshotConfig(

v2/tests/gossip/main.zig

@@ -111,7 +111,7 @@ pub fn main() !void {
     children.cancel();
 
     // and then actually wait for the services to exit
-    try children.wait(2 * std.time.ns_per_s);
+    _ = try children.wait(2 * std.time.ns_per_s);
 
     // -- Verify outgoing messages -- //
 

v2/tests/replay/main.zig

@@ -172,7 +172,7 @@ pub fn main() !void {
     );
 
     spawned.cancel();
-    try spawned.wait(2 * std.time.ns_per_s);
+    _ = try spawned.wait(2 * std.time.ns_per_s);
 }
 
 fn resignPackets(

v2/tests/threaded_exit/failing.zig

@@ -0,0 +1,17 @@
+const start = @import("start_service");
+const lib = @import("lib");
+
+comptime {
+    _ = start;
+}
+
+pub const name = .failing;
+pub const panic = start.panic;
+pub const std_options = start.options;
+
+pub const ReadOnly = struct {};
+pub const ReadWrite = struct {};
+
+pub fn serviceMain(_: lib.runner.Connection, _: ReadOnly, _: ReadWrite) !noreturn {
+    return error.IntentionalFailure;
+}

v2/tests/threaded_exit/healthy.zig

@@ -0,0 +1,25 @@
+const std = @import("std");
+const start = @import("start_service");
+const lib = @import("lib");
+const tracy = @import("tracy");
+
+comptime {
+    _ = start;
+}
+
+pub const name = .healthy;
+pub const panic = start.panic;
+pub const std_options = start.options;
+
+pub const ReadOnly = struct {};
+pub const ReadWrite = struct {};
+
+pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, _: ReadWrite) !noreturn {
+    while (true) {
+        const zone = tracy.Zone.init(@src(), .{});
+        defer zone.deinit();
+
+        try runner.activity.checkCanceled();
+        std.Thread.sleep(std.time.ns_per_ms);
+    }
+}

v2/tests/threaded_exit/main.zig

@@ -0,0 +1,50 @@
+const std = @import("std");
+const topology = @import("topology");
+
+const EmptySpec: topology.ServiceSpec = .{
+    .ReadOnly = struct {},
+    .ReadWrite = struct {},
+};
+
+const Topology = struct {
+    failing: topology.ServiceRegions(EmptySpec),
+    healthy: topology.ServiceRegions(EmptySpec),
+};
+
+pub fn main() !void {
+    var args = std.process.args();
+    _ = args.next();
+    if (args.next()) |arg| {
+        if (std.mem.eql(u8, arg, "--child")) return runChild();
+    }
+
+    var dba_state: std.heap.DebugAllocator(.{}) = .init;
+    defer _ = dba_state.deinit();
+    const allocator = dba_state.allocator();
+
+    const self_exe = try std.fs.selfExePathAlloc(allocator);
+    defer allocator.free(self_exe);
+
+    const result = try std.process.Child.run(.{
+        .allocator = allocator,
+        .argv = &.{ "timeout", "30", self_exe, "--child" },
+        .max_output_bytes = 1024 * 1024,
+    });
+    defer allocator.free(result.stdout);
+    defer allocator.free(result.stderr);
+
+    std.debug.print("{s}{s}", .{ result.stdout, result.stderr });
+    try std.testing.expect(std.mem.indexOf(u8, result.stderr, "IntentionalFailure") != null);
+    try std.testing.expectEqual(std.process.Child.Term{ .Exited = 1 }, result.term);
+}
+
+fn runChild() !void {
+    var children: topology.Children(Topology) = undefined;
+    try children.spawn(.threaded, .{
+        .failing = .{ .ro = .{}, .rw = .{} },
+        .healthy = .{ .ro = .{}, .rw = .{} },
+    });
+
+    const exit_status = try children.wait(null);
+    children.shutdown(if (exit_status == .failed) 1 else 0);
+}