From 3533416fa623389cd2eba2c6c543077f15fd7609 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sat, 4 Apr 2026 12:22:45 -0500 Subject: [PATCH] fix cross-Io heap corruption: subscriber pg.Pool access from Evented fibers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit subscriber fibers (Evented) were calling DiskPersist methods that acquire pg.Pool connections via Threaded futex — NULL Thread.current() on Evented fibers caused heap corruption (~16min crash cycle). add MPSC host_ops queue (atomic spinlock, same pattern as BroadcastQueue): - subscriber pushes ops instead of calling dp.* directly - single background thread (std.Thread on pool_io) pops and executes - covers: cursor flush (~450/s), failure tracking, status updates - cursor loaded at spawn time (slurper passes last_seq through) Co-Authored-By: Claude Opus 4.6 --- src/host_ops.zig | 151 +++++++++++++++++++++++++++++++++++++++++++++ src/main.zig | 17 +++++ src/slurper.zig | 10 ++- src/subscriber.zig | 64 ++++++++++--------- 4 files changed, 209 insertions(+), 33 deletions(-) create mode 100644 src/host_ops.zig diff --git a/src/host_ops.zig b/src/host_ops.zig new file mode 100644 index 0000000..45d28c8 --- /dev/null +++ b/src/host_ops.zig @@ -0,0 +1,151 @@ +//! host ops queue — MPSC ring buffer for cross-Io host bookkeeping +//! +//! subscriber fibers (Evented) push host operations into this queue. +//! a single background thread (std.Thread on pool_io / Threaded) pops and +//! executes them against DiskPersist (pg.Pool). this avoids Evented fibers +//! calling pg.Pool.acquire(), which invokes a Threaded futex → NULL +//! Thread.current() → heap corruption. +//! +//! same atomic spinlock MPSC pattern as BroadcastQueue (broadcaster.zig). + +const std = @import("std"); +const event_log_mod = @import("event_log.zig"); + +const Io = std.Io; +const log = std.log.scoped(.relay); + +pub const HostOp = struct { + host_id: u64, + kind: Kind, + payload: Payload, + + pub const Kind = enum { + flush_cursor, + increment_failures, + reset_failures, + update_status, + }; + + pub const Payload = union { + seq: u64, + /// pointer to subscriber's host_shutdown atomic — set by worker on exhaustion + host_shutdown: *std.atomic.Value(bool), + none: void, + status: Status, + + pub const Status = struct { + buf: [16]u8 = .{0} ** 16, + len: u8 = 0, + + pub fn init(s: []const u8) Status { + var result: Status = .{}; + const n: u8 = @intCast(@min(s.len, 16)); + @memcpy(result.buf[0..n], s[0..n]); + result.len = n; + return result; + } + + pub fn slice(self: *const Status) []const u8 { + return self.buf[0..self.len]; + } + }; + }; +}; + +pub const HostOpsQueue = struct { + const CAPACITY = 4096; + + items: [CAPACITY]HostOp = undefined, + head: std.atomic.Value(u32) = .{ .raw = 0 }, + tail: std.atomic.Value(u32) = .{ .raw = 0 }, + push_lock: std.atomic.Value(u32) = .{ .raw = 0 }, + + persist: *event_log_mod.DiskPersist, + shutdown: *std.atomic.Value(bool), + max_consecutive_failures: u32, + + /// push an op (called from any Io context — Evented fibers or Threaded threads). + /// spins until space is available. + pub fn push(self: *HostOpsQueue, op: HostOp) void { + while (true) { + // acquire spinlock + while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { + std.atomic.spinLoopHint(); + } + + const tail = self.tail.load(.monotonic); + const next_tail = (tail + 1) % CAPACITY; + if (next_tail == self.head.load(.acquire)) { + // full — release lock, yield, retry + self.push_lock.store(0, .release); + std.atomic.spinLoopHint(); + continue; + } + + self.items[tail] = op; + self.tail.store(next_tail, .release); + self.push_lock.store(0, .release); + return; + } + } + + /// pop an op (single-consumer — worker thread only). + fn pop(self: *HostOpsQueue) ?HostOp { + const head = self.head.load(.monotonic); + if (head == self.tail.load(.acquire)) return null; + + const item = self.items[head]; + self.head.store((head + 1) % CAPACITY, .release); + return item; + } + + /// worker thread entry point. runs on pool_io (Threaded). + /// pops ops and executes them against DiskPersist. + pub fn run(self: *HostOpsQueue, pool_io: Io) void { + while (!self.shutdown.load(.acquire)) { + var drained: u32 = 0; + while (self.pop()) |op| { + self.execute(op); + drained += 1; + } + + if (drained == 0) { + // nothing to do — brief sleep via pool_io (Threaded, safe from std.Thread) + pool_io.sleep(Io.Duration.fromMilliseconds(10), .awake) catch return; + } + } + + // drain remaining ops on shutdown + while (self.pop()) |op| { + self.execute(op); + } + } + + fn execute(self: *HostOpsQueue, op: HostOp) void { + switch (op.kind) { + .flush_cursor => { + self.persist.updateHostSeq(op.host_id, op.payload.seq) catch |err| { + log.debug("host_ops: cursor flush failed for host_id={d}: {s}", .{ op.host_id, @errorName(err) }); + }; + }, + .increment_failures => { + const failures = self.persist.incrementHostFailures(op.host_id) catch 0; + if (failures >= self.max_consecutive_failures) { + log.warn("host_ops: host_id={d} exhausted after {d} failures", .{ op.host_id, failures }); + self.persist.updateHostStatus(op.host_id, "exhausted") catch {}; + op.payload.host_shutdown.store(true, .release); + } + }, + .reset_failures => { + self.persist.resetHostFailures(op.host_id) catch |err| { + log.debug("host_ops: reset failures failed for host_id={d}: {s}", .{ op.host_id, @errorName(err) }); + }; + }, + .update_status => { + self.persist.updateHostStatus(op.host_id, op.payload.status.slice()) catch |err| { + log.debug("host_ops: update status failed for host_id={d}: {s}", .{ op.host_id, @errorName(err) }); + }; + }, + } + } +}; diff --git a/src/main.zig b/src/main.zig index aaddac6..81636bb 100644 --- a/src/main.zig +++ b/src/main.zig @@ -35,6 +35,7 @@ const collection_index_mod = @import("collection_index.zig"); const backfill_mod = @import("backfill.zig"); const cleaner_mod = @import("cleaner.zig"); const resync_mod = @import("resync.zig"); +const host_ops_mod = @import("host_ops.zig"); const api = @import("api.zig"); const build_options = @import("build_options"); const malloc_trim: ?*const fn (pad: usize) callconv(.c) c_int = if (builtin.os.tag == .linux) @@ -260,6 +261,18 @@ pub fn main() !void { try resyncer.start(); defer resyncer.deinit(); + // init host ops queue — subscriber fibers (Evented) push DB ops here, + // a background thread (Threaded) executes them. avoids cross-Io pg.Pool access. + var host_ops_queue: host_ops_mod.HostOpsQueue = .{ + .persist = &dp, + .shutdown = &shutdown_flag, + .max_consecutive_failures = 15, + }; + const host_ops_thread = std.Thread.spawn(.{}, host_ops_mod.HostOpsQueue.run, .{ &host_ops_queue, pool_io }) catch |err| { + log.err("failed to start host ops thread: {s}", .{@errorName(err)}); + return err; + }; + // init slurper (multi-host crawl manager) var slurper = slurper_mod.Slurper.init( allocator, @@ -280,6 +293,7 @@ pub fn main() !void { defer slurper.deinit(); slurper.collection_index = &ci; slurper.resyncer = &resyncer; + slurper.host_ops = &host_ops_queue; // start: loads active hosts from DB, spawns subscriber threads try slurper.start(); @@ -369,6 +383,9 @@ pub fn main() !void { // must complete before dp.deinit() runs (dp is stack-owned). gc_thread.join(); + // join host ops thread — drains remaining ops before dp.deinit() + host_ops_thread.join(); + // cancel broadcaster fiber (shutdown flag already set, it will drain remaining) broadcast_future.cancel(io); diff --git a/src/slurper.zig b/src/slurper.zig index bb84e7f..9d99f39 100644 --- a/src/slurper.zig +++ b/src/slurper.zig @@ -20,6 +20,7 @@ const subscriber_mod = @import("subscriber.zig"); const collection_index_mod = @import("collection_index.zig"); const resync_mod = @import("resync.zig"); const frame_worker_mod = @import("frame_worker.zig"); +const host_ops_mod = @import("host_ops.zig"); const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); @@ -228,6 +229,7 @@ pub const Slurper = struct { persist: *event_log_mod.DiskPersist, collection_index: ?*collection_index_mod.CollectionIndex = null, resyncer: ?*resync_mod.Resyncer = null, + host_ops: ?*host_ops_mod.HostOpsQueue = null, shutdown: *std.atomic.Value(bool), options: Options, @@ -453,12 +455,12 @@ pub const Slurper = struct { self.persist.updateHostStatus(host_info.id, "active") catch {}; self.persist.resetHostFailures(host_info.id) catch {}; - try self.spawnWorker(host_info.id, hostname); + try self.spawnWorker(host_info.id, hostname, host_info.last_seq); log.info("added host {s} (id={d})", .{ hostname, host_info.id }); } /// spawn a subscriber thread for a host - fn spawnWorker(self: *Slurper, host_id: u64, hostname: []const u8) !void { + fn spawnWorker(self: *Slurper, host_id: u64, hostname: []const u8, last_seq: u64) !void { const hostname_duped = try self.allocator.dupe(u8, hostname); errdefer self.allocator.free(hostname_duped); @@ -486,6 +488,8 @@ pub const Slurper = struct { sub.resyncer = self.resyncer; if (self.frame_pool) |*fp| sub.pool = fp; sub.pool_io = self.pool_io; + sub.host_ops = self.host_ops; + if (last_seq > 0) sub.last_upstream_seq = last_seq; const future = try self.io.concurrent(runWorker, .{ self, host_id, sub }); @@ -550,7 +554,7 @@ pub const Slurper = struct { var spawned: usize = 0; for (hosts) |host| { if (self.shutdown.load(.acquire)) break; - self.spawnWorker(host.id, host.hostname) catch |err| { + self.spawnWorker(host.id, host.hostname, host.last_seq) catch |err| { log.warn("failed to spawn worker for {s}: {s}", .{ host.hostname, @errorName(err) }); }; spawned += 1; diff --git a/src/subscriber.zig b/src/subscriber.zig index 563b9ff..9153db2 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -15,6 +15,7 @@ const event_log_mod = @import("event_log.zig"); const collection_index_mod = @import("collection_index.zig"); const resync_mod = @import("resync.zig"); const frame_worker_mod = @import("frame_worker.zig"); +const host_ops_mod = @import("host_ops.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; @@ -210,6 +211,8 @@ pub const Subscriber = struct { pool: ?*frame_worker_mod.FramePool = null, /// dedicated Threaded io for frame workers — safe from plain OS threads pool_io: ?Io = null, + /// host ops queue — pushes DB ops to a background thread (avoids cross-Io pg.Pool access) + host_ops: ?*host_ops_mod.HostOpsQueue = null, shutdown: *std.atomic.Value(bool), last_upstream_seq: ?u64 = null, last_cursor_flush: i64 = 0, @@ -256,17 +259,9 @@ pub const Subscriber = struct { var backoff: u64 = 1; const max_backoff: u64 = 60; - // load cursor from DB if we have a host_id - if (self.options.host_id > 0) { - if (self.persist) |dp| { - const host_info = dp.getOrCreateHost(self.options.hostname) catch null; - if (host_info) |info| { - if (info.last_seq > 0) { - self.last_upstream_seq = info.last_seq; - log.info("host {s}: resuming from cursor {d}", .{ self.options.hostname, info.last_seq }); - } - } - } + // cursor is set at spawn time by slurper (avoids cross-Io pg.Pool access) + if (self.last_upstream_seq) |seq| { + log.info("host {s}: resuming from cursor {d}", .{ self.options.hostname, seq }); } while (!self.shouldStop()) { @@ -279,15 +274,14 @@ pub const Subscriber = struct { if (self.shouldStop()) return; - // track failures for this host + // track failures for this host (pushed to background thread via host_ops queue) if (self.options.host_id > 0) { - if (self.persist) |dp| { - const failures = dp.incrementHostFailures(self.options.host_id) catch 0; - if (failures >= max_consecutive_failures) { - log.warn("host {s}: exhausted after {d} failures, stopping", .{ self.options.hostname, failures }); - dp.updateHostStatus(self.options.host_id, "exhausted") catch {}; - return; - } + if (self.host_ops) |hq| { + hq.push(.{ + .host_id = self.options.host_id, + .kind = .increment_failures, + .payload = .{ .host_shutdown = &self.host_shutdown }, + }); } } @@ -302,14 +296,16 @@ pub const Subscriber = struct { } } - /// flush cursor position to the host table + /// flush cursor position to the host table (via host_ops queue — avoids cross-Io pg.Pool) fn flushCursor(self: *Subscriber) void { if (self.options.host_id == 0) return; const seq = self.last_upstream_seq orelse return; - if (self.persist) |dp| { - dp.updateHostSeq(self.options.host_id, seq) catch |err| { - log.debug("host {s}: cursor flush failed: {s}", .{ self.options.hostname, @errorName(err) }); - }; + if (self.host_ops) |hq| { + hq.push(.{ + .host_id = self.options.host_id, + .kind = .flush_cursor, + .payload = .{ .seq = seq }, + }); } } @@ -348,10 +344,14 @@ pub const Subscriber = struct { try client.handshake(path, .{ .headers = host_header }); log.info("host {s}: connected", .{self.options.hostname}); - // reset failures on successful connect + // reset failures on successful connect (via host_ops queue) if (self.options.host_id > 0) { - if (self.persist) |dp| { - dp.resetHostFailures(self.options.host_id) catch {}; + if (self.host_ops) |hq| { + hq.push(.{ + .host_id = self.options.host_id, + .kind = .reset_failures, + .payload = .{ .none = {} }, + }); } } @@ -400,9 +400,13 @@ const FrameHandler = struct { log.warn("host {s}: error frame: {s}: {s}", .{ sub.options.hostname, err_name, err_msg }); if (std.mem.eql(u8, err_name, "FutureCursor")) { // our cursor is ahead of the PDS — set host to idle, stop this subscriber only - if (sub.persist) |dp| { - if (sub.options.host_id > 0) { - dp.updateHostStatus(sub.options.host_id, "idle") catch {}; + if (sub.options.host_id > 0) { + if (sub.host_ops) |hq| { + hq.push(.{ + .host_id = sub.options.host_id, + .kind = .update_status, + .payload = .{ .status = host_ops_mod.HostOp.Payload.Status.init("idle") }, + }); } } sub.host_shutdown.store(true, .release); -- 2.51.2