Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104//! Deterministic segment I/O fault seam for the offline restart oracle.//!//! Every production segment write, fsync, and commit rename passes through//! these wrappers. With no armed fault they delegate directly to std.Io. The//! oracle may arm one operation ordinal to return a real filesystem-shaped//! error before the syscall, matching upstream segment.IOFaultInjector while//! retaining Stream's real JSS implementation and on-disk files.
const std = @import("std");
const Io = std.Io;const log = std.log.scoped(.stream);
pub const Op = enum { write, sync, rename };pub const Kind = enum { enospc, eio, shortwrite };
const Fault = struct { op: ?Op = null, ordinal: u64 = 0, kind: Kind = .eio, seen: std.atomic.Value(u64) = .init(0),
fn check(self: *Fault, op: Op) !void { if (self.op == null or self.op.? != op) return; const seen = self.seen.fetchAdd(1, .monotonic) + 1; if (seen != self.ordinal) return; log.warn("segment fault: injected {s} on {s} at ordinal {d}", .{ @tagName(self.kind), @tagName(op), seen }); return switch (self.kind) { .enospc => error.NoSpaceLeft, .eio => error.InputOutput, .shortwrite => error.ShortWrite, }; }};
var fault: Fault = .{};
/// Parse `<write|sync|rename>:<positive ordinal>:<enospc|eio|shortwrite>`./// Must be called once during startup before worker threads exist.pub fn armSpec(spec: []const u8) bool { var parts = std.mem.splitScalar(u8, spec, ':'); const op = std.meta.stringToEnum(Op, parts.next() orelse return false) orelse return false; const ordinal = std.fmt.parseInt(u64, parts.next() orelse return false, 10) catch return false; const kind = std.meta.stringToEnum(Kind, parts.next() orelse return false) orelse return false; if (ordinal == 0 or parts.next() != null) return false; if (op != .write and (kind == .enospc or kind == .shortwrite)) return false; fault = .{ .op = op, .ordinal = ordinal, .kind = kind }; return true;}
pub fn disarm() void { fault = .{};}
pub fn writeStreamingAll(file: *Io.File, io: Io, bytes: []const u8) !void { try fault.check(.write); try file.writeStreamingAll(io, bytes);}
pub fn writePositionalAll(file: *Io.File, io: Io, bytes: []const u8, offset: u64) !void { try fault.check(.write); try file.writePositionalAll(io, bytes, offset);}
pub fn sync(file: *Io.File, io: Io) !void { try fault.check(.sync); try file.sync(io);}
pub fn rename(dir: Io.Dir, old: []const u8, new_dir: Io.Dir, new: []const u8, io: Io) !void { try fault.check(.rename); try dir.rename(old, new_dir, new, io);}
/// fsync a directory (making a completed rename durable). std.Io.Dir has no/// sync, and borrowing dir.handle into an Io.File assumes the handle is an/// fsync-able fd — false where the backend opens directories O_PATH (observed/// EBADF under the nixery CI runner). Reopen "." beneath the directory as a/// real read-only fd and fsync that. Counts one `.sync` fault ordinal, same/// as the borrowed-handle path it replaces.pub fn syncDir(dir: Io.Dir, io: Io) !void { try fault.check(.sync); const fd = std.posix.openat(dir.handle, ".", .{ .ACCMODE = .RDONLY, .DIRECTORY = true }, 0) catch return error.Unexpected; var file: Io.File = .{ .handle = fd, .flags = .{ .nonblocking = false } }; defer file.close(io); try file.sync(io);}
test "fault spec is strict and fires only at the selected operation ordinal" { const testing = std.testing; disarm(); defer disarm(); try testing.expect(!armSpec("")); try testing.expect(!armSpec("write:0:enospc")); try testing.expect(!armSpec("sync:1:shortwrite")); try testing.expect(!armSpec("rename:1:eio:extra")); try testing.expect(armSpec("write:2:enospc")); try fault.check(.sync); try fault.check(.write); try testing.expectError(error.NoSpaceLeft, fault.check(.write)); try fault.check(.write);}