diff --git a/build.zig b/build.zig index 2466155..9e06c36 100644 --- a/build.zig +++ b/build.zig @@ -79,4 +79,22 @@ pub fn build(b: *std.Build) void { if (b.args) |args| run_example.addArgs(args); const example_step = b.step("example-streamplace-chat", "replay a Streamplace broadcast's chat from the archive"); example_step.dependOn(&run_example.step); + + const smoke = b.addExecutable(.{ + .name = "example-failover-smoke", + .root_module = b.createModule(.{ + .root_source_file = b.path("examples/failover_smoke.zig"), + .target = target, + .optimize = optimize, + .link_libc = true, + .imports = &.{ + .{ .name = "jetstream", .module = mod }, + }, + }), + }); + b.installArtifact(smoke); + const run_smoke = b.addRunArtifact(smoke); + if (b.args) |args| run_smoke.addArgs(args); + const smoke_step = b.step("example-failover-smoke", "live smoke: dead primary rotates to a real fallback"); + smoke_step.dependOn(&run_smoke.step); } diff --git a/examples/failover_smoke.zig b/examples/failover_smoke.zig new file mode 100644 index 0000000..a9fcedb --- /dev/null +++ b/examples/failover_smoke.zig @@ -0,0 +1,68 @@ +//! live failover smoke: a dead primary rotates to a real fallback. +//! +//! the loopback e2e (src/loopback_test.zig) proves the mechanics against +//! fake instances; this proves the wild-network half — DNS, TLS, and a +//! production v2 server as the fallback. the primary is a blackhole by +//! construction (connection refused), so the run exercises the error- +//! threshold rotation path end to end: +//! +//! refused primary ×N → rotate → fresh live tip on the fallback → +//! deliver real events → clean stop +//! +//! run (from a box that may consume the firehose): +//! zig build example-failover-smoke -- https://stream.waow.tech + +const std = @import("std"); +const jetstream_sdk = @import("jetstream"); + +const stop_after = 10; + +const SmokeHandler = struct { + events: usize = 0, + first_seq: u64 = 0, + last_seq: u64 = 0, + + pub fn onEvent(self: *@This(), event: jetstream_sdk.Event) bool { + if (self.events == 0) self.first_seq = event.seq; + self.events += 1; + self.last_seq = event.seq; + std.debug.print(" event {d}/{d}: seq={d} did={s} kind={s}\n", .{ + self.events, stop_after, event.seq, event.did, @tagName(event.payload), + }); + return self.events < stop_after; + } + + pub fn onError(self: *@This(), err: anyerror) bool { + _ = self; + std.debug.print(" (transient: {s})\n", .{@errorName(err)}); + return true; + } +}; + +pub fn main(init: std.process.Init) !void { + const allocator = init.gpa; + const args = try init.minimal.args.toSlice(init.arena.allocator()); + if (args.len != 2) { + std.debug.print("usage: zig build example-failover-smoke -- \n", .{}); + return error.MissingFallback; + } + const fallback = args[1]; + + // loopback port 1: nothing listens there — instant connection refused + const dead_primary = "http://127.0.0.1:1"; + + std.debug.print("primary: {s} (dead by construction)\nfallback: {s}\n\n", .{ dead_primary, fallback }); + const started = std.Io.Timestamp.now(init.io, .awake).nanoseconds; + + var handler = SmokeHandler{}; + try jetstream_sdk.subscribe(init.io, allocator, .{ + .host = dead_primary, + .fallback_hosts = &.{fallback}, + .failover_after_errors = 3, + .backoff_min_ns = 100 * std.time.ns_per_ms, + .backoff_max_ns = 1 * std.time.ns_per_s, + }, &handler); + + const elapsed_ms = @divFloor(std.Io.Timestamp.now(init.io, .awake).nanoseconds - started, std.time.ns_per_ms); + std.debug.print("\nsmoke PASS: rotated off the dead primary, {d} live events from the fallback (seq {d}..{d}) in {d}ms\n", .{ handler.events, handler.first_seq, handler.last_seq, elapsed_ms }); +}