Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154//! crashpoint — named crash-injection seams at durable commit boundaries//!//! the oracle rig (tests/oracle.py, docs/upstream-harness.md) runs the//! real binary with `--crashpoint=<name>`, lets it abort at that seam,//! restarts it clean, and asserts the lifecycle converges to a healthy//! steady state. seams live at every point where upstream's crash//! matrix distinguishes an outcome. production runs never arm one, and//! hit() is a single enum compare when unarmed.//!//! abort() (not exit) so no defers, flushes, or atexit handlers run —//! the closest in-process approximation of SIGKILL.
const std = @import("std");const builtin = @import("builtin");
const log = std.log.scoped(.stream);
pub const Point = enum { // Canonical upstream Jetstream V2 lifecycle boundaries. after_repo_complete, after_merge_dst_flush_before_source_commit, after_merge_dst_seal_before_discovery, after_merge_discovery_before_cleanup, after_merge_cleanup_complete, after_bootstrap_live_close_before_seal, after_steady_phase_before_steady_run, after_compaction_rewrite_before_watermark, after_compaction_chunk_watermark, after_segment_rewrite_temp_written, after_segment_rewrite_temp_synced, after_segment_rewrite_renamed, after_segment_rewrite_dir_synced, after_segment_patch_temp_written, after_segment_patch_temp_synced, after_segment_patch_renamed, after_segment_patch_dir_synced,
/// Between the archive fsync that made a repository's rows durable and the /// metadata commit that records the repository as complete. This is the /// durability-ordering seam the whole design rests on: data first, then the /// bookkeeping that acknowledges it. A cut here must lose nothing — the rows /// are on disk and the repository simply completes again on restart. after_archive_durable_before_completion_commit,
// Additional local boundaries retained by the broader restart oracle. mid_backfill_download, after_backfill_before_phase_merging, after_phase_merging_before_live_stop, after_live_close_before_seal, after_merge_drain_before_pending, after_pending_pass, after_seal_before_discovery, after_discovery_before_cleanup, after_cleanup_remove_before_sync, after_cleanup_before_phase_write,};
const Action = enum { abort_process, await_power_cut };
var armed: ?Point = null;var action: Action = .abort_process;var target_ordinal: u64 = 1;var occurrence: std.atomic.Value(u64) = .init(0);
pub fn pointName(point: Point) []const u8 { return switch (point) { .after_repo_complete => "after-repo-complete", .after_archive_durable_before_completion_commit => "after-archive-durable-before-completion-commit", .after_merge_dst_flush_before_source_commit => "after-merge-dst-flush-before-source-commit", .after_merge_dst_seal_before_discovery => "after-merge-dst-seal-before-discovery", .after_merge_discovery_before_cleanup => "after-merge-discovery-before-cleanup", .after_merge_cleanup_complete => "after-merge-cleanup-complete", .after_bootstrap_live_close_before_seal => "after-bootstrap-live-close-before-seal", .after_steady_phase_before_steady_run => "after-steady-phase-before-steady-run", .after_compaction_rewrite_before_watermark => "after-compaction-rewrite-before-watermark", .after_compaction_chunk_watermark => "after-compaction-chunk-watermark", .after_segment_rewrite_temp_written => "after-segment-rewrite-temp-written", .after_segment_rewrite_temp_synced => "after-segment-rewrite-temp-synced", .after_segment_rewrite_renamed => "after-segment-rewrite-renamed", .after_segment_rewrite_dir_synced => "after-segment-rewrite-dir-synced", .after_segment_patch_temp_written => "after-segment-patch-temp-written", .after_segment_patch_temp_synced => "after-segment-patch-temp-synced", .after_segment_patch_renamed => "after-segment-patch-renamed", .after_segment_patch_dir_synced => "after-segment-patch-dir-synced", else => @tagName(point), };}
fn parse(name: []const u8) ?Point { inline for (comptime std.enums.values(Point)) |point| { if (std.mem.eql(u8, name, pointName(point)) or std.mem.eql(u8, name, @tagName(point))) return point; } return null;}
/// arm from the CLI (debug flag); returns false for an unknown namepub fn arm(name: []const u8) bool { return armAt(name, 1);}
/// Arm a process-abort seam at an explicit 1-based occurrence. This is the/// deterministic predicate-kill coordinate used by the upstream restart/// oracle: `(crashpoint, ordinal)` rather than wall-clock timing.pub fn armAt(name: []const u8, ordinal: u64) bool { if (ordinal == 0) return false; armed = parse(name) orelse return false; action = .abort_process; target_ordinal = ordinal; occurrence.store(0, .release); return true;}
/// Arm a lifecycle boundary for the strict block-device oracle. The real/// process stops at the requested occurrence without running defers; the/// external harness then removes power from its backing device and SIGKILLs/// the process. This action is intentionally Linux-only.pub fn armPowerCut(name: []const u8, ordinal: u64) bool { if (ordinal == 0) return false; armed = parse(name) orelse return false; action = .await_power_cut; target_ordinal = ordinal; occurrence.store(0, .release); return true;}
/// true when this seam is armed — for seams that must make some state/// durable before aborting (e.g. the not_started row a mid-download/// crash leaves behind)pub fn wouldHit(p: Point) bool { return armed == p;}
/// crash here if this seam is armedpub fn hit(p: Point) void { const a = armed orelse return; if (a != p) return; const ordinal = occurrence.fetchAdd(1, .acq_rel) + 1; if (ordinal != target_ordinal) return; switch (action) { .abort_process => { log.warn("crashpoint {s}: aborting", .{pointName(p)}); std.process.abort(); }, .await_power_cut => { log.warn("power cutpoint {s} ordinal {d}: ready", .{ pointName(p), ordinal }); if (comptime builtin.os.tag == .linux) { while (true) _ = std.os.linux.pause(); } else { std.process.abort(); } }, }}