diff --git a/src/api/admin.zig b/src/api/admin.zig
index cffaedf..9bc786a 100644
--- a/src/api/admin.zig
+++ b/src/api/admin.zig
@@ -8,7 +8,6 @@ const Io = std.Io;
 const h = @import("http.zig");
 const router = @import("router.zig");
 const websocket = @import("websocket");
-const broadcaster = @import("../broadcaster.zig");
 const event_log_mod = @import("../event_log.zig");
 const backfill_mod = @import("../backfill.zig");
 const cleaner_mod = @import("../cleaner.zig");
@@ -59,46 +58,37 @@ pub fn handleBan(conn: *h.Conn, body: []const u8, headers: *const websocket.Hand
     defer parsed.deinit();
     const did = parsed.value.did;
 
-    // resolve DID → UID and take down
-    const uid = ctx.persist.uidForDid(did) catch {
+    // resolve DID → UID via Evented pool (skips DID cache which uses pool_io mutex)
+    const uid = ctx.persist.uidForDidEv(did) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to resolve DID\"}");
         return;
     };
-    ctx.persist.takeDownUser(uid) catch {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"takedown failed\"}");
-        return;
-    };
 
     // remove from collection index so banned accounts don't appear in listReposByCollection
     ctx.collection_index.removeAll(did) catch |err| {
         log.debug("collection removeAll after ban failed: {s}", .{@errorName(err)});
     };
 
-    // emit #account event — persist under narrow ordering lock, resequence + enqueue outside.
-    if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| {
-        while (ctx.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
-            std.atomic.spinLoopHint();
-        }
+    // build CBOR #account frame and route takedown + persist + broadcast
+    // through host_ops queue (pool_io thread) — fire and forget.
+    const host_ops_mod = @import("../host_ops.zig");
+    var td: host_ops_mod.HostOp.Payload.Takedown = .{ .uid = uid };
 
-        if (ctx.persist.persist(.account, uid, frame_bytes)) |relay_seq| {
-            ctx.bc.stats.relay_seq.store(relay_seq, .release);
-            ctx.bc.persist_order.store(0, .release);
-
-            const broadcast_data = broadcaster.resequenceFrame(ctx.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes;
-            const owned = ctx.persist.allocator.dupe(u8, broadcast_data) catch {
-                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.stats);
-            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)});
+    if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| {
+        defer ctx.persist.allocator.free(frame_bytes);
+        if (frame_bytes.len <= td.frame_buf.len) {
+            @memcpy(td.frame_buf[0..frame_bytes.len], frame_bytes);
+            td.frame_len = @intCast(frame_bytes.len);
         }
     }
 
-    log.info("admin: banned {s} (uid={d})", .{ did, uid });
+    ctx.host_ops.push(.{
+        .host_id = 0, // not host-specific
+        .kind = .takedown_user,
+        .payload = .{ .takedown = td },
+    });
+
+    log.info("admin: banned {s} (uid={d}), takedown enqueued", .{ did, uid });
     h.respondJson(conn, .ok, "{\"success\":true}");
 }
 
@@ -106,7 +96,7 @@ pub fn handleAdminListHosts(conn: *h.Conn, headers: *const websocket.Handshake.K
     if (!checkAdmin(conn, headers)) return;
 
     const persist = ctx.persist;
-    const hosts = persist.listAllHosts(persist.allocator) catch {
+    const hosts = persist.listAllHostsEv(persist.allocator) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
     };
@@ -159,12 +149,12 @@ pub fn handleAdminBlockHost(conn: *h.Conn, body: []const u8, headers: *const web
     };
     defer parsed.deinit();
 
-    const host_info = persist.getOrCreateHost(parsed.value.hostname) catch {
+    const host_info = persist.getOrCreateHostEv(parsed.value.hostname) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}");
         return;
     };
 
-    persist.updateHostStatus(host_info.id, "blocked") catch {
+    persist.updateHostStatusEv(host_info.id, "blocked") catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}");
         return;
     };
@@ -182,16 +172,16 @@ pub fn handleAdminUnblockHost(conn: *h.Conn, body: []const u8, headers: *const w
     };
     defer parsed.deinit();
 
-    const host_info = persist.getOrCreateHost(parsed.value.hostname) catch {
+    const host_info = persist.getOrCreateHostEv(parsed.value.hostname) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}");
         return;
     };
 
-    persist.updateHostStatus(host_info.id, "active") catch {
+    persist.updateHostStatusEv(host_info.id, "active") catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}");
         return;
     };
-    persist.resetHostFailures(host_info.id) catch {};
+    persist.resetHostFailuresEv(host_info.id) catch {};
 
     log.info("admin: unblocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id });
     h.respondJson(conn, .ok, "{\"success\":true}");
@@ -214,7 +204,7 @@ pub fn handleAdminChangeLimits(conn: *h.Conn, body: []const u8, headers: *const
     };
     defer parsed.deinit();
 
-    const host_id = ctx.persist.getHostIdForHostname(parsed.value.host) catch {
+    const host_id = ctx.persist.getHostIdForHostnameEv(parsed.value.host) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"database error\"}");
         return;
     } orelse {
@@ -222,13 +212,13 @@ pub fn handleAdminChangeLimits(conn: *h.Conn, body: []const u8, headers: *const
         return;
     };
 
-    ctx.persist.setHostAccountLimit(host_id, parsed.value.account_limit) catch {
+    ctx.persist.setHostAccountLimitEv(host_id, parsed.value.account_limit) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to update limit\"}");
         return;
     };
 
     // update running subscriber's rate limits immediately
-    const effective = if (parsed.value.account_limit) |l| l else ctx.persist.getHostAccountCount(host_id);
+    const effective = if (parsed.value.account_limit) |l| l else ctx.persist.getHostAccountCountEv(host_id);
     ctx.slurper.updateHostLimits(host_id, effective);
 
     if (parsed.value.account_limit) |limit| {
diff --git a/src/api/router.zig b/src/api/router.zig
index b15015c..9b02807 100644
--- a/src/api/router.zig
+++ b/src/api/router.zig
@@ -14,6 +14,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 h = @import("http.zig");
 const xrpc = @import("xrpc.zig");
 const admin = @import("admin.zig");
@@ -29,6 +30,7 @@ pub const HttpContext = struct {
     resyncer: *resync_mod.Resyncer,
     bc: *broadcaster.Broadcaster,
     validator: *validator_mod.Validator,
+    host_ops: *host_ops_mod.HostOpsQueue,
     pool_io: Io,
 };
 
diff --git a/src/api/xrpc.zig b/src/api/xrpc.zig
index bc61a4e..505e5d2 100644
--- a/src/api/xrpc.zig
+++ b/src/api/xrpc.zig
@@ -37,7 +37,7 @@ pub fn handleListRepos(conn: *h.Conn, query: []const u8, persist: *event_log_mod
 
     // query accounts with repo state, paginated by UID
     // includes both local status and upstream_status for combined active check
-    var result = persist.db.query(
+    var result = persist.ev_db.?.query(
         \\SELECT a.uid, a.did, a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_data_cid, '')
         \\FROM account a LEFT JOIN account_repo r ON a.uid = r.uid
         \\WHERE a.uid > $1 ORDER BY a.uid ASC LIMIT $2
@@ -122,7 +122,7 @@ pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, persist: *event_log
     }
 
     // look up account (includes both local and upstream status)
-    var row = (persist.db.rowUnsafe(
+    var row = (persist.ev_db.?.rowUnsafe(
         "SELECT a.uid, a.status, a.upstream_status, COALESCE(r.rev, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1",
         .{did},
     ) catch {
@@ -181,7 +181,7 @@ pub fn handleGetRepo(conn: *h.Conn, query: []const u8, persist: *event_log_mod.D
     }
 
     // look up the PDS hostname for this account
-    var row = (persist.db.rowUnsafe(
+    var row = (persist.ev_db.?.rowUnsafe(
         "SELECT h.hostname FROM account a JOIN host h ON a.host_id = h.id WHERE a.did = $1 AND a.host_id > 0",
         .{did},
     ) catch {
@@ -215,7 +215,7 @@ pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, persist: *event_l
     }
 
     // look up account + repo state (includes both local and upstream status)
-    var row = (persist.db.rowUnsafe(
+    var row = (persist.ev_db.?.rowUnsafe(
         "SELECT a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_data_cid, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1",
         .{did},
     ) catch {
@@ -345,7 +345,7 @@ pub fn handleListHosts(conn: *h.Conn, query: []const u8, persist: *event_log_mod
         return;
     }
 
-    var result = persist.db.query(
+    var result = persist.ev_db.?.query(
         "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2",
         .{ cursor_val, limit },
     ) catch {
@@ -400,7 +400,7 @@ pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, persist: *event_log
     };
 
     // look up host
-    var row = (persist.db.rowUnsafe(
+    var row = (persist.ev_db.?.rowUnsafe(
         "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1",
         .{hostname},
     ) catch {
@@ -426,7 +426,7 @@ pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, persist: *event_log
         raw_status; // active, idle pass through
 
     // count accounts on this host
-    const account_count: i64 = if (persist.db.rowUnsafe(
+    const account_count: i64 = if (persist.ev_db.?.rowUnsafe(
         "SELECT COUNT(*) FROM account WHERE host_id = $1",
         .{host_id},
     ) catch null) |cnt_row| blk: {
@@ -473,8 +473,8 @@ pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, slurper: *slurper_mod
     };
     defer slurper.allocator.free(hostname);
 
-    // fast validation: domain ban check
-    if (slurper.persist.isDomainBanned(hostname)) {
+    // fast validation: domain ban check (Evented fiber — use Ev pool)
+    if (slurper.persist.isDomainBannedEv(hostname)) {
         log.warn("requestCrawl rejected '{s}': domain banned", .{hostname});
         h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"domain is banned\"}");
         return;
diff --git a/src/broadcaster.zig b/src/broadcaster.zig
index 2d8898e..2197f53 100644
--- a/src/broadcaster.zig
+++ b/src/broadcaster.zig
@@ -611,26 +611,46 @@ pub const Broadcaster = struct {
     /// two-phase cursor replay: disk (diskpersist) first, then in-memory ring buffer.
     /// the consumer is already in the live broadcast list, so frames arriving
     /// during replay are buffered — no gap possible.
+    ///
+    /// disk playback runs on pool_io (Threaded) via request/reply — playback()
+    /// holds the DiskPersist mutex and reads files, all of which require Threaded Io.
     pub fn replayTo(self: *Broadcaster, consumer: *Consumer, cursor: u64) void {
-        // phase 1: disk replay from diskpersist
+        // phase 1: disk replay from diskpersist via cross-Io request/reply
         if (self.persist) |dp| {
-            var entries: std.ArrayListUnmanaged(event_log_mod.PlaybackEntry) = .empty;
-            defer {
-                for (entries.items) |e| self.allocator.free(e.data);
-                entries.deinit(self.allocator);
+            var req: event_log_mod.PlaybackRequest = .{ .since = cursor, .allocator = self.allocator };
+            dp.enqueuePlayback(&req);
+
+            // poll until pool_io worker completes the request (yields to Evented scheduler).
+            // SAFETY: req is stack-local — we MUST wait for the worker to finish before
+            // returning, otherwise the stack frame unwinds while the worker still holds &req.
+            // if sleep fails (shutdown/io error), fall back to spin-wait. the host_ops worker
+            // drains all pending playback requests before exiting, so this is bounded.
+            while (!req.done.load(.acquire)) {
+                self.io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch {
+                    while (!req.done.load(.acquire)) std.atomic.spinLoopHint();
+                    break;
+                };
             }
 
-            dp.playback(cursor, self.allocator, &entries) catch |err| {
+            if (req.err) |err| {
                 log.warn("disk replay failed: {s}, falling back to memory", .{@errorName(err)});
+                // clean up any partial entries
+                for (req.entries.items) |e| self.allocator.free(e.data);
+                req.entries.deinit(self.allocator);
                 self.replayFromMemory(consumer, cursor);
                 return;
-            };
+            }
+
+            defer {
+                for (req.entries.items) |e| self.allocator.free(e.data);
+                req.entries.deinit(self.allocator);
+            }
 
             var replayed: usize = 0;
             var reseq_arena = std.heap.ArenaAllocator.init(self.allocator);
             defer reseq_arena.deinit();
 
-            for (entries.items) |entry| {
+            for (req.entries.items) |entry| {
                 // resequence: replace upstream seq in CBOR with relay-assigned seq
                 const frame_data = resequenceFrame(reseq_arena.allocator(), entry.data, entry.seq) orelse entry.data;
                 if (!consumer.enqueueRaw(frame_data)) {
@@ -762,7 +782,7 @@ pub const Handler = struct {
             // OutdatedCursor: cursor older than oldest available — info, continue
             const oldest = blk: {
                 if (ctx.persist) |dp| {
-                    if (dp.firstSeq()) |s| break :blk s;
+                    if (dp.firstSeqEv()) |s| break :blk s;
                 }
                 break :blk ctx.history.oldestSeq() orelse 0;
             };
diff --git a/src/event_log.zig b/src/event_log.zig
index 2d0f605..10717d1 100644
--- a/src/event_log.zig
+++ b/src/event_log.zig
@@ -77,11 +77,25 @@ const PersistJob = struct {
 
 // --- disk persistence ---
 
+/// cross-Io playback request — Evented fiber posts, pool_io worker executes
+pub const PlaybackRequest = struct {
+    since: u64,
+    allocator: Allocator,
+    entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty,
+    err: ?anyerror = null,
+    done: std.atomic.Value(bool) = .{ .raw = false },
+    next: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
+};
+
 pub const DiskPersist = struct {
     allocator: Allocator,
     dir_path: []const u8,
     dir: Io.Dir,
     db: *pg.Pool,
+    /// Evented-safe pg.Pool — created on the Evented Io backend.
+    /// Evented callers (slurper, API handlers, broadcaster) use this pool
+    /// for pure DB reads/writes. pool_io callers keep using self.db.
+    ev_db: ?*pg.Pool = null,
     current_file: ?Io.File = null,
     current_file_path: ?[]const u8 = null,
     current_file_pos: u64 = 0,
@@ -113,6 +127,9 @@ pub const DiskPersist = struct {
     /// read by metrics server to report health without cross-Io pg.Pool access.
     last_db_success: std.atomic.Value(i64) = .{ .raw = 0 },
 
+    /// MPSC queue for cross-Io playback requests (Evented → pool_io)
+    playback_head: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
+
     /// current evtbuf entry count (for metrics — non-blocking, returns 0 if lock is contended)
     pub fn evtbufLen(self: *DiskPersist) usize {
         if (!self.mutex.tryLock()) return 0;
@@ -276,6 +293,7 @@ pub const DiskPersist = struct {
         if (self.current_file) |f| f.close(self.io);
         if (self.current_file_path) |p| self.allocator.free(p);
         self.dir.close(self.io);
+        if (self.ev_db) |ev| ev.deinit();
         self.db.deinit();
         self.allocator.free(self.dir_path);
     }
@@ -418,9 +436,18 @@ pub const DiskPersist = struct {
         return if (hid > 0) @intCast(hid) else 0;
     }
 
-    /// count accounts on a host (for rate limit scaling, matches Go relay's host.AccountCount)
+    /// count accounts on a host (Threaded pool)
     pub fn getHostAccountCount(self: *DiskPersist, host_id: u64) u64 {
-        var row = (self.db.rowUnsafe(
+        return getHostAccountCountImpl(host_id, self.db);
+    }
+
+    /// count accounts on a host (Evented pool)
+    pub fn getHostAccountCountEv(self: *DiskPersist, host_id: u64) u64 {
+        return getHostAccountCountImpl(host_id, self.ev_db.?);
+    }
+
+    fn getHostAccountCountImpl(host_id: u64, db: *pg.Pool) u64 {
+        var row = (db.rowUnsafe(
             "SELECT COUNT(*) FROM account WHERE host_id = $1",
             .{@as(i64, @intCast(host_id))},
         ) catch return 0) orelse return 0;
@@ -429,10 +456,19 @@ pub const DiskPersist = struct {
         return if (count > 0) @intCast(count) else 0;
     }
 
-    /// effective account count for rate limit scaling.
-    /// uses admin-configured limit if set, otherwise actual COUNT(*).
+    /// effective account count (Threaded pool)
     pub fn getEffectiveAccountCount(self: *DiskPersist, host_id: u64) u64 {
-        var row = (self.db.rowUnsafe(
+        return getEffectiveAccountCountImpl(host_id, self.db);
+    }
+
+    /// effective account count (Evented pool)
+    pub fn getEffectiveAccountCountEv(self: *DiskPersist, host_id: u64) u64 {
+        return getEffectiveAccountCountImpl(host_id, self.ev_db.?);
+    }
+
+    /// uses admin-configured limit if set, otherwise actual COUNT(*).
+    fn getEffectiveAccountCountImpl(host_id: u64, db: *pg.Pool) u64 {
+        var row = (db.rowUnsafe(
             "SELECT COALESCE(h.account_limit, COUNT(a.uid)) FROM host h LEFT JOIN account a ON a.host_id = h.id WHERE h.id = $1 GROUP BY h.id",
             .{@as(i64, @intCast(host_id))},
         ) catch return 0) orelse return 0;
@@ -441,17 +477,26 @@ pub const DiskPersist = struct {
         return if (count > 0) @intCast(count) else 0;
     }
 
-    /// set admin-configured account limit for a host (overrides COUNT(*) for rate limiting).
-    /// pass null to clear the override and revert to actual COUNT(*).
+    /// set host account limit (Threaded pool)
     pub fn setHostAccountLimit(self: *DiskPersist, host_id: u64, limit: ?u64) !void {
+        return setHostAccountLimitImpl(host_id, limit, self.db);
+    }
+
+    /// set host account limit (Evented pool)
+    pub fn setHostAccountLimitEv(self: *DiskPersist, host_id: u64, limit: ?u64) !void {
+        return setHostAccountLimitImpl(host_id, limit, self.ev_db.?);
+    }
+
+    /// pass null to clear the override and revert to actual COUNT(*).
+    fn setHostAccountLimitImpl(host_id: u64, limit: ?u64, db: *pg.Pool) !void {
         if (limit) |l| {
             const clamped: i64 = if (l > @as(u64, @intCast(std.math.maxInt(i64)))) std.math.maxInt(i64) else @intCast(l);
-            _ = try self.db.exec(
+            _ = try db.exec(
                 "UPDATE host SET account_limit = $2, updated_at = now() WHERE id = $1",
                 .{ @as(i64, @intCast(host_id)), clamped },
             );
         } else {
-            _ = try self.db.exec(
+            _ = try db.exec(
                 "UPDATE host SET account_limit = NULL, updated_at = now() WHERE id = $1",
                 .{@as(i64, @intCast(host_id))},
             );
@@ -502,9 +547,20 @@ pub const DiskPersist = struct {
         account_limit: ?u64 = null,
     };
 
-    /// get or create a host row. returns {id, last_seq}.
-    pub fn getOrCreateHost(self: *DiskPersist, hostname: []const u8) !struct { id: u64, last_seq: u64 } {
-        _ = self.db.exec(
+    const HostResult = struct { id: u64, last_seq: u64 };
+
+    /// get or create a host row (Threaded pool)
+    pub fn getOrCreateHost(self: *DiskPersist, hostname: []const u8) !HostResult {
+        return getOrCreateHostImpl(hostname, self.db);
+    }
+
+    /// get or create a host row (Evented pool)
+    pub fn getOrCreateHostEv(self: *DiskPersist, hostname: []const u8) !HostResult {
+        return getOrCreateHostImpl(hostname, self.ev_db.?);
+    }
+
+    fn getOrCreateHostImpl(hostname: []const u8, db: *pg.Pool) !HostResult {
+        _ = db.exec(
             "INSERT INTO host (hostname) VALUES ($1) ON CONFLICT (hostname) DO NOTHING",
             .{hostname},
         ) catch |err| {
@@ -512,7 +568,7 @@ pub const DiskPersist = struct {
             return err;
         };
 
-        var row = try self.db.rowUnsafe(
+        var row = try db.rowUnsafe(
             "SELECT id, last_seq FROM host WHERE hostname = $1",
             .{hostname},
         ) orelse return error.HostCreationFailed;
@@ -523,9 +579,18 @@ pub const DiskPersist = struct {
         };
     }
 
-    /// check if a host is banned or blocked by status
+    /// check if a host is banned or blocked by status (Threaded pool)
     pub fn isHostBanned(self: *DiskPersist, hostname: []const u8) bool {
-        var row = self.db.rowUnsafe(
+        return isHostBannedImpl(hostname, self.db);
+    }
+
+    /// check if a host is banned or blocked by status (Evented pool)
+    pub fn isHostBannedEv(self: *DiskPersist, hostname: []const u8) bool {
+        return isHostBannedImpl(hostname, self.ev_db.?);
+    }
+
+    fn isHostBannedImpl(hostname: []const u8, db: *pg.Pool) bool {
+        var row = db.rowUnsafe(
             "SELECT status FROM host WHERE hostname = $1",
             .{hostname},
         ) catch return false;
@@ -545,9 +610,18 @@ pub const DiskPersist = struct {
         );
     }
 
-    /// look up host ID by hostname. returns null if not found.
+    /// look up host ID by hostname (Threaded pool)
     pub fn getHostIdForHostname(self: *DiskPersist, hostname: []const u8) !?u64 {
-        var row = (try self.db.rowUnsafe(
+        return getHostIdForHostnameImpl(hostname, self.db);
+    }
+
+    /// look up host ID by hostname (Evented pool)
+    pub fn getHostIdForHostnameEv(self: *DiskPersist, hostname: []const u8) !?u64 {
+        return getHostIdForHostnameImpl(hostname, self.ev_db.?);
+    }
+
+    fn getHostIdForHostnameImpl(hostname: []const u8, db: *pg.Pool) !?u64 {
+        var row = (try db.rowUnsafe(
             "SELECT id FROM host WHERE hostname = $1",
             .{hostname},
         )) orelse return null;
@@ -555,26 +629,44 @@ pub const DiskPersist = struct {
         return @intCast(row.get(i64, 0));
     }
 
-    /// update host status (active, blocked, exhausted)
+    /// update host status (Threaded pool)
     pub fn updateHostStatus(self: *DiskPersist, host_id: u64, status: []const u8) !void {
-        _ = try self.db.exec(
+        return updateHostStatusImpl(host_id, status, self.db);
+    }
+
+    /// update host status (Evented pool)
+    pub fn updateHostStatusEv(self: *DiskPersist, host_id: u64, status: []const u8) !void {
+        return updateHostStatusImpl(host_id, status, self.ev_db.?);
+    }
+
+    fn updateHostStatusImpl(host_id: u64, status: []const u8, db: *pg.Pool) !void {
+        _ = try db.exec(
             "UPDATE host SET status = $2, updated_at = now() WHERE id = $1",
             .{ @as(i64, @intCast(host_id)), status },
         );
     }
 
-    /// list all active hosts
+    /// list all active hosts (Threaded pool)
     pub fn listActiveHosts(self: *DiskPersist, allocator: Allocator) ![]Host {
+        return listActiveHostsImpl(allocator, self.db);
+    }
+
+    /// list all active hosts (Evented pool)
+    pub fn listActiveHostsEv(self: *DiskPersist, allocator: Allocator) ![]Host {
+        return listActiveHostsImpl(allocator, self.ev_db.?);
+    }
+
+    fn listActiveHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
         var hosts: std.ArrayListUnmanaged(Host) = .empty;
         errdefer {
-            for (hosts.items) |h| {
-                allocator.free(h.hostname);
-                allocator.free(h.status);
+            for (hosts.items) |host| {
+                allocator.free(host.hostname);
+                allocator.free(host.status);
             }
             hosts.deinit(allocator);
         }
 
-        var result = try self.db.query(
+        var result = try db.query(
             "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host WHERE status = 'active' ORDER BY id ASC",
             .{},
         );
@@ -594,18 +686,27 @@ pub const DiskPersist = struct {
         return try hosts.toOwnedSlice(allocator);
     }
 
-    /// list all hosts (any status) for admin view
+    /// list all hosts (Threaded pool)
     pub fn listAllHosts(self: *DiskPersist, allocator: Allocator) ![]Host {
+        return listAllHostsImpl(allocator, self.db);
+    }
+
+    /// list all hosts (Evented pool)
+    pub fn listAllHostsEv(self: *DiskPersist, allocator: Allocator) ![]Host {
+        return listAllHostsImpl(allocator, self.ev_db.?);
+    }
+
+    fn listAllHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
         var hosts: std.ArrayListUnmanaged(Host) = .empty;
         errdefer {
-            for (hosts.items) |h| {
-                allocator.free(h.hostname);
-                allocator.free(h.status);
+            for (hosts.items) |host| {
+                allocator.free(host.hostname);
+                allocator.free(host.status);
             }
             hosts.deinit(allocator);
         }
 
-        var result = try self.db.query(
+        var result = try db.query(
             "SELECT id, hostname, status, last_seq, failed_attempts, account_limit FROM host ORDER BY id ASC",
             .{},
         );
@@ -639,14 +740,23 @@ pub const DiskPersist = struct {
         return @intCast(row.get(i32, 0));
     }
 
-    /// check if a hostname (or any parent domain) is banned.
-    /// Go relay: domain_ban.go DomainIsBanned — suffix-based check.
+    /// check if a hostname (or any parent domain) is banned (Threaded pool).
     pub fn isDomainBanned(self: *DiskPersist, hostname: []const u8) bool {
+        return self.isDomainBannedImpl(hostname, self.db);
+    }
+
+    /// check if a hostname (or any parent domain) is banned (Evented pool).
+    pub fn isDomainBannedEv(self: *DiskPersist, hostname: []const u8) bool {
+        return self.isDomainBannedImpl(hostname, self.ev_db.?);
+    }
+
+    /// Go relay: domain_ban.go DomainIsBanned — suffix-based check.
+    fn isDomainBannedImpl(_: *DiskPersist, hostname: []const u8, db: *pg.Pool) bool {
         // check each suffix: "pds.host.example.com", "host.example.com", "example.com"
         var offset: usize = 0;
         while (offset < hostname.len) {
             const suffix = hostname[offset..];
-            var row = self.db.rowUnsafe(
+            var row = db.rowUnsafe(
                 "SELECT 1 FROM domain_ban WHERE domain = $1",
                 .{suffix},
             ) catch return false;
@@ -662,14 +772,73 @@ pub const DiskPersist = struct {
         return false;
     }
 
-    /// reset failure count (on successful connection)
+    /// reset failure count (Threaded pool)
     pub fn resetHostFailures(self: *DiskPersist, host_id: u64) !void {
-        _ = try self.db.exec(
+        return resetHostFailuresImpl(host_id, self.db);
+    }
+
+    /// reset failure count (Evented pool)
+    pub fn resetHostFailuresEv(self: *DiskPersist, host_id: u64) !void {
+        return resetHostFailuresImpl(host_id, self.ev_db.?);
+    }
+
+    fn resetHostFailuresImpl(host_id: u64, db: *pg.Pool) !void {
+        _ = try db.exec(
             "UPDATE host SET failed_attempts = 0, updated_at = now() WHERE id = $1",
             .{@as(i64, @intCast(host_id))},
         );
     }
 
+    /// resolve a DID to UID using the Evented pool. skips the DID cache
+    /// (which uses pool_io mutex). only used from admin ban (rare path).
+    pub fn uidForDidEv(self: *DiskPersist, did: []const u8) !u64 {
+        const db = self.ev_db.?;
+        // check database
+        if (try db.rowUnsafe(
+            "SELECT uid FROM account WHERE did = $1",
+            .{did},
+        )) |row| {
+            var r = row;
+            defer r.deinit() catch {};
+            return @intCast(r.get(i64, 0));
+        }
+
+        // create new account row
+        _ = db.exec(
+            "INSERT INTO account (did) VALUES ($1) ON CONFLICT (did) DO NOTHING",
+            .{did},
+        ) catch |err| {
+            log.warn("failed to create account for {s}: {s}", .{ did, @errorName(err) });
+            return err;
+        };
+
+        // read back the UID
+        var row = try db.rowUnsafe(
+            "SELECT uid FROM account WHERE did = $1",
+            .{did},
+        ) orelse return error.AccountCreationFailed;
+        defer row.deinit() catch {};
+        return @intCast(row.get(i64, 0));
+    }
+
+    /// enqueue a playback request for the pool_io worker to execute.
+    /// uses MPSC push (lock-free Treiber stack).
+    pub fn enqueuePlayback(self: *DiskPersist, req: *PlaybackRequest) void {
+        while (true) {
+            const head = self.playback_head.load(.acquire);
+            req.next.store(head, .monotonic);
+            if (self.playback_head.cmpxchgWeak(@as(?*PlaybackRequest, head), req, .release, .monotonic) == null) {
+                return;
+            }
+        }
+    }
+
+    /// pop all pending playback requests (single-consumer — pool_io worker only).
+    /// returns a singly-linked list via .next pointers, or null if empty.
+    pub fn popPlaybackBatch(self: *DiskPersist) ?*PlaybackRequest {
+        return self.playback_head.swap(null, .acq_rel);
+    }
+
     /// persist an event. assigns a sequence number. returns the assigned seq.
     /// the event is buffered and will be flushed to disk asynchronously.
     pub fn persist(self: *DiskPersist, kind: EvtKind, uid: u64, payload: []const u8) !u64 {
@@ -766,9 +935,18 @@ pub const DiskPersist = struct {
         return self.cur_seq - 1;
     }
 
-    /// oldest available sequence number on disk, or null if no log files exist
+    /// oldest available sequence number (Threaded pool)
     pub fn firstSeq(self: *DiskPersist) ?u64 {
-        if (self.db.rowUnsafe(
+        return firstSeqImpl(self.db);
+    }
+
+    /// oldest available sequence number (Evented pool)
+    pub fn firstSeqEv(self: *DiskPersist) ?u64 {
+        return firstSeqImpl(self.ev_db.?);
+    }
+
+    fn firstSeqImpl(db: *pg.Pool) ?u64 {
+        if (db.rowUnsafe(
             "SELECT seq_start FROM log_file_refs ORDER BY seq_start ASC LIMIT 1",
             .{},
         ) catch null) |row| {
diff --git a/src/frame_worker.zig b/src/frame_worker.zig
index 7ae227e..ff41796 100644
--- a/src/frame_worker.zig
+++ b/src/frame_worker.zig
@@ -273,33 +273,32 @@ pub fn processFrame(work: *FrameWork) void {
     else
         .identity;
 
-    // persist under narrow ordering lock (seq assignment only), then
-    // resequence + enqueue outside the lock. slight out-of-order in the
-    // ring is fine — seq is embedded in frame data and consumers track by seq.
+    // persist + resequence + enqueue under ordering lock to guarantee
+    // broadcast_queue insertion order matches seq assignment order.
     if (work.persist) |dp| {
-        const relay_seq = blk: {
-            var spins: u64 = 0;
-            while (work.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
-                spins += 1;
-                std.atomic.spinLoopHint();
-            }
-            if (spins > 0) {
-                _ = work.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic);
-            }
+        var spins: u64 = 0;
+        while (work.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+            spins += 1;
+            std.atomic.spinLoopHint();
+        }
+        if (spins > 0) {
+            _ = work.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic);
+        }
 
-            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 relay_seq = dp.persist(kind, uid, data) catch |err| {
             work.bc.persist_order.store(0, .release);
-            break :blk seq;
+            log.warn("persist failed: {s}", .{@errorName(err)});
+            return;
         };
+        work.bc.stats.relay_seq.store(relay_seq, .release);
 
         const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data;
-        const owned = work.allocator.dupe(u8, broadcast_data) catch return;
+        const owned = work.allocator.dupe(u8, broadcast_data) catch {
+            work.bc.persist_order.store(0, .release);
+            return;
+        };
         work.bc.broadcast_queue.push(relay_seq, owned, &work.bc.stats);
+        work.bc.persist_order.store(0, .release);
 
         // update per-DID state outside the ordering lock (Postgres round-trip)
         if ((is_commit or is_sync) and uid > 0) {
diff --git a/src/host_ops.zig b/src/host_ops.zig
index e976c77..0b05b2c 100644
--- a/src/host_ops.zig
+++ b/src/host_ops.zig
@@ -13,6 +13,7 @@
 
 const std = @import("std");
 const event_log_mod = @import("event_log.zig");
+const broadcaster = @import("broadcaster.zig");
 
 const Io = std.Io;
 const log = std.log.scoped(.relay);
@@ -132,6 +133,7 @@ pub const HostOp = struct {
         increment_failures,
         reset_failures,
         update_status,
+        takedown_user,
     };
 
     pub const Payload = union {
@@ -139,6 +141,7 @@ pub const HostOp = struct {
         host_shutdown: *std.atomic.Value(bool),
         none: void,
         status: Status,
+        takedown: Takedown,
 
         pub const Status = struct {
             buf: [16]u8 = .{0} ** 16,
@@ -156,6 +159,17 @@ pub const HostOp = struct {
                 return self.buf[0..self.len];
             }
         };
+
+        /// inline buffer for takedown — #account CBOR frame is <200 bytes
+        pub const Takedown = struct {
+            uid: u64,
+            frame_buf: [256]u8 = .{0} ** 256,
+            frame_len: u16 = 0,
+
+            pub fn frameSlice(self: *const Takedown) []const u8 {
+                return self.frame_buf[0..self.frame_len];
+            }
+        };
     };
 };
 
@@ -172,6 +186,7 @@ pub const HostOpsQueue = struct {
     cursor_map: *CursorMap,
     shutdown: *std.atomic.Value(bool),
     max_consecutive_failures: u32,
+    bc: ?*broadcaster.Broadcaster = null,
 
     /// push a rare op (called from any Io context). spins until space is available.
     /// only used for failures/status — cursors go through CursorMap.
@@ -209,18 +224,21 @@ pub const HostOpsQueue = struct {
     }
 
     /// worker thread entry point. runs on pool_io (Threaded).
-    /// drains rare ops immediately, sweeps cursor map every 5s.
+    /// drains rare ops + playback requests immediately, sweeps cursor map every 5s.
     pub fn run(self: *HostOpsQueue, pool_io: Io) void {
         var last_cursor_flush: i64 = timestamp(pool_io);
 
         while (!self.shutdown.load(.acquire)) {
-            // drain rare ops (failures, status)
+            // drain rare ops (failures, status, takedowns)
             var drained: u32 = 0;
             while (self.pop()) |op| {
                 self.execute(op);
                 drained += 1;
             }
 
+            // drain playback requests (cross-Io: Evented fibers post, we execute)
+            drained += self.drainPlaybackRequests();
+
             // periodic cursor sweep
             const now = timestamp(pool_io);
             if (now - last_cursor_flush >= CURSOR_FLUSH_INTERVAL_SEC) {
@@ -239,6 +257,23 @@ pub const HostOpsQueue = struct {
         while (self.pop()) |op| {
             self.execute(op);
         }
+        _ = self.drainPlaybackRequests();
+    }
+
+    /// drain all pending playback requests from the MPSC queue
+    fn drainPlaybackRequests(self: *HostOpsQueue) u32 {
+        var maybe_batch = self.persist.popPlaybackBatch();
+        var count: u32 = 0;
+        // Treiber stack pops in LIFO order — fine for playback (each request is independent)
+        while (maybe_batch) |req| {
+            maybe_batch = req.next.load(.acquire);
+            self.persist.playback(req.since, req.allocator, &req.entries) catch |e| {
+                req.err = e;
+            };
+            req.done.store(true, .release);
+            count += 1;
+        }
+        return count;
     }
 
     fn execute(self: *HostOpsQueue, op: HostOp) void {
@@ -261,6 +296,43 @@ pub const HostOpsQueue = struct {
                     log.debug("host_ops: update status failed for host_id={d}: {s}", .{ op.host_id, @errorName(err) });
                 };
             },
+            .takedown_user => {
+                self.executeTakedown(op.payload.takedown);
+            },
+        }
+    }
+
+    /// execute takedown on pool_io: takeDownUser + persist + broadcast
+    fn executeTakedown(self: *HostOpsQueue, td: HostOp.Payload.Takedown) void {
+        self.persist.takeDownUser(td.uid) catch |err| {
+            log.warn("host_ops: takedown failed for uid={d}: {s}", .{ td.uid, @errorName(err) });
+            return;
+        };
+
+        if (td.frame_len == 0) return;
+        const frame_bytes = td.frameSlice();
+        const bc = self.bc orelse return;
+
+        // persist the #account event under ordering lock
+        while (bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+            std.atomic.spinLoopHint();
+        }
+
+        if (self.persist.persist(.account, td.uid, frame_bytes)) |relay_seq| {
+            bc.stats.relay_seq.store(relay_seq, .release);
+
+            const broadcast_data = broadcaster.resequenceFrame(self.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes;
+            const owned = self.persist.allocator.dupe(u8, broadcast_data) catch {
+                bc.persist_order.store(0, .release);
+                log.warn("host_ops: failed to alloc broadcast data for takedown uid={d}", .{td.uid});
+                return;
+            };
+            bc.broadcast_queue.push(relay_seq, owned, &bc.stats);
+            bc.persist_order.store(0, .release);
+            log.info("host_ops: emitted #account takedown for uid={d} (seq={d})", .{ td.uid, relay_seq });
+        } else |err| {
+            bc.persist_order.store(0, .release);
+            log.warn("host_ops: persist #account takedown failed: {s}", .{@errorName(err)});
         }
     }
 
diff --git a/src/main.zig b/src/main.zig
index 50cc62d..91d6806 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -227,6 +227,21 @@ pub fn main() !void {
     dp.retention_hours = retention_hours;
     dp.max_dir_bytes = max_events_gb * 1024 * 1024 * 1024;
 
+    // create Evented pg.Pool — safe for use from Evented fibers (slurper, API, broadcaster).
+    // the Threaded pool (dp.db) is only safe from pool_io threads.
+    {
+        const uri = std.Uri.parse(database_url) catch {
+            log.err("failed to parse DATABASE_URL for ev_db", .{});
+            return error.InvalidDatabaseUrl;
+        };
+        const pg_mod = @import("pg");
+        dp.ev_db = pg_mod.Pool.initUri(allocator, io, uri, .{ .size = db_pool_size }) catch |err| {
+            log.err("failed to init Evented pg.Pool: {s}", .{@errorName(err)});
+            return err;
+        };
+    }
+    log.info("created Evented pg.Pool (size={d})", .{db_pool_size});
+
     if (dp.lastSeq()) |last| {
         log.info("event log recovered: last_seq={d}", .{last});
     }
@@ -247,10 +262,12 @@ pub fn main() !void {
     defer ci.deinit();
 
     // init backfiller (collection index backfill from source relay)
-    var backfiller = backfill_mod.Backfiller.init(allocator, &ci, dp.db, io);
+    // uses ev_db (Evented pool) — backfiller spawns Evented fibers via io.concurrent()
+    var backfiller = backfill_mod.Backfiller.init(allocator, &ci, dp.ev_db.?, io);
 
     // init cleaner (removes stale entries from collection index)
-    var cleaner = cleaner_mod.Cleaner.init(allocator, io, &ci, dp.db);
+    // uses ev_db (Evented pool) — cleaner spawns Evented fibers via io.concurrent()
+    var cleaner = cleaner_mod.Cleaner.init(allocator, io, &ci, dp.ev_db.?);
 
     // init resyncer (updates collection index on #sync events)
     // runs entirely on pool_io (Threaded) — enqueue() is called from frame worker
@@ -271,6 +288,7 @@ pub fn main() !void {
         .cursor_map = &cursor_map,
         .shutdown = &shutdown_flag,
         .max_consecutive_failures = 15,
+        .bc = &bc,
     };
     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)});
@@ -328,6 +346,7 @@ pub fn main() !void {
         .resyncer = &resyncer,
         .bc = &bc,
         .validator = &val,
+        .host_ops = &host_ops_queue,
         .pool_io = pool_io,
     };
     bc.http_fallback = api.handleHttpRequest;
diff --git a/src/slurper.zig b/src/slurper.zig
index aa49666..f667f46 100644
--- a/src/slurper.zig
+++ b/src/slurper.zig
@@ -362,11 +362,11 @@ pub const Slurper = struct {
                 const normalized = validateHostname(self.allocator, host.hostname) catch continue;
                 defer self.allocator.free(normalized);
 
-                // skip banned domains
-                if (self.persist.isDomainBanned(normalized)) continue;
+                // skip banned domains (Evented fiber — use Ev pool)
+                if (self.persist.isDomainBannedEv(normalized)) continue;
 
                 // insert into DB (no describeServer check — the seed relay already vetted them)
-                _ = self.persist.getOrCreateHost(normalized) catch continue;
+                _ = self.persist.getOrCreateHostEv(normalized) catch continue;
                 added += 1;
             }
             total += added;
@@ -417,23 +417,23 @@ pub const Slurper = struct {
         };
         defer self.allocator.free(hostname);
 
-        // step 2: domain ban check (suffix-based)
+        // step 2: domain ban check (Evented fiber — use Ev pool)
         // Go relay: domain_ban.go DomainIsBanned
-        if (self.persist.isDomainBanned(hostname)) {
+        if (self.persist.isDomainBannedEv(hostname)) {
             log.warn("host {s}: domain is banned, rejecting", .{hostname});
             return;
         }
 
-        // step 3: check if host is banned/blocked in DB
+        // step 3: check if host is banned/blocked in DB (Evented pool)
         // Go relay: crawl.go checks host.Status == HostStatusBanned
-        if (self.persist.isHostBanned(hostname)) {
+        if (self.persist.isHostBannedEv(hostname)) {
             log.warn("host {s}: banned/blocked in DB, rejecting", .{hostname});
             return;
         }
 
         // step 4: dedup — check if already tracked
         // Go relay: crawl.go CheckIfSubscribed
-        const host_info = try self.persist.getOrCreateHost(hostname);
+        const host_info = try self.persist.getOrCreateHostEv(hostname);
         {
             self.workers_mutex.lockUncancelable(self.io);
             defer self.workers_mutex.unlock(self.io);
@@ -450,11 +450,11 @@ pub const Slurper = struct {
             return;
         };
 
-        // reset status and failure count — host passed describeServer, give it a fresh start.
+        // reset status and failure count (Evented pool) — host passed describeServer, give it a fresh start.
         // without this, exhausted hosts accumulate failures across requestCrawl cycles
         // and immediately re-exhaust on a single failure.
-        self.persist.updateHostStatus(host_info.id, "active") catch {};
-        self.persist.resetHostFailures(host_info.id) catch {};
+        self.persist.updateHostStatusEv(host_info.id, "active") catch {};
+        self.persist.resetHostFailuresEv(host_info.id) catch {};
 
         try self.spawnWorker(host_info.id, hostname, host_info.last_seq);
         log.info("added host {s} (id={d})", .{ hostname, host_info.id });
@@ -468,7 +468,7 @@ pub const Slurper = struct {
         const sub = try self.allocator.create(subscriber_mod.Subscriber);
         errdefer self.allocator.destroy(sub);
 
-        const account_count: u64 = self.persist.getEffectiveAccountCount(host_id);
+        const account_count: u64 = self.persist.getEffectiveAccountCountEv(host_id);
 
         sub.* = subscriber_mod.Subscriber.init(
             self.allocator,
@@ -546,7 +546,7 @@ pub const Slurper = struct {
             log.info("no seed host configured, skipping bootstrap", .{});
         }
 
-        const hosts = self.persist.listActiveHosts(self.allocator) catch |err| {
+        const hosts = self.persist.listActiveHostsEv(self.allocator) catch |err| {
             log.err("failed to load hosts: {s}", .{@errorName(err)});
             return;
         };
diff --git a/src/subscriber.zig b/src/subscriber.zig
index 70f5cb4..51ed362 100644
--- a/src/subscriber.zig
+++ b/src/subscriber.zig
@@ -709,32 +709,32 @@ const FrameHandler = struct {
         else // is_identity (unknown types already filtered above)
             .identity;
 
-        // persist under narrow ordering lock (seq assignment only), then
-        // resequence + enqueue outside the lock.
+        // persist + resequence + enqueue under ordering lock to guarantee
+        // broadcast_queue insertion order matches seq assignment order.
         if (sub.persist) |dp| {
-            const relay_seq = blk: {
-                var spins: u64 = 0;
-                while (sub.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
-                    spins += 1;
-                    std.atomic.spinLoopHint();
-                }
-                if (spins > 0) {
-                    _ = sub.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic);
-                }
+            var spins: u64 = 0;
+            while (sub.bc.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+                spins += 1;
+                std.atomic.spinLoopHint();
+            }
+            if (spins > 0) {
+                _ = sub.bc.stats.persist_order_spins.fetchAdd(spins, .monotonic);
+            }
 
-                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 relay_seq = dp.persist(kind, uid, data) catch |err| {
                 sub.bc.persist_order.store(0, .release);
-                break :blk seq;
+                log.warn("persist failed: {s}", .{@errorName(err)});
+                return;
             };
+            sub.bc.stats.relay_seq.store(relay_seq, .release);
 
             const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data;
-            const owned = sub.allocator.dupe(u8, broadcast_data) catch return;
+            const owned = sub.allocator.dupe(u8, broadcast_data) catch {
+                sub.bc.persist_order.store(0, .release);
+                return;
+            };
             sub.bc.broadcast_queue.push(relay_seq, owned, &sub.bc.stats);
+            sub.bc.persist_order.store(0, .release);
 
             // update per-DID state outside the ordering lock (Postgres round-trip)
             if ((is_commit or is_sync) and uid > 0) {