diff --git a/build.zig b/build.zig index 70cd61d..7f74be4 100644 --- a/build.zig +++ b/build.zig @@ -144,6 +144,25 @@ pub fn build(b: *std.Build) void { const tcp_ping_pong_step = b.step("tcp_ping_pong", "Run the TCP ping-pong example"); tcp_ping_pong_step.dependOn(&run_tcp_ping_pong_cmd.step); + const fault_coverage_exe = b.addExecutable(.{ + .name = "fault_coverage", + .root_module = b.createModule(.{ + .root_source_file = b.path("examples/fault_coverage.zig"), + .target = target, + .optimize = optimize, + }), + }); + fault_coverage_exe.root_module.addImport("zigmulator", zigmulator); + const install_fault_coverage_exe = b.addInstallArtifact(fault_coverage_exe, .{}); + examples_step.dependOn(&install_fault_coverage_exe.step); + + const run_fault_coverage_cmd = b.addRunArtifact(fault_coverage_exe); + if (b.args) |args| { + run_fault_coverage_cmd.addArgs(args); + } + const fault_coverage_step = b.step("fault_coverage", "Run the fault coverage probe example"); + fault_coverage_step.dependOn(&run_fault_coverage_cmd.step); + const diagram_step = b.step("diagram", "Build the diagram generator"); const diagram_exe = b.addExecutable(.{ diff --git a/examples/fault_coverage.zig b/examples/fault_coverage.zig new file mode 100644 index 0000000..62409f0 --- /dev/null +++ b/examples/fault_coverage.zig @@ -0,0 +1,156 @@ +const std = @import("std"); +const Io = std.Io; +const net = Io.net; + +const Simulator = @import("zigmulator"); +const assertSometimes = Simulator.assertSometimes; + +const server_ip = net.Ip4Address.loopback(9090); +const server_address: net.IpAddress = .{ .ip4 = server_ip }; +const payload = "payload:v1:write-sync-rename-send"; +const ack = "ack"; + +fn serverProgram(init: std.process.Init) anyerror!void { + var server = try server_address.listen(init.io, .{}); + defer server.deinit(init.io); + + var stream = try server.accept(init.io); + defer stream.close(init.io); + + var reader_buffer: [5]u8 = undefined; + var reader = stream.reader(init.io, &reader_buffer); + + var received: [payload.len]u8 = undefined; + var copied: usize = 0; + while (copied < received.len) { + const n = try reader.interface.readSliceShort(received[copied..]); + assertSometimes(n < received.len - copied, @src(), "server saw short socket read"); + if (n == 0) { + assertSometimes(true, @src(), "server saw EOF before full request"); + return error.UnexpectedEof; + } + copied += n; + } + + assertSometimes(!std.mem.eql(u8, &received, payload), @src(), "server saw corrupted request"); + + var writer_buffer: [0]u8 = undefined; + var writer = stream.writer(init.io, &writer_buffer); + const written = try writer.interface.write(ack); + assertSometimes(written < ack.len, @src(), "server saw short socket write"); + try writer.interface.flush(); + assertSometimes(true, @src(), "server completed happy path"); +} + +fn connectWithRetry(init: std.process.Init) !net.Stream { + var attempt: usize = 0; + while (attempt < 8) : (attempt += 1) { + if (server_address.connect(init.io, .{ .mode = .stream })) |stream| { + assertSometimes(attempt > 0, @src(), "client connected after retry"); + return stream; + } else |err| switch (err) { + error.ConnectionRefused, error.HostUnreachable => { + assertSometimes(true, @src(), "client saw transient connect failure"); + try Io.sleep(init.io, .fromMilliseconds(1), .awake); + }, + else => return err, + } + } + + return error.ConnectionRefused; +} + +fn clientProgram(init: std.process.Init) anyerror!void { + try Io.Dir.cwd().createDir(init.io, "spool", .default_dir); + + const spool = try Io.Dir.cwd().openDir(init.io, "spool", .{}); + defer spool.close(init.io); + + const staged = try spool.createFile(init.io, "message.tmp", .{}); + defer staged.close(init.io); + + const file_written = try staged.writePositional(init.io, &.{payload}, 0); + assertSometimes(file_written < payload.len, @src(), "client saw short file write"); + + try staged.sync(init.io); + + try spool.rename("message.tmp", spool, "message.ready", init.io); + + const ready = try spool.openFile(init.io, "message.ready", .{}); + defer ready.close(init.io); + + var buffer: [payload.len]u8 = undefined; + const read_len = try ready.readPositionalAll(init.io, &buffer, 0); + assertSometimes(read_len < payload.len, @src(), "client saw short file read"); + assertSometimes(!std.mem.eql(u8, buffer[0..read_len], payload), @src(), "client saw corrupted file data"); + + var stream = try connectWithRetry(init); + defer stream.close(init.io); + + var writer_buffer: [0]u8 = undefined; + var writer = stream.writer(init.io, &writer_buffer); + const socket_written = try writer.interface.write(buffer[0..read_len]); + assertSometimes(socket_written < read_len, @src(), "client saw short socket write"); + try writer.interface.flush(); + + var reader_buffer: [1]u8 = undefined; + var reader = stream.reader(init.io, &reader_buffer); + var response: [ack.len]u8 = undefined; + var copied: usize = 0; + while (copied < response.len) { + const n = try reader.interface.readSliceShort(response[copied..]); + assertSometimes(n < response.len - copied, @src(), "client saw short socket read"); + if (n == 0) { + assertSometimes(true, @src(), "client saw EOF before full ack"); + return error.UnexpectedEof; + } + copied += n; + } + + assertSometimes(!std.mem.eql(u8, &response, ack), @src(), "client saw corrupted ack"); + + try spool.deleteFile(init.io, "message.ready"); + assertSometimes(true, @src(), "client completed happy path"); +} + +pub fn main(init: std.process.Init) !void { + var args = std.process.Args.Iterator.init(init.minimal.args); + _ = args.skip(); + + const first_seed_text = args.next() orelse "0"; + const first_seed = try std.fmt.parseInt(u64, first_seed_text, 10); + const count_text = args.next() orelse "1"; + const count = try std.fmt.parseInt(u64, count_text, 10); + + for (0..count) |offset| { + try runSimulation(init, first_seed + offset); + } +} + +fn runSimulation(init: std.process.Init, seed: u64) !void { + const server_addresses = [_]u32{std.mem.readInt(u32, &server_ip.bytes, .big)}; + + var sim: Simulator = undefined; + sim.init(std.heap.page_allocator, init.io, seed); + defer sim.deinit(); + + sim.enablePartitionFaults(.{ + .weights = .{ + .none = 40, + .isolate_one = 60, + .split_two_groups = 0, + }, + .min_interval_us = 500, + .max_interval_us = 2_000, + }); + + try sim.setTraceOutputFile("fault_coverage.log"); + + try sim.addExecutable("server", serverProgram); + try sim.addExecutable("client", clientProgram); + + try sim.spawn("server", .{ .addresses = &server_addresses }); + try sim.spawn("client", .{}); + + while (sim.scheduleOne()) {} +}