diff --git a/build.zig b/build.zig index 7b5a232..337801f 100644 --- a/build.zig +++ b/build.zig @@ -18,12 +18,26 @@ pub fn build(b: *std.Build) void { .{ .name = "websocket", .module = websocket.module("websocket") }, }; + // build provenance for the stream_build_info canary metric + const build_options = b.addOptions(); + build_options.addOption([]const u8, "git_sha", git_sha: { + var code: u8 = 0; + const result = b.runAllowFail(&.{ "git", "rev-parse", "--short", "HEAD" }, &code, .ignore); + if (result) |output| { + break :git_sha std.mem.trimEnd(u8, output, "\n \t"); + } else |_| { + break :git_sha "unknown"; + } + }); + build_options.addOption([]const u8, "optimize", @tagName(optimize)); + const exe_mod = b.createModule(.{ .root_source_file = b.path("src/main.zig"), .target = target, .optimize = optimize, .imports = imports, }); + exe_mod.addImport("build_options", build_options.createModule()); exe_mod.link_libc = true; const exe = b.addExecutable(.{ .name = "stream", @@ -43,6 +57,7 @@ pub fn build(b: *std.Build) void { .optimize = optimize, .imports = imports, }); + test_mod.addImport("build_options", build_options.createModule()); test_mod.link_libc = true; const t = b.addTest(.{ .root_module = test_mod, diff --git a/docs/lessons-from-zlay.md b/docs/lessons-from-zlay.md new file mode 100644 index 0000000..3eb292b --- /dev/null +++ b/docs/lessons-from-zlay.md @@ -0,0 +1,41 @@ +# lessons from zlay (devlogs 006–010 + incident docs) + +distilled 2026-07-08 from `zat/devlog/006–010` and `zlay/docs/{gotchas,allocation-audit,incident-*}.md`. +read those for the full stories; these are the rules stream inherits. + +1. **Io.Threaded + ReleaseSafe + c_allocator + 8 MB thread stacks** — each bought with an incident. + ReleaseFast hid heap corruption for days (008) and has a known double-free; GPA never returns + memory to the OS; 2–4 MB stacks overflow in ReleaseSafe TLS/CBOR chains (134 KiB for + tls.Client.init alone, 007); default 16 MB stacks map 44 GB VM at ~2.7k threads (006). +2. **never mix Io backends** — Io.Mutex/Condition/io.sleep crash at runtime (NULL Thread.current) + when called from a different backend than they were created under. cross-context comms = pure + atomics (DbRequestQueue pattern) (008). +3. **durability before emission** — never broadcast a seq a subscriber could later replay from + before it is durable. atomic committed_seq watermark, NOT broadcast-inside-flush (that + serialized 2,670 producers and OOM'd). recovery scans validate every record and truncate torn + tails (incident-2026-05-31). applies the day the segment archive lands. +4. **socket close ownership** — only the thread that owns a consumer's write loop closes the + socket; everyone else flips an `alive` atomic (EBADF race, 009). stream's Handler.close joins + the subscriber thread before conn teardown — keep it that way. +5. **serialize ws client writes; background loops must be cancellation-cooperative** — never + `catch {}` an io.sleep (swallowed error.Canceled = use-after-free, 008); throttle mass + reconnects in batches. +6. **assume TCP splits everything** — handshake/CRLF/HTTP parsing must tolerate byte-at-a-time + delivery; one CRLF off-by-one was misblamed on the fiber scheduler for days (006, 008). +7. **thin reader threads** — websocket read + queue only; decode/verify/fan-out on a shared pool. + 0.45 vs 3.9 MiB per-reader RSS is run-vs-OOM (007). stream's single ingest thread currently + decodes+verifies inline: fine at one upstream, revisit if per-PDS crawling ever lands. +8. **memory observability from day one** — smaps-based RSS (mallinfo overflows at 2 GiB), + malloc_info all-arena gauges, MALLOC_ARENA_MAX=4 in deploy env, build_info{git_sha,optimize} + canary metric. balanced allocs + growing RSS = glibc arena fragmentation, not a leak + (allocation-audit, incident-2026-05-30). +9. **build/deploy** — `-Dtarget=x86_64-linux-gnu` (musl SIGILLs C++ deps), native build on the + server, immutable SHA tags, probes on the concurrent port, probe-path+image changes atomic + (006, incident-2026-03-04). +10. **the network is input** — DID/handle resolution is SSRF surface: reject private hosts, + DNS-preflight, no redirects, dial the checked address (010). XRPC errors are data: parse the + error envelope, retry only transient with capped jittered backoff, honor retry-after. +11. **observe → measure → enforce** for any new verification, behind a flag; beware checks that + trivially pass on empty input (007's extractOps bug). +12. **incident discipline** — one variable per experiment, verify for hours, keep a last-known-good + SHA, roll back when hypothesis #3 fails (incident-2026-03-04, 009). diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index 7ab1eb9..fd4fe92 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -8,6 +8,7 @@ const std = @import("std"); const zat = @import("zat"); const convert = @import("convert.zig"); const cursor_mod = @import("cursor.zig"); +const metrics = @import("metrics.zig"); const tail_mod = @import("tail.zig"); const verify = @import("verify.zig"); @@ -25,6 +26,7 @@ pub const Consumer = struct { to_stdout: bool = false, verifier: ?*verify.Verifier = null, cursor_store: ?*cursor_mod.Store = null, + stats: ?*metrics.Stats = null, drops: convert.Drops = .initFill(0), /// blocks forever; zat handles reconnect with backoff and host rotation. @@ -49,6 +51,7 @@ pub const Consumer = struct { const alloc = arena.allocator(); const time_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); + if (self.stats) |st| _ = st.events_total.fetchAdd(1, .monotonic); if (self.verifier) |v| { switch (event) { @@ -56,6 +59,7 @@ pub const Consumer = struct { // proven-bad after a fresh key resolve: drop the event .invalid_signature => { self.drops.getPtr(.invalid_signature).* += 1; + if (self.stats) |st| _ = st.drops.getPtr(.invalid_signature).fetchAdd(1, .monotonic); log.warn("dropped commit with invalid signature: {s} seq={d}", .{ c.repo, c.seq }); return self.noteSeq(event, time_us); }, @@ -70,6 +74,10 @@ pub const Consumer = struct { const drops_before = self.drops; try convert.convert(alloc, event, time_us, &frames, &self.drops); logNewDrops(&drops_before, &self.drops); + if (self.stats) |st| { + st.addDrops(&drops_before, &self.drops); + _ = st.frames_total.fetchAdd(frames.items.len, .monotonic); + } for (frames.items) |frame| { try self.tail.append(frame.kind, frame.did, frame.collection, frame.time_us, frame.json); @@ -88,9 +96,9 @@ pub const Consumer = struct { } fn noteSeq(self: *Consumer, event: zat.FirehoseEvent, time_us: i64) void { - if (self.cursor_store) |cs| { - if (event.seq()) |s| cs.update(self.io, s, time_us); - } + const seq = event.seq() orelse return; + if (self.stats) |st| st.upstream_seq.store(seq, .monotonic); + if (self.cursor_store) |cs| cs.update(self.io, seq, time_us); } pub fn close(_: *Consumer) void { diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig new file mode 100644 index 0000000..70e12ed --- /dev/null +++ b/src/internal/metrics.zig @@ -0,0 +1,135 @@ +//! shared counters + prometheus text formatting +//! +//! observability from day one (docs/lessons-from-zlay.md #8, #11): the +//! build_info canary proves which binary runs, and verification counters +//! prove verification is actually verifying — a check that silently passes +//! everything is indistinguishable from a working one without them. +//! +//! all counters are atomics: single-writer ingest thread, read by whichever +//! server thread serves /metrics. + +const std = @import("std"); +const build_options = @import("build_options"); +const convert = @import("convert.zig"); + +const Io = std.Io; + +pub const Counter = std.atomic.Value(u64); + +pub const Stats = struct { + start_time_s: i64 = 0, + events_total: Counter = .init(0), + frames_total: Counter = .init(0), + upstream_seq: std.atomic.Value(i64) = .init(0), + drops: std.enums.EnumArray(convert.DropReason, Counter) = .initFill(.init(0)), + verify_valid: Counter = .init(0), + verify_invalid: Counter = .init(0), + verify_unverified: Counter = .init(0), + verify_cache_hits: Counter = .init(0), + verify_cache_misses: Counter = .init(0), + + pub fn addDrops(self: *Stats, before: *const convert.Drops, after: *const convert.Drops) void { + inline for (comptime std.enums.values(convert.DropReason)) |reason| { + const delta = after.get(reason) - before.get(reason); + if (delta > 0) _ = self.drops.getPtr(reason).fetchAdd(delta, .monotonic); + } + } +}; + +pub const TailGauges = struct { + entries: usize, + bytes: usize, + base: u64, + tip: u64, +}; + +pub fn format( + buf: []u8, + stats: *const Stats, + subscribers_active: u32, + tail: TailGauges, + now_s: i64, +) []const u8 { + var w: Io.Writer = .fixed(buf); + + // canary: proves what binary is running (lessons #8) + w.print( + \\# TYPE stream_build_info gauge + \\stream_build_info{{git_sha="{s}",optimize="{s}"}} 1 + \\# TYPE stream_uptime_seconds gauge + \\stream_uptime_seconds {d} + \\# TYPE stream_events_total counter + \\stream_events_total {d} + \\# TYPE stream_frames_total counter + \\stream_frames_total {d} + \\# TYPE stream_upstream_seq gauge + \\stream_upstream_seq {d} + \\# TYPE stream_subscribers_active gauge + \\stream_subscribers_active {d} + \\# TYPE stream_tail_entries gauge + \\stream_tail_entries {d} + \\# TYPE stream_tail_bytes gauge + \\stream_tail_bytes {d} + \\# TYPE stream_tail_base gauge + \\stream_tail_base {d} + \\# TYPE stream_tail_tip gauge + \\stream_tail_tip {d} + \\ + , .{ + build_options.git_sha, + build_options.optimize, + now_s - stats.start_time_s, + stats.events_total.load(.monotonic), + stats.frames_total.load(.monotonic), + stats.upstream_seq.load(.monotonic), + subscribers_active, + tail.entries, + tail.bytes, + tail.base, + tail.tip, + }) catch {}; + + w.print( + \\# TYPE stream_verify_total counter + \\stream_verify_total{{result="valid"}} {d} + \\stream_verify_total{{result="invalid"}} {d} + \\stream_verify_total{{result="unverified"}} {d} + \\# TYPE stream_verify_key_cache_total counter + \\stream_verify_key_cache_total{{result="hit"}} {d} + \\stream_verify_key_cache_total{{result="miss"}} {d} + \\ + , .{ + stats.verify_valid.load(.monotonic), + stats.verify_invalid.load(.monotonic), + stats.verify_unverified.load(.monotonic), + stats.verify_cache_hits.load(.monotonic), + stats.verify_cache_misses.load(.monotonic), + }) catch {}; + + w.print("# TYPE stream_dropped_events_total counter\n", .{}) catch {}; + inline for (comptime std.enums.values(convert.DropReason)) |reason| { + w.print("stream_dropped_events_total{{reason=\"{s}\"}} {d}\n", .{ + @tagName(reason), + stats.drops.getPtrConst(reason).load(.monotonic), + }) catch {}; + } + + return w.buffered(); +} + +// === tests === + +test "format renders counters" { + var stats: Stats = .{}; + _ = stats.events_total.fetchAdd(7, .monotonic); + _ = stats.verify_valid.fetchAdd(3, .monotonic); + _ = stats.drops.getPtr(.invalid_signature).fetchAdd(1, .monotonic); + + var buf: [8192]u8 = undefined; + const out = format(&buf, &stats, 2, .{ .entries = 5, .bytes = 100, .base = 1, .tip = 6 }, 60); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_events_total 7") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_verify_total{result=\"valid\"} 3") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_dropped_events_total{reason=\"invalid_signature\"} 1") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_build_info{git_sha=") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_subscribers_active 2") != null); +} diff --git a/src/internal/server.zig b/src/internal/server.zig index 6750adb..c495a45 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -11,6 +11,7 @@ const std = @import("std"); const websocket = @import("websocket"); const filter_mod = @import("filter.zig"); +const metrics = @import("metrics.zig"); const tail_mod = @import("tail.zig"); const wire = @import("wire.zig"); @@ -18,10 +19,14 @@ const Io = std.Io; const Allocator = std.mem.Allocator; const log = std.log.scoped(.stream); +/// 8 MB, matching the Io.Threaded stacks (docs/lessons-from-zlay.md #1) +const subscriber_stack_size: usize = 8 * 1024 * 1024; + pub const Hub = struct { allocator: Allocator, io: Io, tail: *tail_mod.Tail, + stats: *metrics.Stats, active: std.atomic.Value(u32) = .init(0), }; @@ -148,7 +153,7 @@ pub const Handler = struct { self.pending_filter = null; // ownership moved to the subscriber self.subscriber = sub; _ = hub.active.fetchAdd(1, .monotonic); - sub.thread = std.Thread.spawn(.{}, Subscriber.run, .{sub}) catch |err| { + sub.thread = std.Thread.spawn(.{ .stack_size = subscriber_stack_size }, Subscriber.run, .{sub}) catch |err| { _ = hub.active.fetchSub(1, .monotonic); sub.filter.deinit(); hub.allocator.destroy(sub); @@ -248,8 +253,48 @@ pub const Handler = struct { sub.stopped.store(true, .release); self.hub.tail.wake(); } + + /// plain-HTTP requests on the websocket port: /metrics and /healthz + pub fn httpFallback( + conn: *websocket.Conn, + method: []const u8, + url: []const u8, + body: []const u8, + headers: *const websocket.Handshake.KeyValue, + hub: *Hub, + ) void { + _ = body; + _ = headers; + const path = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[0..i] else url; + if (!std.mem.eql(u8, method, "GET")) return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); + if (std.mem.eql(u8, path, "/healthz")) return respond(conn, "200 OK", "text/plain", "ok\n"); + if (std.mem.eql(u8, path, "/metrics")) { + const g = hub.tail.gauges(); + const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); + var buf: [16 * 1024]u8 = undefined; + const out = metrics.format(&buf, hub.stats, hub.active.load(.monotonic), .{ + .entries = g.entries, + .bytes = g.bytes, + .base = g.base, + .tip = g.tip, + }, now_s); + return respond(conn, "200 OK", "text/plain; version=0.0.4", out); + } + respond(conn, "404 Not Found", "text/plain", "not found\n"); + } }; +fn respond(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8) void { + var buf: [256]u8 = undefined; + const header = std.fmt.bufPrint( + &buf, + "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ status, content_type, resp_body.len }, + ) catch return; + conn.writeFramed(header) catch return; + if (resp_body.len > 0) conn.writeFramed(resp_body) catch return; +} + fn addAll( filter: *filter_mod.Filter, v: std.json.Value, diff --git a/src/internal/tail.zig b/src/internal/tail.zig index e9b4d70..41a25f0 100644 --- a/src/internal/tail.zig +++ b/src/internal/tail.zig @@ -75,6 +75,19 @@ pub const Tail = struct { return e.json.len + e.did.len + e.collection.len + 64; } + pub const Gauges = struct { entries: usize, bytes: usize, base: u64, tip: u64 }; + + pub fn gauges(self: *Tail) Gauges { + self.mutex.lockUncancelable(self.io); + defer self.mutex.unlock(self.io); + return .{ + .entries = self.entries.items.len, + .bytes = self.cur_bytes, + .base = self.base, + .tip = self.base + self.entries.items.len, + }; + } + /// index of the next frame to be appended (start here for live tail). pub fn tip(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); diff --git a/src/internal/verify.zig b/src/internal/verify.zig index 15aaa9a..53502ea 100644 --- a/src/internal/verify.zig +++ b/src/internal/verify.zig @@ -19,6 +19,7 @@ const std = @import("std"); const zat = @import("zat"); +const metrics = @import("metrics.zig"); const Io = std.Io; const Allocator = std.mem.Allocator; @@ -38,21 +39,13 @@ pub const Result = enum { unverified, }; -pub const Stats = struct { - verified: u64 = 0, - invalid: u64 = 0, - unverified: u64 = 0, - cache_hits: u64 = 0, - cache_misses: u64 = 0, -}; - pub const Verifier = struct { allocator: Allocator, io: Io, resolver: zat.DidResolver, cache: std.StringHashMapUnmanaged(CachedKey) = .empty, max_cache: u32 = 100_000, - stats: Stats = .{}, + stats: ?*metrics.Stats = null, /// plc_url: base URL of the PLC directory. for the simulator this is its /// http address (it serves GET /did:...); production is plc.directory. @@ -79,26 +72,30 @@ pub const Verifier = struct { /// against `did`'s signing key. pub fn verifyCommit(self: *Verifier, did: []const u8, blocks: []const u8) Result { const key = self.getKey(did, false) orelse { - self.stats.unverified += 1; + self.count("verify_unverified"); return .unverified; }; if (self.checkSignature(did, blocks, key)) { - self.stats.verified += 1; + self.count("verify_valid"); return .valid; } // key may have rotated: re-resolve once and retry const fresh = self.getKey(did, true) orelse { - self.stats.unverified += 1; + self.count("verify_unverified"); return .unverified; }; if (self.checkSignature(did, blocks, fresh)) { - self.stats.verified += 1; + self.count("verify_valid"); return .valid; } - self.stats.invalid += 1; + self.count("verify_invalid"); return .invalid_signature; } + fn count(self: *Verifier, comptime field: []const u8) void { + if (self.stats) |st| _ = @field(st, field).fetchAdd(1, .monotonic); + } + /// rotation signal: #identity events invalidate the cached key pub fn evict(self: *Verifier, did: []const u8) void { if (self.cache.fetchRemove(did)) |kv| self.allocator.free(kv.key); @@ -122,10 +119,10 @@ pub const Verifier = struct { fn getKey(self: *Verifier, did: []const u8, force_resolve: bool) ?CachedKey { if (!force_resolve) { if (self.cache.get(did)) |k| { - self.stats.cache_hits += 1; + self.count("verify_cache_hits"); return k; } - self.stats.cache_misses += 1; + self.count("verify_cache_misses"); } else { self.evict(did); } diff --git a/src/main.zig b/src/main.zig index 2d30455..6f9eb7e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -5,6 +5,7 @@ const websocket = @import("websocket"); const cursor_mod = @import("internal/cursor.zig"); const ingest = @import("internal/ingest.zig"); const verify = @import("internal/verify.zig"); +const metrics = @import("internal/metrics.zig"); const server_mod = @import("internal/server.zig"); const tail_mod = @import("internal/tail.zig"); @@ -13,10 +14,15 @@ const log = std.log.scoped(.stream); const default_tail_bytes: usize = 256 * 1024 * 1024; +/// 8 MB: ReleaseSafe inlining makes TLS/CBOR/crypto call chains deep enough +/// to overflow 2-4 MB stacks; zig's 16 MB default maps needless VM at scale +/// (docs/lessons-from-zlay.md #1) +const default_stack_size: usize = 8 * 1024 * 1024; + pub fn main(init: std.process.Init.Minimal) !void { const allocator = std.heap.c_allocator; - var threaded: Io.Threaded = .init(allocator, .{}); + var threaded: Io.Threaded = .init(allocator, .{ .stack_size = default_stack_size }); defer threaded.deinit(); const io = threaded.io(); @@ -55,7 +61,10 @@ pub fn main(init: std.process.Init.Minimal) !void { var hot_tail = tail_mod.Tail.init(allocator, io, default_tail_bytes); defer hot_tail.deinit(); - var hub: server_mod.Hub = .{ .allocator = allocator, .io = io, .tail = &hot_tail }; + var stats: metrics.Stats = .{ + .start_time_s = @divTrunc(Io.Timestamp.now(io, .real).toMicroseconds(), std.time.us_per_s), + }; + var hub: server_mod.Hub = .{ .allocator = allocator, .io = io, .tail = &hot_tail, .stats = &stats }; var ws_server = try websocket.Server(server_mod.Handler).init(allocator, io, .{ .port = port, @@ -80,6 +89,7 @@ pub fn main(init: std.process.Init.Minimal) !void { } var verifier: ?verify.Verifier = if (do_verify) verify.Verifier.init(allocator, io, plc_url) else null; + if (verifier) |*v| v.stats = &stats; defer if (verifier) |*v| v.deinit(); if (do_verify) log.info("signature verification on (plc: {s})", .{plc_url}); @@ -92,6 +102,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .to_stdout = to_stdout, .verifier = if (verifier) |*v| v else null, .cursor_store = &cursor_store, + .stats = &stats, }; try consumer.run(); } diff --git a/src/tests.zig b/src/tests.zig index c066290..9ad1a7c 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -2,6 +2,7 @@ comptime { _ = @import("main.zig"); _ = @import("internal/ingest.zig"); _ = @import("internal/cursor.zig"); + _ = @import("internal/metrics.zig"); _ = @import("internal/verify.zig"); _ = @import("internal/wire.zig"); _ = @import("internal/convert.zig"); diff --git a/tests/e2e.py b/tests/e2e.py index fee21e2..e98e331 100644 --- a/tests/e2e.py +++ b/tests/e2e.py @@ -52,6 +52,30 @@ async def basics(): pass print("basics: PASS") +def check_metrics(): + import urllib.request + body = urllib.request.urlopen("http://localhost:6008/metrics").read().decode() + assert 'stream_build_info{git_sha="' in body + def val(name): + for line in body.splitlines(): + if line.startswith(name + " ") or line.startswith(name + "{"): + pass + for line in body.splitlines(): + if line.split(" ")[0] == name: + return float(line.split(" ")[1]) + raise AssertionError(f"metric {name} missing") + assert val("stream_events_total") > 0 + assert val("stream_frames_total") > 0 + # a verifier that silently passes everything unverified is + # indistinguishable from a working one without this + valid = [l for l in body.splitlines() if l.startswith('stream_verify_total{result="valid"}')] + assert valid and float(valid[0].split(" ")[1]) > 0, "verification never verified anything" + invalid = [l for l in body.splitlines() if l.startswith('stream_verify_total{result="invalid"}')] + assert float(invalid[0].split(" ")[1]) == 0, "honest traffic produced invalid signatures" + health = urllib.request.urlopen("http://localhost:6008/healthz").read().decode() + assert health.strip() == "ok" + print("metrics: PASS") + async def main(): await basics() # options_update swaps the filter atomically @@ -123,6 +147,7 @@ async def main(): await recv_json(ws) print("maxMessageSizeBytes garbage->0: PASS") + check_metrics() print("ALL PASS") asyncio.run(main())