diff --git a/build.zig.zon b/build.zig.zon
index eee92fb..3fce98c 100644
--- a/build.zig.zon
+++ b/build.zig.zon
@@ -5,12 +5,12 @@
     .minimum_zig_version = "0.16.0",
     .dependencies = .{
         .zat = .{
-            .url = "https://tangled.org/zat.dev/zat/archive/v0.3.0-alpha.11.tar.gz",
-            .hash = "zat-0.3.0-alpha.11-5PuC7nVhBQCNUnJEi_YUqQK6V8bbZacA-QN54nUunu4K",
+            .url = "https://tangled.org/zat.dev/zat/archive/v0.3.0-alpha.15.tar.gz",
+            .hash = "zat-0.3.0-alpha.15-5PuC7nVhBQCJEzz9LuzSbtLb68Wd0x_yjDgTP3EqV8dH",
         },
         .websocket = .{
-            .url = "https://github.com/zzstoatzz/websocket.zig/archive/104608b.tar.gz",
-            .hash = "websocket-0.1.0-ZPISdXjUAwC3rN7rT5NMG8HQJRug1NOboVWeX09SvSGv",
+            .url = "https://github.com/zzstoatzz/websocket.zig/archive/ac3df25.tar.gz",
+            .hash = "websocket-0.1.0-ZPISdUvvAwDQN3W3AYDxmzMj5ipuTnB3vpQinQPF9LqI",
         },
         .pg = .{
             .url = "git+https://github.com/zzstoatzz/pg.zig?ref=dev#5ce2355b1d851075523709c7d3068dcdb0224322",
diff --git a/src/api/admin.zig b/src/api/admin.zig
index 6d6e8e6..5ae9446 100644
--- a/src/api/admin.zig
+++ b/src/api/admin.zig
@@ -74,14 +74,28 @@ pub fn handleBan(conn: *h.Conn, body: []const u8, headers: *const websocket.Hand
         log.debug("collection removeAll after ban failed: {s}", .{@errorName(err)});
     };
 
-    // emit #account event so downstream consumers see the takedown
+    // emit #account event — same ordered publication path as workers:
+    // persist_order spinlock → dp.persist → resequence → queue.push.
+    // one publication path for all relay-sequenced events.
     if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| {
+        while (ctx.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+            std.atomic.spinLoopHint();
+        }
+
         if (ctx.persist.persist(.account, uid, frame_bytes)) |relay_seq| {
             ctx.bc.stats.relay_seq.store(relay_seq, .release);
             const broadcast_data = broadcaster.resequenceFrame(ctx.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes;
-            ctx.bc.broadcast(relay_seq, broadcast_data);
+            const owned = ctx.persist.allocator.dupe(u8, broadcast_data) catch {
+                ctx.bc.persist_order.store(0, .release);
+                log.warn("admin: failed to alloc broadcast data for {s}", .{did});
+                h.respondJson(conn, .ok, "{\"success\":true}");
+                return;
+            };
+            ctx.bc.broadcast_queue.push(relay_seq, owned);
+            ctx.bc.persist_order.store(0, .release);
             log.info("admin: emitted #account takedown event for {s} (seq={d})", .{ did, relay_seq });
         } else |err| {
+            ctx.bc.persist_order.store(0, .release);
             log.warn("admin: failed to persist #account takedown event: {s}", .{@errorName(err)});
         }
     }
diff --git a/src/api/router.zig b/src/api/router.zig
index 114b722..37db586 100644
--- a/src/api/router.zig
+++ b/src/api/router.zig
@@ -4,6 +4,7 @@
 //! top-level request router that delegates to xrpc and admin handler modules.
 
 const std = @import("std");
+const Io = std.Io;
 const websocket = @import("websocket");
 const broadcaster = @import("../broadcaster.zig");
 const validator_mod = @import("../validator.zig");
@@ -28,6 +29,7 @@ pub const HttpContext = struct {
     resyncer: *resync_mod.Resyncer,
     bc: *broadcaster.Broadcaster,
     validator: *validator_mod.Validator,
+    pool_io: Io,
 };
 
 /// top-level HTTP request router — installed as bc.http_fallback
diff --git a/src/backfill.zig b/src/backfill.zig
index 75c9ae2..84506b6 100644
--- a/src/backfill.zig
+++ b/src/backfill.zig
@@ -59,6 +59,10 @@ pub const Backfiller = struct {
         errdefer self.running.store(false, .release);
 
         self.source = try self.allocator.dupe(u8, source);
+        errdefer {
+            self.allocator.free(self.source);
+            self.source = "";
+        }
         self.future = try self.io.concurrent(run, .{self});
     }
 
diff --git a/src/broadcaster.zig b/src/broadcaster.zig
index 96583f2..82b68fb 100644
--- a/src/broadcaster.zig
+++ b/src/broadcaster.zig
@@ -231,6 +231,68 @@ pub fn resequenceFrame(allocator: Allocator, data: []const u8, relay_seq: u64) ?
     return result;
 }
 
+// --- broadcast queue (worker → fiber handoff) ---
+
+/// item produced by frame workers, consumed by the broadcaster fiber.
+/// data is heap-allocated by the worker; the broadcaster fiber frees it after broadcast.
+pub const BroadcastItem = struct {
+    seq: u64,
+    data: []const u8,
+};
+
+/// MPSC ring buffer: multiple frame worker threads push, one broadcaster fiber pops.
+/// uses an atomic spinlock for push (no Io.Mutex — works from any execution context).
+/// pop is single-consumer (broadcaster fiber only), no locking needed.
+pub const BroadcastQueue = struct {
+    const CAPACITY = 8192;
+
+    items: [CAPACITY]BroadcastItem = undefined,
+    head: std.atomic.Value(u32) = .{ .raw = 0 },
+    tail: std.atomic.Value(u32) = .{ .raw = 0 },
+    push_lock: std.atomic.Value(u32) = .{ .raw = 0 },
+    allocator: Allocator,
+
+    pub fn init(allocator: Allocator) BroadcastQueue {
+        return .{ .allocator = allocator };
+    }
+
+    /// push an item (called by worker threads). spins until space is available.
+    /// matches Indigo semantics: every persisted event reaches live broadcast.
+    /// the broadcaster fiber drains at memory speed (no I/O), so spin is brief.
+    pub fn push(self: *BroadcastQueue, seq: u64, data: []const u8) 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] = .{ .seq = seq, .data = data };
+            self.tail.store(next_tail, .release);
+            self.push_lock.store(0, .release);
+            return;
+        }
+    }
+
+    /// pop an item (called by broadcaster fiber only). returns null if empty.
+    pub fn pop(self: *BroadcastQueue) ?BroadcastItem {
+        const head = self.head.load(.monotonic);
+        if (head == self.tail.load(.acquire)) return null; // empty
+
+        const item = self.items[head];
+        self.head.store((head + 1) % CAPACITY, .release);
+        return item;
+    }
+};
+
 // --- consumer ---
 
 const ping_interval_ns: u64 = 30 * std.time.ns_per_s;
@@ -370,7 +432,12 @@ pub const Broadcaster = struct {
     allocator: Allocator,
     consumers: std.ArrayListUnmanaged(*Consumer) = .empty,
     consumers_mutex: Io.Mutex = Io.Mutex.init,
-    broadcast_order: Io.Mutex = Io.Mutex.init,
+    /// ordering spinlock for persist → queue push atomicity.
+    /// atomic spinlock (not Io.Mutex) so both Threaded workers and Evented admin
+    /// can participate in the same ordered publication path.
+    persist_order: std.atomic.Value(u32) = .{ .raw = 0 },
+    /// worker → fiber handoff: frame workers push here, broadcaster fiber pops.
+    broadcast_queue: BroadcastQueue,
     history: FrameHistory,
     persist: ?*event_log_mod.DiskPersist = null,
     stats: Stats = .{},
@@ -378,14 +445,17 @@ pub const Broadcaster = struct {
     http_fallback: ?HttpFallbackFn = null,
     http_fallback_ctx: ?*anyopaque = null,
     io: Io,
+    shutdown: *std.atomic.Value(bool),
 
-    pub fn init(allocator: Allocator, io: Io) Broadcaster {
+    pub fn init(allocator: Allocator, io: Io, shutdown: *std.atomic.Value(bool)) Broadcaster {
         return .{
             .allocator = allocator,
+            .broadcast_queue = BroadcastQueue.init(allocator),
             .history = FrameHistory.init(allocator, io),
             .stats = .{ .start_time = timestamp(io) },
             .error_frame = buildErrorFrame(allocator),
             .io = io,
+            .shutdown = shutdown,
         };
     }
 
@@ -563,6 +633,31 @@ pub const Broadcaster = struct {
         defer self.consumers_mutex.unlock(self.io);
         return self.consumers.items.len;
     }
+
+    /// broadcast loop — runs as an Evented fiber, drains the broadcast queue
+    /// and calls broadcast() for each item. this is the only path that touches
+    /// consumers_mutex / consumer.mutex / consumer.cond.
+    pub fn runBroadcastLoop(self: *Broadcaster) void {
+        log.info("broadcaster fiber started", .{});
+        while (!self.shutdown.load(.acquire)) {
+            var drained: usize = 0;
+            while (self.broadcast_queue.pop()) |item| {
+                self.broadcast(item.seq, item.data);
+                self.allocator.free(@constCast(item.data));
+                drained += 1;
+            }
+            if (drained == 0) {
+                // no items — yield briefly so other fibers run
+                self.io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return;
+            }
+        }
+        // drain remaining items on shutdown
+        while (self.broadcast_queue.pop()) |item| {
+            self.broadcast(item.seq, item.data);
+            self.allocator.free(@constCast(item.data));
+        }
+        log.info("broadcaster fiber stopped", .{});
+    }
 };
 
 // --- websocket handler ---
@@ -1068,8 +1163,10 @@ fn timestamp(io: Io) i64 {
 
 // --- tests ---
 
+var test_shutdown: std.atomic.Value(bool) = .{ .raw = false };
+
 test "broadcaster add and remove consumer" {
-    var b = Broadcaster.init(std.testing.allocator, std.testing.io);
+    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
     defer b.deinit();
 
     try std.testing.expectEqual(@as(u64, 0), b.stats.seq.load(.acquire));
@@ -1077,7 +1174,7 @@ test "broadcaster add and remove consumer" {
 }
 
 test "broadcast updates stats and history" {
-    var b = Broadcaster.init(std.testing.allocator, std.testing.io);
+    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
     defer b.deinit();
 
     b.broadcast(1, "frame1");
@@ -1092,7 +1189,7 @@ test "broadcast updates stats and history" {
 }
 
 test "frame history supports cursor replay" {
-    var b = Broadcaster.init(std.testing.allocator, std.testing.io);
+    var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
     defer b.deinit();
 
     for (1..6) |i| {
@@ -1304,16 +1401,12 @@ test "encodeInfoMessage produces valid #info message frame CBOR" {
     try std.testing.expect(p.getString("error") == null);
 }
 
-test "concurrent broadcast through ordering mutex produces monotonic sequences" {
-    // regression test: without broadcast_order serialization, concurrent
-    // subscriber threads can interleave persist (seq assignment) and broadcast,
-    // delivering frames out of order to consumers.
-    //
-    // this simulates the subscriber pattern: N threads each acquire the
-    // ordering lock, assign a seq (atomic increment, like persist), and
-    // broadcast. the ring buffer history must be strictly monotonic.
+test "concurrent broadcast queue + drain produces monotonic sequences" {
+    // regression test: multiple worker threads push to the broadcast queue
+    // under persist_order lock, then the broadcaster fiber drains and broadcasts.
+    // the ring buffer history must be strictly monotonic.
 
-    var bc = Broadcaster.init(std.testing.allocator, std.testing.io);
+    var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown);
     defer bc.deinit();
 
     const num_threads = 8;
@@ -1324,11 +1417,16 @@ test "concurrent broadcast through ordering mutex produces monotonic sequences"
     const Worker = struct {
         fn run(bc_ptr: *Broadcaster, counter: *std.atomic.Value(u64)) void {
             for (0..frames_per_thread) |_| {
-                bc_ptr.broadcast_order.lockUncancelable(bc_ptr.io);
-                defer bc_ptr.broadcast_order.unlock(bc_ptr.io);
-
+                while (bc_ptr.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+                    std.atomic.spinLoopHint();
+                }
                 const seq = counter.fetchAdd(1, .monotonic) + 1;
-                bc_ptr.broadcast(seq, "x");
+                const data = std.testing.allocator.dupe(u8, "x") catch {
+                    bc_ptr.persist_order.store(0, .release);
+                    continue;
+                };
+                bc_ptr.broadcast_queue.push(seq, data);
+                bc_ptr.persist_order.store(0, .release);
             }
         }
     };
@@ -1339,6 +1437,12 @@ test "concurrent broadcast through ordering mutex produces monotonic sequences"
     }
     for (&threads) |*t| t.join();
 
+    // drain the queue (simulating the broadcaster fiber)
+    while (bc.broadcast_queue.pop()) |item| {
+        bc.broadcast(item.seq, item.data);
+        std.testing.allocator.free(@constCast(item.data));
+    }
+
     // verify: ring buffer history has strictly monotonic sequences
     const frames = try bc.history.framesSince(std.testing.allocator, 0);
     defer {
diff --git a/src/frame_worker.zig b/src/frame_worker.zig
index 54d4d25..87cdd9e 100644
--- a/src/frame_worker.zig
+++ b/src/frame_worker.zig
@@ -273,19 +273,29 @@ pub fn processFrame(work: *FrameWork) void {
     else
         .identity;
 
-    // persist and broadcast under ordering lock
+    // persist under ordering lock, then push to broadcast queue.
+    // the broadcaster fiber (Evented) drains the queue and does fan-out —
+    // worker threads never touch consumer state directly.
     if (work.persist) |dp| {
         const relay_seq = blk: {
-            work.bc.broadcast_order.lockUncancelable(work.io);
-            defer work.bc.broadcast_order.unlock(work.io);
+            while (work.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+                std.atomic.spinLoopHint();
+            }
 
             const seq = dp.persist(kind, uid, data) catch |err| {
+                work.bc.persist_order.store(0, .release);
                 log.warn("persist failed: {s}", .{@errorName(err)});
                 return;
             };
             work.bc.stats.relay_seq.store(seq, .release);
             const broadcast_data = broadcaster.resequenceFrame(alloc, data, seq) orelse data;
-            work.bc.broadcast(seq, broadcast_data);
+            // dupe for the broadcast queue — arena will free broadcast_data
+            const owned = work.allocator.dupe(u8, broadcast_data) catch {
+                work.bc.persist_order.store(0, .release);
+                return;
+            };
+            work.bc.broadcast_queue.push(seq, owned);
+            work.bc.persist_order.store(0, .release);
             break :blk seq;
         };
         _ = relay_seq;
@@ -310,7 +320,8 @@ pub fn processFrame(work: *FrameWork) void {
         }
     } else {
         const upstream_seq = payload.getUint("seq") orelse 0;
-        work.bc.broadcast(upstream_seq, data);
+        const owned = work.allocator.dupe(u8, data) catch return;
+        work.bc.broadcast_queue.push(upstream_seq, owned);
     }
 }
 
diff --git a/src/main.zig b/src/main.zig
index 6cf0069..9a7286f 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -47,12 +47,10 @@ const log = std.log.scoped(.relay);
 pub const default_stack_size = 8 * 1024 * 1024;
 
 // -- Io backend selection --
-// Evented backends (Io.Uring, Io.Dispatch, Io.Kqueue) use fiber-local state
-// for futex wait/wake. Plain std.Thread workers (frame pool) calling
-// Io.Mutex with an Evented io segfault because they lack that state.
-// Use Threaded until frame workers are migrated to io.concurrent or
-// cross-boundary mutexes get a dedicated Threaded sync_io.
-const Backend = Io.Threaded;
+// Evented (fibers): network orchestration, subscriber connections, WS server, broadcasting.
+// Worker threads use a dedicated pool_io (Threaded) for their sync — they never touch Evented io.
+// The broadcast queue bridges workers → broadcaster fiber (atomics only, no Io dependency).
+const Backend = Io.Evented;
 
 var backend: Backend = undefined;
 var debug_threaded_io: Io.Threaded = undefined;
@@ -160,17 +158,33 @@ pub fn main() !void {
     };
     const allocator = if (build_options.use_gpa) gpa.allocator() else std.heap.c_allocator;
 
+    // shared options for both debug and primary runtimes
+    const io_opts: Io.Threaded.InitOptions = .{
+        .stack_size = default_stack_size, // 8MB (default is 16MB)
+        .concurrent_limit = Io.Limit.limited(4096), // safety rail (steady-state ~2,800 hosts)
+    };
+
     // init debug io (for std.debug.print thread safety)
-    debug_threaded_io = Io.Threaded.init(allocator, .{});
+    debug_threaded_io = Io.Threaded.init(allocator, io_opts);
 
-    // init primary runtime
+    // init primary runtime (Evented: fibers for network orchestration)
     if (Backend == Io.Threaded) {
-        backend = Io.Threaded.init(allocator, .{});
+        backend = Io.Threaded.init(allocator, io_opts);
     } else {
         try Backend.init(&backend, allocator, .{});
     }
     const io = backend.io();
 
+    // dedicated Threaded runtime for the frame worker pool.
+    // worker threads are plain std.Thread — they cannot use Evented io
+    // (Evented futex calls ev.yield() which requires fiber context).
+    // this io is used for: persist ordering mutex, timestamps, validator cache,
+    // DID resolution HTTP, and thread pool internal sync.
+    var pool_io_backend = Io.Threaded.init(allocator, .{
+        .stack_size = default_stack_size,
+    });
+    const pool_io = pool_io_backend.io();
+
     log.info("io backend: {s}", .{if (Backend == Io.Threaded) "Threaded" else "Evented"});
 
     // parse config from env
@@ -188,16 +202,17 @@ pub fn main() !void {
     installSignalHandlers();
 
     // init components — pass io to network-facing modules
-    var bc = broadcaster.Broadcaster.init(allocator, io);
+    var bc = broadcaster.Broadcaster.init(allocator, io, &shutdown_flag);
     defer bc.deinit();
 
-    var val = validator_mod.Validator.init(allocator, &bc.stats, io);
+    // validator uses pool_io — its cache LRU and host resolvers are called from worker threads
+    var val = validator_mod.Validator.init(allocator, &bc.stats, pool_io);
     defer val.deinit();
     try val.start();
 
     // init disk persistence (indigo-compatible diskpersist format + Postgres index)
     const database_url = getenv("DATABASE_URL") orelse "postgres://relay:relay@localhost:5432/relay";
-    var dp = event_log_mod.DiskPersist.init(allocator, data_dir, database_url, db_pool_size, io) catch |err| {
+    var dp = event_log_mod.DiskPersist.init(allocator, data_dir, database_url, db_pool_size, pool_io) catch |err| {
         log.err("failed to init disk persist at {s}: {s}", .{ data_dir, @errorName(err) });
         return err;
     };
@@ -249,6 +264,7 @@ pub fn main() !void {
             .frame_queue_capacity = frame_queue_capacity,
         },
         io,
+        pool_io,
     );
     defer slurper.deinit();
     slurper.collection_index = &ci;
@@ -257,8 +273,15 @@ pub fn main() !void {
     // start: loads active hosts from DB, spawns subscriber threads
     try slurper.start();
 
+    // start broadcaster fiber — drains broadcast queue, owns all consumer state.
+    // this is the Evented-side sequencer: frame workers push results to the queue,
+    // this fiber does the actual fan-out to downstream consumers.
+    var broadcast_future = try io.concurrent(broadcaster.Broadcaster.runBroadcastLoop, .{&bc});
+    defer _ = broadcast_future.cancel(io);
+
     // start GC loop (runs as background task — does disk I/O + malloc_trim)
     var gc_future = try io.concurrent(gcLoop, .{ &dp, io });
+    defer _ = gc_future.cancel(io);
 
     // wire HTTP fallback into broadcaster (all API endpoints served on WS port)
     var http_context = api.HttpContext{
@@ -271,6 +294,7 @@ pub fn main() !void {
         .resyncer = &resyncer,
         .bc = &bc,
         .validator = &val,
+        .pool_io = pool_io,
     };
     bc.http_fallback = api.handleHttpRequest;
     bc.http_fallback_ctx = @ptrCast(&http_context);
@@ -291,13 +315,14 @@ pub fn main() !void {
         .slurper = &slurper,
     };
     var metrics_future = try io.concurrent(MetricsServer.run, .{&metrics_srv});
+    defer _ = metrics_future.cancel(io);
 
     // start downstream WebSocket server (also serves HTTP API via httpFallback)
     log.info("relay listening on :{d} (ws+http), :{d} (metrics)", .{ port, metrics_port });
     log.info("seed host: {s}", .{upstream});
     log.info("data dir: {s} (retention: {d}h, max: {d} GB)", .{ data_dir, retention_hours, max_events_gb });
 
-    var server = try websocket.Server(broadcaster.Handler).init(allocator, .{
+    var server = try websocket.Server(broadcaster.Handler).init(allocator, io, .{
         .port = port,
         .address = "0.0.0.0",
         .max_conn = 4096,
@@ -305,7 +330,14 @@ pub fn main() !void {
     });
     defer server.deinit();
 
-    const server_thread = try server.listenInNewThread(&bc);
+    // Io-native accept loop: fiber-based under Evented, thread-based under Threaded
+    const ws_address = Io.net.Ip4Address.unspecified(port);
+    var ws_listener = (Io.net.IpAddress{ .ip4 = ws_address }).listen(io, .{ .reuse_address = true }) catch |err| {
+        log.err("websocket server failed to listen on :{d}: {s}", .{ port, @errorName(err) });
+        return err;
+    };
+    var server_future = try io.concurrent(runWsServer, .{ &server, &ws_listener, &bc });
+    defer _ = server_future.cancel(io);
 
     // wait for shutdown signal
     while (!shutdown_flag.load(.acquire)) {
@@ -314,13 +346,16 @@ pub fn main() !void {
 
     log.info("shutdown signal received, stopping...", .{});
 
-    // stop WebSocket server (closes all downstream connections)
-    server.stop();
-    server_thread.join();
+    // stop WebSocket server: close listener to unblock accept, then cancel task
+    ws_listener.deinit(io);
+    server_future.cancel(io);
 
     // cancel GC task
     gc_future.cancel(io);
 
+    // cancel broadcaster fiber (shutdown flag already set, it will drain remaining)
+    broadcast_future.cancel(io);
+
     // close metrics listener to unblock accept(), then cancel task
     metrics_srv.server.deinit(io);
     metrics_future.cancel(io);
@@ -330,6 +365,11 @@ pub fn main() !void {
 
 const builtin = @import("builtin");
 
+/// concrete wrapper for runIo — io.concurrent needs ArgsTuple, which can't handle anytype
+fn runWsServer(server: *websocket.Server(broadcaster.Handler), listener: *Io.net.Server, bc: *broadcaster.Broadcaster) void {
+    server.runIo(listener, bc);
+}
+
 fn gcLoop(dp: *event_log_mod.DiskPersist, io: Io) void {
     const gc_interval: u64 = 10 * 60; // 10 minutes in seconds
     while (!shutdown_flag.load(.acquire)) {
diff --git a/src/slurper.zig b/src/slurper.zig
index 00a8f6e..3cc301d 100644
--- a/src/slurper.zig
+++ b/src/slurper.zig
@@ -247,6 +247,8 @@ pub const Slurper = struct {
     crawl_future: ?Io.Future(void) = null,
 
     io: Io,
+    /// dedicated Threaded io for the frame worker pool — safe from plain OS threads
+    pool_io: Io,
 
     pub fn init(
         allocator: Allocator,
@@ -256,6 +258,7 @@ pub const Slurper = struct {
         shutdown: *std.atomic.Value(bool),
         options: Options,
         io: Io,
+        pool_io: Io,
     ) Slurper {
         return .{
             .allocator = allocator,
@@ -265,6 +268,7 @@ pub const Slurper = struct {
             .shutdown = shutdown,
             .options = options,
             .io = io,
+            .pool_io = pool_io,
         };
     }
 
@@ -277,12 +281,13 @@ pub const Slurper = struct {
         self.ca_bundle = bundle;
         log.info("loaded shared CA bundle", .{});
 
-        // create frame processing pool — worker threads handle heavy decode/validate/persist
+        // create frame processing pool — worker threads use pool_io (Threaded),
+        // safe from plain OS threads even when the app uses Evented io
         self.frame_pool = try frame_worker_mod.FramePool.init(self.allocator, .{
             .num_workers = self.options.frame_workers,
             .queue_capacity = self.options.frame_queue_capacity,
             .stack_size = @import("main.zig").default_stack_size,
-        }, self.io);
+        }, self.pool_io);
         log.info("frame pool started: {d} workers, queue capacity {d}", .{ self.options.frame_workers, self.options.frame_queue_capacity });
 
         // spawn worker startup in background so HTTP server + probes come up immediately.
@@ -476,6 +481,7 @@ pub const Slurper = struct {
         sub.collection_index = self.collection_index;
         sub.resyncer = self.resyncer;
         if (self.frame_pool) |*fp| sub.pool = fp;
+        sub.pool_io = self.pool_io;
 
         const future = try self.io.concurrent(runWorker, .{ self, host_id, sub });
 
diff --git a/src/subscriber.zig b/src/subscriber.zig
index a5d9549..55e7612 100644
--- a/src/subscriber.zig
+++ b/src/subscriber.zig
@@ -208,6 +208,8 @@ pub const Subscriber = struct {
     collection_index: ?*collection_index_mod.CollectionIndex = null,
     resyncer: ?*resync_mod.Resyncer = null,
     pool: ?*frame_worker_mod.FramePool = null,
+    /// dedicated Threaded io for frame workers — safe from plain OS threads
+    pool_io: ?Io = null,
     shutdown: *std.atomic.Value(bool),
     last_upstream_seq: ?u64 = null,
     last_cursor_flush: i64 = 0,
@@ -351,45 +353,13 @@ pub const Subscriber = struct {
             .subscriber = self,
         };
 
-        // spawn keepalive ping task (Go relay: 30s ticker goroutine in consumer.go)
-        // prevents intermediate proxies (e.g. Cloudflare) from killing idle connections
-        var ping_future = self.io.concurrent(pingLoop, .{ &client, self }) catch |err| {
-            log.warn("host {s}: failed to spawn ping task: {s}", .{ self.options.hostname, @errorName(err) });
-            return err;
-        };
-
-        defer ping_future.cancel(self.io);
-        try client.readLoop(&handler);
-    }
-
-    /// periodic WebSocket ping to keep the connection alive.
-    /// matches indigo relay: 30s interval, close after 4 consecutive failures.
-    fn pingLoop(client: *websocket.Client, self: *Subscriber) void {
-        var fail_count: u32 = 0;
-        while (!self.shouldStop()) {
-            // sleep in 1s increments so we can check shutdown.
-            // return on any sleep error — critically, error.Canceled from
-            // ping_future.cancel() must not be swallowed, otherwise deinit
-            // frees the client while we're still running.
-            var elapsed: u32 = 0;
-            while (elapsed < ping_interval_sec and !self.shouldStop()) {
-                self.io.sleep(Io.Duration.fromSeconds(1), .awake) catch return;
-                elapsed += 1;
-            }
-            if (self.shouldStop() or client.isClosed()) return;
-
-            client.writeFrame(.ping, &.{}) catch {
-                fail_count += 1;
-                log.warn("host {s}: ping failed ({d}/{d})", .{ self.options.hostname, fail_count, max_ping_failures });
-                if (fail_count >= max_ping_failures) {
-                    log.err("host {s}: too many ping failures, closing connection", .{self.options.hostname});
-                    client.close(.{}) catch {};
-                    return;
-                }
-                continue;
-            };
-            fail_count = 0;
-        }
+        // heartbeat read loop: merged keepalive + frame reading in a single task.
+        // SO_RCVTIMEO fires after interval_ms, triggering a ping. closes after
+        // max_failures consecutive idle intervals with no frames received.
+        try client.readLoopWithHeartbeat(&handler, .{
+            .interval_ms = ping_interval_sec * 1000,
+            .max_failures = max_ping_failures,
+        });
     }
 };
 
@@ -503,7 +473,7 @@ const FrameHandler = struct {
                 .host_id = sub.options.host_id,
                 .hostname = sub.options.hostname,
                 .allocator = sub.allocator,
-                .io = sub.io,
+                .io = sub.pool_io orelse sub.io, // pool_io (Threaded) for worker-safe ops
                 .bc = sub.bc,
                 .validator = sub.validator,
                 .persist = sub.persist,
@@ -728,22 +698,27 @@ const FrameHandler = struct {
         else // is_identity (unknown types already filtered above)
             .identity;
 
-        // persist and get relay-assigned seq, broadcast raw bytes.
-        // ordering mutex ensures frames are broadcast in seq order —
-        // without it, concurrent subscriber threads can interleave
-        // persist (seq assignment) and broadcast, delivering out-of-order.
+        // persist and push to broadcast queue (broadcaster fiber handles fan-out).
+        // ordering spinlock ensures frames are persisted + enqueued in seq order.
         if (sub.persist) |dp| {
             const relay_seq = blk: {
-                sub.bc.broadcast_order.lockUncancelable(sub.io);
-                defer sub.bc.broadcast_order.unlock(sub.io);
+                while (sub.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+                    std.atomic.spinLoopHint();
+                }
 
                 const seq = dp.persist(kind, uid, data) catch |err| {
+                    sub.bc.persist_order.store(0, .release);
                     log.warn("persist failed: {s}", .{@errorName(err)});
                     return;
                 };
                 sub.bc.stats.relay_seq.store(seq, .release);
                 const broadcast_data = broadcaster.resequenceFrame(alloc, data, seq) orelse data;
-                sub.bc.broadcast(seq, broadcast_data);
+                const owned = sub.allocator.dupe(u8, broadcast_data) catch {
+                    sub.bc.persist_order.store(0, .release);
+                    return;
+                };
+                sub.bc.broadcast_queue.push(seq, owned);
+                sub.bc.persist_order.store(0, .release);
                 break :blk seq;
             };
             _ = relay_seq;
@@ -767,7 +742,9 @@ const FrameHandler = struct {
                 }
             }
         } else {
-            sub.bc.broadcast(upstream_seq orelse 0, data);
+            const seq = upstream_seq orelse 0;
+            const owned = sub.allocator.dupe(u8, data) catch return;
+            sub.bc.broadcast_queue.push(seq, owned);
         }
     }