diff --git a/src/backfill.zig b/src/backfill.zig
--- a/src/backfill.zig
+++ b/src/backfill.zig
@@ -27,12 +27,13 @@
     collection_index: *collection_index_mod.CollectionIndex,
     persist: *event_log_mod.DiskPersist,
     running: std.atomic.Value(bool),
-    future: ?Io.Future(void),
+    thread: ?std.Thread,
     source: []const u8,
     io: Io,
+    shutdown: *std.atomic.Value(bool),
 
-    fn db(self: *Backfiller) !*pg.Pool {
-        return self.persist.ensureEvDb();
+    fn db(self: *Backfiller) *pg.Pool {
+        return self.persist.db;
     }
 
     pub fn init(
@@ -40,20 +41,31 @@
         collection_index: *collection_index_mod.CollectionIndex,
         persist: *event_log_mod.DiskPersist,
         io: Io,
+        shutdown: *std.atomic.Value(bool),
     ) Backfiller {
         return .{
             .allocator = allocator,
             .collection_index = collection_index,
             .persist = persist,
             .running = .{ .raw = false },
-            .future = null,
+            .thread = null,
             .source = "",
             .io = io,
+            .shutdown = shutdown,
         };
     }
 
     pub fn isRunning(self: *Backfiller) bool {
         return self.running.load(.acquire);
+    }
+
+    /// block until any in-progress backfill thread completes.
+    /// must be called before tearing down DiskPersist or CollectionIndex.
+    pub fn waitForCompletion(self: *Backfiller) void {
+        if (self.thread) |t| {
+            t.join();
+            self.thread = null;
+        }
     }
 
     /// start a backfill from the given source relay. returns error if already running.
@@ -68,21 +80,20 @@
             self.allocator.free(self.source);
             self.source = "";
         }
-        self.future = try self.io.concurrent(run, .{self});
+        self.thread = std.Thread.spawn(.{}, run, .{self}) catch return error.SpawnFailed;
     }
 
     fn run(self: *Backfiller) void {
         defer {
             self.allocator.free(self.source);
             self.source = "";
-            self.future = null;
+            // note: do NOT clear self.thread here — waitForCompletion() needs
+            // the handle to join this thread before dp/ci teardown. the running
+            // flag gates start(), so a stale thread handle is harmless.
             self.running.store(false, .release);
         }
 
-        const pool = self.db() catch |err| {
-            log.err("backfill: database unavailable: {s}", .{@errorName(err)});
-            return;
-        };
+        const pool = self.db();
 
         // discover collections
         const collections = self.discoverCollections() catch |err| {
@@ -108,6 +119,10 @@
 
         // backfill each collection
         for (collections) |collection| {
+            if (self.shutdown.load(.acquire)) {
+                log.info("backfill interrupted by shutdown", .{});
+                return;
+            }
             self.backfillCollection(collection) catch |err| {
                 log.warn("backfill failed for {s}: {s}", .{ collection, @errorName(err) });
             };
@@ -212,7 +227,7 @@
     }
 
     fn backfillCollection(self: *Backfiller, collection: []const u8) !void {
-        const pool = try self.db();
+        const pool = self.db();
 
         // single query: check completion, get cursor + count for resume
         var cursor: ?[]const u8 = null;
@@ -241,7 +256,7 @@
         defer client.deinit();
 
         var page_count: usize = 0;
-        while (true) {
+        while (!self.shutdown.load(.acquire)) {
             const fetch_result = self.fetchPage(&client, collection, cursor) catch |err| {
                 log.warn("{s}: fetch page failed: {s}", .{ collection, @errorName(err) });
                 break;
@@ -349,7 +364,7 @@
 
     /// return status summary for the admin endpoint
     pub fn getStatus(self: *Backfiller, allocator: Allocator) ![]u8 {
-        const pool = try self.db();
+        const pool = self.db();
 
         var aw: Io.Writer.Allocating = .init(allocator);
         defer aw.deinit();
diff --git a/src/broadcaster.zig b/src/broadcaster.zig
--- a/src/broadcaster.zig
+++ b/src/broadcaster.zig
@@ -470,6 +470,7 @@
     broadcast_queue: BroadcastQueue,
     history: FrameHistory,
     persist: ?*event_log_mod.DiskPersist = null,
+    db_queue: ?*event_log_mod.DbRequestQueue = null,
     stats: Stats = .{},
     error_frame: ?[]const u8 = null,
     http_fallback: ?HttpFallbackFn = null,
@@ -781,8 +782,20 @@
 
             // OutdatedCursor: cursor older than oldest available — info, continue
             const oldest = blk: {
-                if (ctx.persist) |dp| {
-                    if (dp.firstSeqEv() catch null) |s| break :blk s;
+                if (ctx.persist != null and ctx.db_queue != null) {
+                    const FirstSeqReq = struct {
+                        base: event_log_mod.DbRequest = .{ .callback = &execute },
+                        result: ?u64 = null,
+
+                        fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
+                            const s: *@This() = @fieldParentPtr("base", b);
+                            s.result = dp.firstSeq();
+                        }
+                    };
+                    var first_req: FirstSeqReq = .{};
+                    ctx.db_queue.?.push(&first_req.base);
+                    first_req.base.wait(ctx.io, ctx.shutdown);
+                    if (first_req.result) |s| break :blk s;
                 }
                 break :blk ctx.history.oldestSeq() orelse 0;
             };
diff --git a/src/cleaner.zig b/src/cleaner.zig
--- a/src/cleaner.zig
+++ b/src/cleaner.zig
@@ -19,12 +19,13 @@
     collection_index: *collection_index_mod.CollectionIndex,
     persist: *event_log_mod.DiskPersist,
     running: std.atomic.Value(bool),
-    future: ?Io.Future(void),
+    thread: ?std.Thread,
     scanned: std.atomic.Value(u64),
     removed: std.atomic.Value(u64),
+    shutdown: *std.atomic.Value(bool),
 
-    fn db(self: *Cleaner) !*pg.Pool {
-        return self.persist.ensureEvDb();
+    fn db(self: *Cleaner) *pg.Pool {
+        return self.persist.db;
     }
 
     pub fn init(
@@ -32,6 +33,7 @@
         io: Io,
         collection_index: *collection_index_mod.CollectionIndex,
         persist: *event_log_mod.DiskPersist,
+        shutdown: *std.atomic.Value(bool),
     ) Cleaner {
         return .{
             .allocator = allocator,
@@ -39,14 +41,24 @@
             .collection_index = collection_index,
             .persist = persist,
             .running = .{ .raw = false },
-            .future = null,
+            .thread = null,
             .scanned = .{ .raw = 0 },
             .removed = .{ .raw = 0 },
+            .shutdown = shutdown,
         };
     }
 
     pub fn isRunning(self: *Cleaner) bool {
         return self.running.load(.acquire);
+    }
+
+    /// block until any in-progress cleanup thread completes.
+    /// must be called before tearing down DiskPersist or CollectionIndex.
+    pub fn waitForCompletion(self: *Cleaner) void {
+        if (self.thread) |t| {
+            t.join();
+            self.thread = null;
+        }
     }
 
     /// start cleanup. returns error if already running.
@@ -58,25 +70,23 @@
 
         self.scanned.store(0, .release);
         self.removed.store(0, .release);
-        self.future = try self.io.concurrent(run, .{self});
+        self.thread = std.Thread.spawn(.{}, run, .{self}) catch return error.SpawnFailed;
     }
 
     fn run(self: *Cleaner) void {
         defer {
-            self.future = null;
+            // note: do NOT clear self.thread here — waitForCompletion() needs
+            // the handle to join this thread before dp/ci teardown.
             self.running.store(false, .release);
         }
 
         log.info("cleanup started", .{});
 
-        const pool = self.db() catch |err| {
-            log.err("cleanup: database unavailable: {s}", .{@errorName(err)});
-            return;
-        };
+        const pool = self.db();
 
         // page through inactive accounts by uid
         var last_uid: i64 = 0;
-        while (true) {
+        while (!self.shutdown.load(.acquire)) {
             var batch_count: u64 = 0;
             {
                 var result = pool.query(
diff --git a/src/event_log.zig b/src/event_log.zig
--- a/src/event_log.zig
+++ b/src/event_log.zig
@@ -87,23 +87,101 @@
     next: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
 };
 
+/// cross-Io DB request — Evented fiber posts, pool_io worker executes.
+/// callers define typed structs embedding DbRequest + @fieldParentPtr.
+pub const DbRequest = struct {
+    callback: *const fn (*DbRequest, *DiskPersist) void,
+    done: std.atomic.Value(bool) = .{ .raw = false },
+    err: ?anyerror = null,
+
+    /// spin-wait for completion, yielding via io.sleep when available.
+    /// never returns until `done` is true — the request struct lives on the
+    /// caller's stack, so we must outlive the worker's callback execution.
+    /// the queue's shutdown drain sets done+err on unprocessed requests,
+    /// and workers set done after the callback returns, so this always terminates.
+    pub fn wait(self: *DbRequest, io: Io, shutdown: *std.atomic.Value(bool)) void {
+        _ = shutdown;
+        while (!self.done.load(.acquire)) {
+            io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch {
+                // sleep failed (Io shutting down) — busy-wait for worker to finish
+                while (!self.done.load(.acquire)) {
+                    std.atomic.spinLoopHint();
+                }
+                break;
+            };
+        }
+    }
+};
+
+/// MPSC FIFO ring buffer for general DB traffic.
+/// multiple producers (Evented fibers) push via spinlock,
+/// multiple consumers (pool_io worker threads) pop via CAS.
+pub const DbRequestQueue = struct {
+    const CAPACITY = 4096;
+
+    items: [CAPACITY]*DbRequest = undefined,
+    head: std.atomic.Value(u32) = .{ .raw = 0 },
+    tail: std.atomic.Value(u32) = .{ .raw = 0 },
+    push_lock: std.atomic.Value(u32) = .{ .raw = 0 }, // producer spinlock
+    shutdown: *std.atomic.Value(bool),
+    persist: *DiskPersist,
+
+    pub fn push(self: *DbRequestQueue, req: *DbRequest) void {
+        // acquire producer spinlock
+        while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+            std.atomic.spinLoopHint();
+        }
+        defer self.push_lock.store(0, .release);
+
+        const tail = self.tail.load(.monotonic);
+        const head = self.head.load(.acquire);
+        // if full, spin until space opens (workers are fast, this shouldn't happen)
+        if (tail -% head >= CAPACITY) {
+            self.push_lock.store(0, .release);
+            while (self.tail.load(.monotonic) -% self.head.load(.acquire) >= CAPACITY) {
+                std.atomic.spinLoopHint();
+            }
+            while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) {
+                std.atomic.spinLoopHint();
+            }
+        }
+
+        self.items[self.tail.load(.monotonic) % CAPACITY] = req;
+        self.tail.store(self.tail.load(.monotonic) +% 1, .release);
+    }
+
+    pub fn pop(self: *DbRequestQueue) ?*DbRequest {
+        while (true) {
+            const head = self.head.load(.acquire);
+            if (head == self.tail.load(.acquire)) return null;
+            const item = self.items[head % CAPACITY];
+            if (self.head.cmpxchgWeak(head, head +% 1, .acq_rel, .acquire) == null)
+                return item;
+        }
+    }
+
+    pub fn run(self: *DbRequestQueue, pool_io: Io) void {
+        while (!self.shutdown.load(.acquire)) {
+            if (self.pop()) |req| {
+                req.callback(req, self.persist);
+                req.done.store(true, .release);
+            } else {
+                pool_io.sleep(Io.Duration.fromMilliseconds(5), .awake) catch return;
+            }
+        }
+        // shutdown drain — signal error on unprocessed requests
+        while (self.pop()) |req| {
+            req.err = error.ShuttingDown;
+            req.done.store(true, .release);
+        }
+    }
+};
+
 pub const DiskPersist = struct {
     allocator: Allocator,
     dir_path: []const u8,
     dir: Io.Dir,
     db: *pg.Pool,
-    /// Evented-safe pg.Pool — lazy-initialized on first use from an Evented fiber.
-    /// pg.Pool.initUri requires the event loop to be running (TCP connect goes through
-    /// io_uring), so we can't create it during main() init. Evented callers access it
-    /// via ensureEvDb(). pool_io callers keep using self.db.
-    ev_db: ?*pg.Pool = null,
-    ev_db_state: std.atomic.Value(u8) = .{ .raw = 0 },
-    /// Evented Io — set by main before any fibers run. used for lazy ev_db init.
-    ev_io: ?Io = null,
-    /// database URL for lazy ev_db init (stable ref to process env string)
-    db_url: []const u8 = "",
-    /// pool size for lazy ev_db init
-    ev_db_pool_size: u16 = 0,
     current_file: ?Io.File = null,
     current_file_path: ?[]const u8 = null,
     current_file_pos: u64 = 0,
@@ -137,58 +215,6 @@
 
     /// MPSC queue for cross-Io playback requests (Evented → pool_io)
     playback_head: std.atomic.Value(?*PlaybackRequest) = .{ .raw = null },
-
-    const EvDbInit = enum(u8) { uninit = 0, initializing = 1, ready = 2 };
-
-    /// returns the Evented pg.Pool, lazy-initializing on first call.
-    /// pg.Pool.initUri does TCP connects via io_uring, so it can only run inside
-    /// an Evented fiber (after the event loop is spinning). callers from main()
-    /// init set ev_io/db_url/ev_db_pool_size; the actual pool is created here.
-    /// on failure, resets state to uninit so the next call retries.
-    pub fn ensureEvDb(self: *DiskPersist) !*pg.Pool {
-        // fast path — already initialized (single atomic load)
-        if (self.ev_db) |db| return db;
-
-        const ev_io = self.ev_io orelse return error.EvDbNotConfigured;
-
-        while (true) {
-            const state: EvDbInit = @enumFromInt(self.ev_db_state.load(.acquire));
-            switch (state) {
-                .ready => return self.ev_db orelse error.EvDbNotConfigured,
-                .uninit => {
-                    if (self.ev_db_state.cmpxchgWeak(
-                        @intFromEnum(EvDbInit.uninit),
-                        @intFromEnum(EvDbInit.initializing),
-                        .acquire,
-                        .monotonic,
-                    ) == null) {
-                        // won the race — create the pool
-                        const uri = std.Uri.parse(self.db_url) catch
-                            return error.InvalidDatabaseUrl;
-                        self.ev_db = pg.Pool.initUri(
-                            self.allocator,
-                            ev_io,
-                            uri,
-                            .{ .size = self.ev_db_pool_size },
-                        ) catch |err| {
-                            log.err("ensureEvDb: initUri failed: {s}", .{@errorName(err)});
-                            // reset to uninit so next call retries
-                            self.ev_db_state.store(@intFromEnum(EvDbInit.uninit), .release);
-                            return error.EvDbInitFailed;
-                        };
-                        self.ev_db_state.store(@intFromEnum(EvDbInit.ready), .release);
-                        log.info("lazy-initialized Evented pg.Pool (size={d})", .{self.ev_db_pool_size});
-                        return self.ev_db.?;
-                    }
-                    // lost CAS — another fiber is initializing, fall through
-                },
-                .initializing => {
-                    // yield to let the initializing fiber complete
-                    ev_io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch {};
-                },
-            }
-        }
-    }
 
     /// current evtbuf entry count (for metrics — non-blocking, returns 0 if lock is contended)
     pub fn evtbufLen(self: *DiskPersist) usize {
@@ -353,7 +379,6 @@
         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);
     }
@@ -501,11 +526,6 @@
         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, try self.ensureEvDb());
-    }
-
     fn getHostAccountCountImpl(host_id: u64, db: *pg.Pool) u64 {
         var row = (db.rowUnsafe(
             "SELECT COUNT(*) FROM account WHERE host_id = $1",
@@ -519,11 +539,6 @@
     /// effective account count (Threaded pool)
     pub fn getEffectiveAccountCount(self: *DiskPersist, host_id: u64) u64 {
         return getEffectiveAccountCountImpl(host_id, self.db);
-    }
-
-    /// effective account count (Evented pool)
-    pub fn getEffectiveAccountCountEv(self: *DiskPersist, host_id: u64) !u64 {
-        return getEffectiveAccountCountImpl(host_id, try self.ensureEvDb());
     }
 
     /// uses admin-configured limit if set, otherwise actual COUNT(*).
@@ -540,11 +555,6 @@
     /// 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, try self.ensureEvDb());
     }
 
     /// pass null to clear the override and revert to actual COUNT(*).
@@ -614,11 +624,6 @@
         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, try self.ensureEvDb());
-    }
-
     fn getOrCreateHostImpl(hostname: []const u8, db: *pg.Pool) !HostResult {
         _ = db.exec(
             "INSERT INTO host (hostname) VALUES ($1) ON CONFLICT (hostname) DO NOTHING",
@@ -642,11 +647,6 @@
     /// check if a host is banned or blocked by status (Threaded pool)
     pub fn isHostBanned(self: *DiskPersist, hostname: []const u8) bool {
         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, try self.ensureEvDb());
     }
 
     fn isHostBannedImpl(hostname: []const u8, db: *pg.Pool) bool {
@@ -675,11 +675,6 @@
         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, try self.ensureEvDb());
-    }
-
     fn getHostIdForHostnameImpl(hostname: []const u8, db: *pg.Pool) !?u64 {
         var row = (try db.rowUnsafe(
             "SELECT id FROM host WHERE hostname = $1",
@@ -694,11 +689,6 @@
         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, try self.ensureEvDb());
-    }
-
     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",
@@ -709,11 +699,6 @@
     /// 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, try self.ensureEvDb());
     }
 
     fn listActiveHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
@@ -749,11 +734,6 @@
     /// 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, try self.ensureEvDb());
     }
 
     fn listAllHostsImpl(allocator: Allocator, db: *pg.Pool) ![]Host {
@@ -805,11 +785,6 @@
         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, try self.ensureEvDb());
-    }
-
     /// 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"
@@ -837,48 +812,11 @@
         return resetHostFailuresImpl(host_id, self.db);
     }
 
-    /// reset failure count (Evented pool)
-    pub fn resetHostFailuresEv(self: *DiskPersist, host_id: u64) !void {
-        return resetHostFailuresImpl(host_id, try self.ensureEvDb());
-    }
-
     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 = try self.ensureEvDb();
-        // 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.
@@ -998,11 +936,6 @@
     /// oldest available sequence number (Threaded pool)
     pub fn firstSeq(self: *DiskPersist) ?u64 {
         return firstSeqImpl(self.db);
-    }
-
-    /// oldest available sequence number (Evented pool)
-    pub fn firstSeqEv(self: *DiskPersist) !?u64 {
-        return firstSeqImpl(try self.ensureEvDb());
     }
 
     fn firstSeqImpl(db: *pg.Pool) ?u64 {
diff --git a/src/main.zig b/src/main.zig
--- a/src/main.zig
+++ b/src/main.zig
@@ -227,12 +227,12 @@
     dp.retention_hours = retention_hours;
     dp.max_dir_bytes = max_events_gb * 1024 * 1024 * 1024;
 
-    // configure lazy Evented pg.Pool — can't create it here because pg.Pool.initUri
-    // does TCP connects via io_uring, which requires the event loop to be running.
-    // the pool is created on first use from an Evented fiber (via dp.ensureEvDb()).
-    dp.ev_io = io;
-    dp.db_url = database_url;
-    dp.ev_db_pool_size = db_pool_size;
+    // DbRequestQueue — general DB traffic from Evented fibers routed to pool_io workers.
+    // replaces the broken ev_db (Evented pg.Pool) approach.
+    var db_queue: event_log_mod.DbRequestQueue = .{
+        .shutdown = &shutdown_flag,
+        .persist = &dp,
+    };
 
     if (dp.lastSeq()) |last| {
         log.info("event log recovered: last_seq={d}", .{last});
@@ -241,8 +241,19 @@
     // start flush thread
     try dp.start();
 
+    // spawn 2 DbRequestQueue worker threads on pool_io
+    const db_worker_1 = std.Thread.spawn(.{}, event_log_mod.DbRequestQueue.run, .{ &db_queue, pool_io }) catch |err| {
+        log.err("failed to start db queue worker 1: {s}", .{@errorName(err)});
+        return err;
+    };
+    const db_worker_2 = std.Thread.spawn(.{}, event_log_mod.DbRequestQueue.run, .{ &db_queue, pool_io }) catch |err| {
+        log.err("failed to start db queue worker 2: {s}", .{@errorName(err)});
+        return err;
+    };
+
     // wire persist into broadcaster for cursor replay and validator for migration checks
     bc.persist = &dp;
+    bc.db_queue = &db_queue;
     val.persist = &dp;
 
     // init collection index (RocksDB — inspired by lightrail/microcosm.blue)
@@ -254,12 +265,12 @@
     defer ci.deinit();
 
     // init backfiller (collection index backfill from source relay)
-    // uses ev_db (Evented pool) — backfiller spawns Evented fibers via io.concurrent()
-    var backfiller = backfill_mod.Backfiller.init(allocator, &ci, &dp, io);
+    // uses pool_io (Threaded) — backfiller spawns std.Thread, DNS works
+    var backfiller = backfill_mod.Backfiller.init(allocator, &ci, &dp, pool_io, &shutdown_flag);
 
     // init cleaner (removes stale entries from collection index)
-    // uses ev_db (Evented pool) — cleaner spawns Evented fibers via io.concurrent()
-    var cleaner = cleaner_mod.Cleaner.init(allocator, io, &ci, &dp);
+    // uses pool_io (Threaded) — cleaner spawns std.Thread, checks shutdown
+    var cleaner = cleaner_mod.Cleaner.init(allocator, pool_io, &ci, &dp, &shutdown_flag);
 
     // init resyncer (updates collection index on #sync events)
     // runs entirely on pool_io (Threaded) — enqueue() is called from frame worker
@@ -309,9 +320,24 @@
     slurper.resyncer = &resyncer;
     slurper.host_ops = &host_ops_queue;
     slurper.cursor_map = &cursor_map;
+    slurper.db_queue = &db_queue;
 
-    // start: loads active hosts from DB, spawns subscriber threads
+    // start: loads active hosts from DB, spawns subscriber threads.
+    // pullHosts runs on its own std.Thread (parallel, not gating spawnWorkers).
     try slurper.start();
+
+    // spawn pullHosts on a separate thread — runs in parallel with spawnWorkers,
+    // uses pool_io for HTTP (DNS works), writes to DB via Threaded pool directly.
+    const pull_hosts_thread = if (slurper.options.seed_host.len > 0) blk: {
+        log.info("pulling hosts from {s} (background thread)...", .{slurper.options.seed_host});
+        break :blk std.Thread.spawn(.{}, slurper_mod.Slurper.pullHosts, .{&slurper}) catch |err| {
+            log.warn("failed to spawn pullHosts thread: {s}", .{@errorName(err)});
+            break :blk null;
+        };
+    } else blk: {
+        log.info("no seed host configured, skipping bootstrap", .{});
+        break :blk null;
+    };
 
     // start broadcaster fiber — drains broadcast queue, owns all consumer state.
     // this is the Evented-side sequencer: frame workers push results to the queue,
@@ -340,6 +366,9 @@
         .validator = &val,
         .host_ops = &host_ops_queue,
         .pool_io = pool_io,
+        .io = io,
+        .db_queue = &db_queue,
+        .shutdown = &shutdown_flag,
     };
     bc.http_fallback = api.handleHttpRequest;
     bc.http_fallback_ctx = @ptrCast(&http_context);
@@ -399,8 +428,19 @@
     // must complete before dp.deinit() runs (dp is stack-owned).
     gc_thread.join();
 
+    // join pullHosts thread if running
+    if (pull_hosts_thread) |t| t.join();
+
+    // join db queue workers — shutdown flag already set, they will drain and exit
+    db_worker_1.join();
+    db_worker_2.join();
+
     // join host ops thread — drains remaining ops before dp.deinit()
     host_ops_thread.join();
+
+    // join backfiller/cleaner threads if running — they touch dp and ci
+    backfiller.waitForCompletion();
+    cleaner.waitForCompletion();
 
     // 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
--- a/src/slurper.zig
+++ b/src/slurper.zig
@@ -231,6 +231,7 @@
     resyncer: ?*resync_mod.Resyncer = null,
     host_ops: ?*host_ops_mod.HostOpsQueue = null,
     cursor_map: ?*host_ops_mod.CursorMap = null,
+    db_queue: ?*event_log_mod.DbRequestQueue = null,
     shutdown: *std.atomic.Value(bool),
     options: Options,
 
@@ -304,13 +305,14 @@
     }
 
     /// pull PDS host list from the seed relay's com.atproto.sync.listHosts endpoint.
+    /// runs on its own std.Thread (pool_io) — DNS and outbound HTTP work.
     /// Go relay: cmd/relay/pull.go — one-time bootstrap, reads REST API, not firehose.
-    pub fn pullHosts(self: *Slurper) !void {
+    pub fn pullHosts(self: *Slurper) void {
         var cursor: ?[]const u8 = null;
         var total: usize = 0;
         const limit = 500;
 
-        var client: http.Client = .{ .allocator = self.allocator, .io = self.io };
+        var client: http.Client = .{ .allocator = self.allocator, .io = self.pool_io };
         defer client.deinit();
 
         while (true) {
@@ -362,11 +364,11 @@
                 const normalized = validateHostname(self.allocator, host.hostname) catch continue;
                 defer self.allocator.free(normalized);
 
-                // skip banned domains (Evented fiber — use Ev pool)
-                if (self.persist.isDomainBannedEv(normalized) catch true) continue;
+                // skip banned domains (Threaded pool — runs on own std.Thread)
+                if (self.persist.isDomainBanned(normalized)) continue;
 
                 // insert into DB (no describeServer check — the seed relay already vetted them)
-                _ = self.persist.getOrCreateHostEv(normalized) catch continue;
+                _ = self.persist.getOrCreateHost(normalized) catch continue;
                 added += 1;
             }
             total += added;
@@ -408,7 +410,10 @@
 
     /// validate and add a host: format check, domain ban, describeServer, then spawn.
     /// mirrors Go relay's requestCrawl → SubscribeToHost pipeline.
+    /// uses phased approach: DB checks via DbRequestQueue, HTTP via temp thread.
     fn addHost(self: *Slurper, raw_hostname: []const u8) !void {
+        const db_queue = self.db_queue orelse return error.DbQueueNotConfigured;
+
         // step 1: validate and normalize hostname format
         // Go relay: host.go ParseHostname
         const hostname = validateHostname(self.allocator, raw_hostname) catch |err| {
@@ -417,47 +422,113 @@
         };
         defer self.allocator.free(hostname);
 
-        // step 2: domain ban check (Evented fiber — use Ev pool)
-        // Go relay: domain_ban.go DomainIsBanned
-        if (self.persist.isDomainBannedEv(hostname) catch true) {
-            log.warn("host {s}: domain is banned, rejecting", .{hostname});
+        // phase 1: DB checks via DbRequestQueue
+        const AddHostDbReq = struct {
+            base: event_log_mod.DbRequest = .{ .callback = &execute },
+            hostname: []const u8,
+            host_id: u64 = 0,
+            last_seq: u64 = 0,
+            rejected: bool = false,
+
+            fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
+                const s: *@This() = @fieldParentPtr("base", b);
+                // domain ban check
+                if (dp.isDomainBanned(s.hostname)) {
+                    s.rejected = true;
+                    return;
+                }
+                // host ban check
+                if (dp.isHostBanned(s.hostname)) {
+                    s.rejected = true;
+                    return;
+                }
+                // get or create host
+                const info = dp.getOrCreateHost(s.hostname) catch |e| {
+                    b.err = e;
+                    return;
+                };
+                s.host_id = info.id;
+                s.last_seq = info.last_seq;
+            }
+        };
+        var db_req: AddHostDbReq = .{ .hostname = hostname };
+        db_queue.push(&db_req.base);
+        db_req.base.wait(self.io, self.shutdown);
+        if (db_req.base.err != null) return error.DbRequestFailed;
+        if (db_req.rejected) {
+            log.warn("host {s}: banned/blocked, rejecting", .{hostname});
             return;
         }
 
-        // step 3: check if host is banned/blocked in DB (Evented pool)
-        // Go relay: crawl.go checks host.Status == HostStatusBanned
-        if (self.persist.isHostBannedEv(hostname) catch true) {
-            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.getOrCreateHostEv(hostname);
+        // phase 2: dedup — check if already tracked (Evented, local)
         {
             self.workers_mutex.lockUncancelable(self.io);
             defer self.workers_mutex.unlock(self.io);
-            if (self.workers.contains(host_info.id)) {
+            if (self.workers.contains(db_req.host_id)) {
                 log.debug("host {s} already has a worker, skipping", .{hostname});
                 return;
             }
         }
 
-        // step 5: describeServer liveness check
-        // Go relay: host_checker.go CheckHost (with SSRF protection)
-        checkHost(self.allocator, hostname, self.io) catch |err| {
-            log.warn("host {s}: describeServer check failed: {s}", .{ hostname, @errorName(err) });
+        // phase 3: describeServer liveness check (outbound HTTP via temp thread)
+        const CheckHostReq = struct {
+            allocator: Allocator,
+            hostname_copy: []const u8,
+            check_io: Io,
+            check_err: ?HostValidationError = null,
+            done: std.atomic.Value(bool) = .{ .raw = false },
+
+            fn run(s: *@This()) void {
+                checkHost(s.allocator, s.hostname_copy, s.check_io) catch |e| {
+                    s.check_err = e;
+                };
+                s.done.store(true, .release);
+            }
+        };
+        var check_req: CheckHostReq = .{
+            .allocator = self.allocator,
+            .hostname_copy = hostname,
+            .check_io = self.pool_io,
+        };
+        const check_thread = std.Thread.spawn(.{}, CheckHostReq.run, .{&check_req}) catch {
+            log.warn("host {s}: failed to spawn check thread", .{hostname});
             return;
         };
+        // wait for check to complete (Evented fiber yields)
+        while (!check_req.done.load(.acquire)) {
+            if (self.shutdown.load(.acquire)) return;
+            self.io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch {
+                while (!check_req.done.load(.acquire)) {
+                    if (self.shutdown.load(.acquire)) return;
+                    std.atomic.spinLoopHint();
+                }
+                break;
+            };
+        }
+        check_thread.join();
+        if (check_req.check_err) |err| {
+            log.warn("host {s}: describeServer check failed: {s}", .{ hostname, @errorName(err) });
+            return;
+        }
 
-        // 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.updateHostStatusEv(host_info.id, "active") catch {};
-        self.persist.resetHostFailuresEv(host_info.id) catch {};
+        // phase 4: reset status + failures via DbRequestQueue
+        const ResetHostReq = struct {
+            base: event_log_mod.DbRequest = .{ .callback = &execute },
+            host_id: u64,
 
-        try self.spawnWorker(host_info.id, hostname, host_info.last_seq);
-        log.info("added host {s} (id={d})", .{ hostname, host_info.id });
+            fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
+                const s: *@This() = @fieldParentPtr("base", b);
+                dp.updateHostStatus(s.host_id, "active") catch {};
+                dp.resetHostFailures(s.host_id) catch {};
+            }
+        };
+        var reset_req: ResetHostReq = .{ .host_id = db_req.host_id };
+        db_queue.push(&reset_req.base);
+        reset_req.base.wait(self.io, self.shutdown);
+
+        // phase 5: spawn worker (Evented)
+        try self.spawnWorker(db_req.host_id, hostname, db_req.last_seq);
+        log.info("added host {s} (id={d})", .{ hostname, db_req.host_id });
     }
 
     /// spawn a subscriber thread for a host
@@ -468,7 +539,24 @@
         const sub = try self.allocator.create(subscriber_mod.Subscriber);
         errdefer self.allocator.destroy(sub);
 
-        const account_count: u64 = self.persist.getEffectiveAccountCountEv(host_id) catch 0;
+        // get effective account count via DbRequestQueue
+        var account_count: u64 = 0;
+        if (self.db_queue) |db_queue| {
+            const GetCountReq = struct {
+                base: event_log_mod.DbRequest = .{ .callback = &execute },
+                hid: u64,
+                count: u64 = 0,
+
+                fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
+                    const s: *@This() = @fieldParentPtr("base", b);
+                    s.count = dp.getEffectiveAccountCount(s.hid);
+                }
+            };
+            var count_req: GetCountReq = .{ .hid = host_id };
+            db_queue.push(&count_req.base);
+            count_req.base.wait(self.io, self.shutdown);
+            account_count = count_req.count;
+        }
 
         sub.* = subscriber_mod.Subscriber.init(
             self.allocator,
@@ -532,24 +620,40 @@
 
     /// background fiber: load hosts from DB and spawn all workers.
     /// runs in background so HTTP server + probes come up immediately.
+    /// pullHosts runs on its own std.Thread in parallel (not gated).
     /// Go relay: ResubscribeAllHosts loops with 1ms sleep per host (goroutines).
     /// we batch-spawn with yields between batches to keep the event loop responsive
     /// for health checks and metrics during the initial TLS handshake ramp.
     fn spawnWorkers(self: *Slurper) void {
-        // pull hosts from seed relay first — idempotent (getOrCreateHost skips existing)
-        if (self.options.seed_host.len > 0) {
-            log.info("pulling hosts from {s}...", .{self.options.seed_host});
-            self.pullHosts() catch |err| {
-                log.warn("pullHosts from {s} failed: {s}", .{ self.options.seed_host, @errorName(err) });
-            };
-        } else {
-            log.info("no seed host configured, skipping bootstrap", .{});
-        }
-
-        const hosts = self.persist.listActiveHostsEv(self.allocator) catch |err| {
-            log.err("failed to load hosts: {s}", .{@errorName(err)});
+        const db_queue = self.db_queue orelse {
+            log.err("spawnWorkers: db_queue not set", .{});
             return;
         };
+
+        // load active hosts via DbRequestQueue
+        const ListActiveHostsReq = struct {
+            base: event_log_mod.DbRequest = .{ .callback = &execute },
+            alloc: Allocator,
+            result: ?[]event_log_mod.DiskPersist.Host = null,
+
+            fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void {
+                const s: *@This() = @fieldParentPtr("base", b);
+                s.result = dp.listActiveHosts(s.alloc) catch |e| {
+                    b.err = e;
+                    return;
+                };
+            }
+        };
+        var list_req: ListActiveHostsReq = .{ .alloc = self.allocator };
+        db_queue.push(&list_req.base);
+        list_req.base.wait(self.io, self.shutdown);
+
+        if (list_req.base.err != null or list_req.result == null) {
+            log.err("failed to load hosts: {s}", .{if (list_req.base.err) |e| @errorName(e) else "null result"});
+            return;
+        }
+
+        const hosts = list_req.result.?;
         defer {
             for (hosts) |h| {
                 self.allocator.free(h.hostname);
diff --git a/src/api/admin.zig b/src/api/admin.zig
--- a/src/api/admin.zig
+++ b/src/api/admin.zig
@@ -2,6 +2,9 @@
 //!
 //! all handlers require Bearer token auth against RELAY_ADMIN_PASSWORD.
 //! includes host blocking/unblocking, account bans, and backfill control.
+//!
+//! DB-accessing handlers use DbRequest + DbRequestQueue to route queries through
+//! pool_io (Threaded) workers.
 
 const std = @import("std");
 const Io = std.Io;
@@ -16,6 +19,8 @@
 const log = std.log.scoped(.relay);
 
 const HttpContext = router.HttpContext;
+const DbRequest = event_log_mod.DbRequest;
+const DiskPersist = event_log_mod.DiskPersist;
 
 /// check admin auth via headers, send error response if not authorized. returns true if authorized.
 pub fn checkAdmin(conn: *h.Conn, headers: ?*const websocket.Handshake.KeyValue) bool {
@@ -58,11 +63,50 @@
     defer parsed.deinit();
     const did = parsed.value.did;
 
-    // resolve DID → UID via Evented pool (skips DID cache which uses pool_io mutex)
-    const uid = ctx.persist.uidForDidEv(did) catch {
+    // resolve DID → UID via DbRequestQueue
+    const UidReq = struct {
+        base: DbRequest = .{ .callback = &execute },
+        did_buf: [256]u8 = undefined,
+        did_len: usize = 0,
+        uid: u64 = 0,
+
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base", b);
+            const d = self.did_buf[0..self.did_len];
+            // check database
+            if (dp.db.rowUnsafe("SELECT uid FROM account WHERE did = $1", .{d}) catch null) |row| {
+                var r = row;
+                defer r.deinit() catch {};
+                self.uid = @intCast(r.get(i64, 0));
+                return;
+            }
+            // create new account row
+            _ = dp.db.exec("INSERT INTO account (did) VALUES ($1) ON CONFLICT (did) DO NOTHING", .{d}) catch {
+                b.err = error.DatabaseError;
+                return;
+            };
+            var row = dp.db.rowUnsafe("SELECT uid FROM account WHERE did = $1", .{d}) catch {
+                b.err = error.DatabaseError;
+                return;
+            } orelse {
+                b.err = error.AccountCreationFailed;
+                return;
+            };
+            defer row.deinit() catch {};
+            self.uid = @intCast(row.get(i64, 0));
+        }
+    };
+    var uid_req: UidReq = .{};
+    const copy_len = @min(did.len, uid_req.did_buf.len);
+    @memcpy(uid_req.did_buf[0..copy_len], did[0..copy_len]);
+    uid_req.did_len = copy_len;
+    ctx.db_queue.push(&uid_req.base);
+    uid_req.base.wait(ctx.io, ctx.shutdown);
+
+    if (uid_req.base.err != null) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to resolve DID\"}");
         return;
-    };
+    }
 
     // remove from collection index so banned accounts don't appear in listReposByCollection
     ctx.collection_index.removeAll(did) catch |err| {
@@ -72,7 +116,7 @@
     // 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 };
+    var td: host_ops_mod.HostOp.Payload.Takedown = .{ .uid = uid_req.uid };
 
     if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| {
         defer ctx.persist.allocator.free(frame_bytes);
@@ -88,27 +132,46 @@
         .payload = .{ .takedown = td },
     });
 
-    log.info("admin: banned {s} (uid={d}), takedown enqueued", .{ did, uid });
+    log.info("admin: banned {s} (uid={d}), takedown enqueued", .{ did, uid_req.uid });
     h.respondJson(conn, .ok, "{\"success\":true}");
 }
 
 pub fn handleAdminListHosts(conn: *h.Conn, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void {
     if (!checkAdmin(conn, headers)) return;
 
-    const persist = ctx.persist;
-    const hosts = persist.listAllHostsEv(persist.allocator) catch {
+    // list all hosts via DbRequestQueue
+    const ListAllHostsReq = struct {
+        base: DbRequest = .{ .callback = &execute },
+        alloc: std.mem.Allocator,
+        result: ?[]event_log_mod.DiskPersist.Host = null,
+
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base", b);
+            self.result = dp.listAllHosts(self.alloc) catch |e| {
+                b.err = e;
+                return;
+            };
+        }
+    };
+    var list_req: ListAllHostsReq = .{ .alloc = ctx.persist.allocator };
+    ctx.db_queue.push(&list_req.base);
+    list_req.base.wait(ctx.io, ctx.shutdown);
+
+    if (list_req.base.err != null or list_req.result == null) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    };
-    defer {
-        for (hosts) |host| {
-            persist.allocator.free(host.hostname);
-            persist.allocator.free(host.status);
-        }
-        persist.allocator.free(hosts);
     }
 
-    var aw: Io.Writer.Allocating = .init(persist.allocator);
+    const hosts = list_req.result.?;
+    defer {
+        for (hosts) |host| {
+            ctx.persist.allocator.free(host.hostname);
+            ctx.persist.allocator.free(host.status);
+        }
+        ctx.persist.allocator.free(hosts);
+    }
+
+    var aw: Io.Writer.Allocating = .init(ctx.persist.allocator);
     defer aw.deinit();
     const w = &aw.writer;
 
@@ -140,56 +203,98 @@
     h.respondJson(conn, .ok, aw.written());
 }
 
-pub fn handleAdminBlockHost(conn: *h.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void {
+pub fn handleAdminBlockHost(conn: *h.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void {
     if (!checkAdmin(conn, headers)) return;
 
-    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
+    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}");
         return;
     };
     defer parsed.deinit();
 
-    const host_info = persist.getOrCreateHostEv(parsed.value.hostname) catch {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}");
-        return;
-    };
+    const BlockHostReq = struct {
+        base: DbRequest = .{ .callback = &execute },
+        hostname_buf: [256]u8 = undefined,
+        hostname_len: usize = 0,
+        host_id: u64 = 0,
 
-    persist.updateHostStatusEv(host_info.id, "blocked") catch {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}");
-        return;
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base", b);
+            const hn = self.hostname_buf[0..self.hostname_len];
+            const info = dp.getOrCreateHost(hn) catch |e| {
+                b.err = e;
+                return;
+            };
+            self.host_id = info.id;
+            dp.updateHostStatus(info.id, "blocked") catch |e| {
+                b.err = e;
+                return;
+            };
+        }
     };
+    var req: BlockHostReq = .{};
+    const copy_len = @min(parsed.value.hostname.len, req.hostname_buf.len);
+    @memcpy(req.hostname_buf[0..copy_len], parsed.value.hostname[0..copy_len]);
+    req.hostname_len = copy_len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
 
-    log.info("admin: blocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id });
+    if (req.base.err != null) {
+        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"operation failed\"}");
+        return;
+    }
+
+    log.info("admin: blocked host {s} (id={d})", .{ parsed.value.hostname, req.host_id });
     h.respondJson(conn, .ok, "{\"success\":true}");
 }
 
-pub fn handleAdminUnblockHost(conn: *h.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void {
+pub fn handleAdminUnblockHost(conn: *h.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void {
     if (!checkAdmin(conn, headers)) return;
 
-    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
+    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}");
         return;
     };
     defer parsed.deinit();
 
-    const host_info = persist.getOrCreateHostEv(parsed.value.hostname) catch {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}");
-        return;
-    };
+    const UnblockHostReq = struct {
+        base: DbRequest = .{ .callback = &execute },
+        hostname_buf: [256]u8 = undefined,
+        hostname_len: usize = 0,
+        host_id: u64 = 0,
 
-    persist.updateHostStatusEv(host_info.id, "active") catch {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}");
-        return;
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base", b);
+            const hn = self.hostname_buf[0..self.hostname_len];
+            const info = dp.getOrCreateHost(hn) catch |e| {
+                b.err = e;
+                return;
+            };
+            self.host_id = info.id;
+            dp.updateHostStatus(info.id, "active") catch |e| {
+                b.err = e;
+                return;
+            };
+            dp.resetHostFailures(info.id) catch {};
+        }
     };
-    persist.resetHostFailuresEv(host_info.id) catch {};
+    var req: UnblockHostReq = .{};
+    const copy_len = @min(parsed.value.hostname.len, req.hostname_buf.len);
+    @memcpy(req.hostname_buf[0..copy_len], parsed.value.hostname[0..copy_len]);
+    req.hostname_len = copy_len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
 
-    log.info("admin: unblocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id });
+    if (req.base.err != null) {
+        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"operation failed\"}");
+        return;
+    }
+
+    log.info("admin: unblocked host {s} (id={d})", .{ parsed.value.hostname, req.host_id });
     h.respondJson(conn, .ok, "{\"success\":true}");
 }
 
 /// set or clear the account_limit override for a host.
-/// POST {"host": "...", "account_limit": 100000}  — set override
-/// POST {"host": "...", "account_limit": null}     — clear override (revert to COUNT(*))
 pub fn handleAdminChangeLimits(conn: *h.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void {
     if (!checkAdmin(conn, headers)) return;
 
@@ -204,22 +309,47 @@
     };
     defer parsed.deinit();
 
-    const host_id = ctx.persist.getHostIdForHostnameEv(parsed.value.host) catch {
+    const ChangeLimitsReq = struct {
+        base: DbRequest = .{ .callback = &execute },
+        hostname_buf: [256]u8 = undefined,
+        hostname_len: usize = 0,
+        new_limit: ?u64,
+        host_id: ?u64 = null,
+        effective: u64 = 0,
+
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base", b);
+            const hn = self.hostname_buf[0..self.hostname_len];
+            self.host_id = dp.getHostIdForHostname(hn) catch |e| {
+                b.err = e;
+                return;
+            };
+            const hid = self.host_id orelse return;
+            dp.setHostAccountLimit(hid, self.new_limit) catch |e| {
+                b.err = e;
+                return;
+            };
+            self.effective = if (self.new_limit) |l| l else dp.getHostAccountCount(hid);
+        }
+    };
+    var req: ChangeLimitsReq = .{ .new_limit = parsed.value.account_limit };
+    const copy_len = @min(parsed.value.host.len, req.hostname_buf.len);
+    @memcpy(req.hostname_buf[0..copy_len], parsed.value.host[0..copy_len]);
+    req.hostname_len = copy_len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"database error\"}");
         return;
-    } orelse {
+    }
+    const host_id = req.host_id orelse {
         h.respondJson(conn, .not_found, "{\"error\":\"host not found\"}");
         return;
     };
 
-    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.getHostAccountCountEv(host_id) catch 0;
-    ctx.slurper.updateHostLimits(host_id, effective);
+    ctx.slurper.updateHostLimits(host_id, req.effective);
 
     if (parsed.value.account_limit) |limit| {
         log.info("admin: set account_limit for {s} (id={d}): {d}", .{ parsed.value.host, host_id, limit });
@@ -337,7 +467,6 @@
 // --- protocol helpers (used only by handleBan) ---
 
 /// build a CBOR #account frame for a takedown event.
-/// header: {op: 1, t: "#account"}, payload: {seq: 0, did: "...", time: "...", active: false, status: "takendown"}
 fn buildAccountFrame(allocator: std.mem.Allocator, did: []const u8) ?[]const u8 {
     const zat = @import("zat");
     const cbor = zat.cbor;
@@ -378,7 +507,6 @@
     return frame;
 }
 
-/// format current UTC time as ISO 8601 (YYYY-MM-DDTHH:MM:SSZ)
 fn formatTimestamp(buf: *[24]u8) []const u8 {
     var tp: std.c.timespec = undefined;
     _ = std.c.clock_gettime(.REALTIME, &tp);
diff --git a/src/api/router.zig b/src/api/router.zig
--- a/src/api/router.zig
+++ b/src/api/router.zig
@@ -32,6 +32,9 @@
     validator: *validator_mod.Validator,
     host_ops: *host_ops_mod.HostOpsQueue,
     pool_io: Io,
+    io: Io,
+    db_queue: *event_log_mod.DbRequestQueue,
+    shutdown: *std.atomic.Value(bool),
 };
 
 /// top-level HTTP request router — installed as bc.http_fallback
@@ -74,19 +77,19 @@
         const body = broadcaster.formatStatsResponse(ctx.stats, &stats_buf, ctx.bc.io);
         h.respondJson(conn, .ok, body);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listRepos")) {
-        xrpc.handleListRepos(conn, query, ctx.persist);
+        xrpc.handleListRepos(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) {
-        xrpc.handleGetRepoStatus(conn, query, ctx.persist);
+        xrpc.handleGetRepoStatus(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepo")) {
-        xrpc.handleGetRepo(conn, query, ctx.persist);
+        xrpc.handleGetRepo(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) {
-        xrpc.handleGetLatestCommit(conn, query, ctx.persist);
+        xrpc.handleGetLatestCommit(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) {
         xrpc.handleListReposByCollection(conn, query, ctx.collection_index);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listHosts")) {
-        xrpc.handleListHosts(conn, query, ctx.persist);
+        xrpc.handleListHosts(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getHostStatus")) {
-        xrpc.handleGetHostStatus(conn, query, ctx.persist);
+        xrpc.handleGetHostStatus(conn, query, ctx);
     } else if (std.mem.eql(u8, path, "/admin/hosts")) {
         admin.handleAdminListHosts(conn, headers, ctx);
     } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) {
@@ -126,11 +129,11 @@
     if (std.mem.eql(u8, path, "/admin/repo/ban")) {
         admin.handleBan(conn, body, headers, ctx);
     } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.requestCrawl")) {
-        xrpc.handleRequestCrawl(conn, body, ctx.slurper);
+        xrpc.handleRequestCrawl(conn, body, ctx);
     } else if (std.mem.eql(u8, path, "/admin/hosts/block")) {
-        admin.handleAdminBlockHost(conn, body, headers, ctx.persist);
+        admin.handleAdminBlockHost(conn, body, headers, ctx);
     } else if (std.mem.eql(u8, path, "/admin/hosts/unblock")) {
-        admin.handleAdminUnblockHost(conn, body, headers, ctx.persist);
+        admin.handleAdminUnblockHost(conn, body, headers, ctx);
     } else if (std.mem.eql(u8, path, "/admin/hosts/changeLimits")) {
         admin.handleAdminChangeLimits(conn, body, headers, ctx);
     } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) {
diff --git a/src/api/xrpc.zig b/src/api/xrpc.zig
--- a/src/api/xrpc.zig
+++ b/src/api/xrpc.zig
@@ -3,17 +3,100 @@
 //! implements com.atproto.sync.* lexicon endpoints:
 //!   listRepos, getRepo, getRepoStatus, getLatestCommit, listReposByCollection,
 //!   listHosts, getHostStatus, requestCrawl
+//!
+//! DB-accessing handlers use DbRequest + DbRequestQueue to route queries through
+//! pool_io (Threaded) workers, avoiding the broken Evented pg.Pool.
 
 const std = @import("std");
 const Io = std.Io;
+const pg = @import("pg");
 const h = @import("http.zig");
+const router = @import("router.zig");
 const event_log_mod = @import("../event_log.zig");
 const collection_index_mod = @import("../collection_index.zig");
 const slurper_mod = @import("../slurper.zig");
 
+const Allocator = std.mem.Allocator;
+const HttpContext = router.HttpContext;
+const DbRequest = event_log_mod.DbRequest;
+const DiskPersist = event_log_mod.DiskPersist;
 const log = std.log.scoped(.relay);
 
-pub fn handleListRepos(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
+// --- listRepos ---
+
+const ListReposReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    cursor_val: i64,
+    limit: i64,
+    json_buf: [65536]u8 = undefined,
+    json_len: usize = 0,
+    db_err: bool = false,
+
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        var w: Io.Writer = .fixed(&self.json_buf);
+
+        var result = dp.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
+        , .{ self.cursor_val, self.limit }) catch {
+            self.db_err = true;
+            return;
+        };
+        defer result.deinit();
+
+        var count: i64 = 0;
+        var last_uid: i64 = 0;
+
+        w.writeAll("{\"repos\":[") catch return;
+
+        while (result.nextUnsafe() catch null) |row| {
+            if (count > 0) w.writeByte(',') catch return;
+
+            const uid = row.get(i64, 0);
+            const did = row.get([]const u8, 1);
+            const local_status = row.get([]const u8, 2);
+            const upstream_status = row.get([]const u8, 3);
+            const rev = row.get([]const u8, 4);
+            const head = row.get([]const u8, 5);
+
+            const local_ok = std.mem.eql(u8, local_status, "active");
+            const upstream_ok = std.mem.eql(u8, upstream_status, "active");
+            const active = local_ok and upstream_ok;
+            const status = if (!local_ok) local_status else upstream_status;
+
+            w.writeAll("{\"did\":\"") catch return;
+            w.writeAll(did) catch return;
+            w.writeAll("\",\"head\":\"") catch return;
+            w.writeAll(head) catch return;
+            w.writeAll("\",\"rev\":\"") catch return;
+            w.writeAll(rev) catch return;
+            w.writeAll("\"") catch return;
+
+            if (active) {
+                w.writeAll(",\"active\":true") catch return;
+            } else {
+                w.writeAll(",\"active\":false,\"status\":\"") catch return;
+                w.writeAll(status) catch return;
+                w.writeAll("\"") catch return;
+            }
+
+            w.writeByte('}') catch return;
+            last_uid = uid;
+            count += 1;
+        }
+
+        w.writeByte(']') catch return;
+        if (count >= self.limit and count >= 2) {
+            w.print(",\"cursor\":\"{d}\"", .{last_uid}) catch return;
+        }
+        w.writeByte('}') catch return;
+        self.json_len = w.end;
+    }
+};
+
+pub fn handleListRepos(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
     const cursor_str = h.queryParam(query, "cursor") orelse "0";
     const limit_str = h.queryParam(query, "limit") orelse "500";
 
@@ -35,57 +118,60 @@
         return;
     }
 
-    const ev_db = persist.ensureEvDb() catch {
+    var req: ListReposReq = .{ .cursor_val = cursor_val, .limit = limit };
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
         return;
-    };
-
-    // query accounts with repo state, paginated by UID
-    // includes both local status and upstream_status for combined active check
-    var result = 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
-    , .{ cursor_val, limit }) catch {
+    }
+    if (req.db_err) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    };
-    defer result.deinit();
+    }
 
-    // build JSON response into a buffer
-    var buf: [65536]u8 = undefined;
-    var w: Io.Writer = .fixed(&buf);
+    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
+}
 
-    var count: i64 = 0;
-    var last_uid: i64 = 0;
+// --- getRepoStatus ---
 
-    w.writeAll("{\"repos\":[") catch return;
+const GetRepoStatusReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    did_buf: [256]u8 = undefined,
+    did_len: usize = 0,
+    json_buf: [4096]u8 = undefined,
+    json_len: usize = 0,
+    not_found: bool = false,
+    db_err: bool = false,
 
-    while (result.nextUnsafe() catch null) |row| {
-        if (count > 0) w.writeByte(',') catch return;
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        const did = self.did_buf[0..self.did_len];
 
-        const uid = row.get(i64, 0);
-        const did = row.get([]const u8, 1);
-        const local_status = row.get([]const u8, 2);
-        const upstream_status = row.get([]const u8, 3);
-        const rev = row.get([]const u8, 4);
-        const head = row.get([]const u8, 5);
+        var row = (dp.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 {
+            self.db_err = true;
+            return;
+        }) orelse {
+            self.not_found = true;
+            return;
+        };
+        defer row.deinit() catch {};
 
-        // Go relay: Account.IsActive() — both local AND upstream must be active
+        const local_status = row.get([]const u8, 1);
+        const upstream_status = row.get([]const u8, 2);
+        const rev = row.get([]const u8, 3);
         const local_ok = std.mem.eql(u8, local_status, "active");
         const upstream_ok = std.mem.eql(u8, upstream_status, "active");
         const active = local_ok and upstream_ok;
-        // Go relay: Account.AccountStatus() — local takes priority
         const status = if (!local_ok) local_status else upstream_status;
 
+        var w: Io.Writer = .fixed(&self.json_buf);
         w.writeAll("{\"did\":\"") catch return;
         w.writeAll(did) catch return;
-        w.writeAll("\"") catch return;
-
-        w.writeAll(",\"head\":\"") catch return;
-        w.writeAll(head) catch return;
-        w.writeAll("\",\"rev\":\"") catch return;
-        w.writeAll(rev) catch return;
         w.writeAll("\"") catch return;
 
         if (active) {
@@ -96,196 +182,222 @@
             w.writeAll("\"") catch return;
         }
 
+        if (rev.len > 0) {
+            w.writeAll(",\"rev\":\"") catch return;
+            w.writeAll(rev) catch return;
+            w.writeAll("\"") catch return;
+        }
+
         w.writeByte('}') catch return;
-        last_uid = uid;
-        count += 1;
+        self.json_len = w.end;
     }
+};
 
-    w.writeByte(']') catch return;
-
-    // include cursor if we got a full page
-    if (count >= limit and count >= 2) {
-        w.print(",\"cursor\":\"{d}\"", .{last_uid}) catch return;
-    }
-
-    w.writeByte('}') catch return;
-
-    h.respondJson(conn, .ok, w.buffered());
-}
-
-pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
+pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
     var did_buf: [256]u8 = undefined;
     const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
         return;
     };
-
-    // basic DID syntax check
     if (!std.mem.startsWith(u8, did, "did:")) {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
         return;
     }
 
-    const ev_db = persist.ensureEvDb() catch {
+    var req: GetRepoStatusReq = .{};
+    @memcpy(req.did_buf[0..did.len], did);
+    req.did_len = did.len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
         return;
-    };
-
-    // look up account (includes both local and upstream status)
-    var row = (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 {
+    }
+    if (req.db_err) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    }) orelse {
+    }
+    if (req.not_found) {
         h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
         return;
-    };
-    defer row.deinit() catch {};
-
-    const local_status = row.get([]const u8, 1);
-    const upstream_status = row.get([]const u8, 2);
-    const rev = row.get([]const u8, 3);
-    // Go relay: Account.IsActive() / AccountStatus()
-    const local_ok = std.mem.eql(u8, local_status, "active");
-    const upstream_ok = std.mem.eql(u8, upstream_status, "active");
-    const active = local_ok and upstream_ok;
-    const status = if (!local_ok) local_status else upstream_status;
-
-    var buf: [4096]u8 = undefined;
-    var w: Io.Writer = .fixed(&buf);
-
-    w.writeAll("{\"did\":\"") catch return;
-    w.writeAll(did) catch return;
-    w.writeAll("\"") catch return;
-
-    if (active) {
-        w.writeAll(",\"active\":true") catch return;
-    } else {
-        w.writeAll(",\"active\":false,\"status\":\"") catch return;
-        w.writeAll(status) catch return;
-        w.writeAll("\"") catch return;
     }
 
-    if (rev.len > 0) {
-        w.writeAll(",\"rev\":\"") catch return;
+    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
+}
+
+// --- getRepo ---
+
+const GetRepoReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    did_buf: [256]u8 = undefined,
+    did_len: usize = 0,
+    url_buf: [512]u8 = undefined,
+    url_len: usize = 0,
+    not_found: bool = false,
+    db_err: bool = false,
+
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        const did = self.did_buf[0..self.did_len];
+
+        var row = (dp.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 {
+            self.db_err = true;
+            return;
+        }) orelse {
+            self.not_found = true;
+            return;
+        };
+        defer row.deinit() catch {};
+
+        const hostname = row.get([]const u8, 0);
+        const url = std.fmt.bufPrint(&self.url_buf, "https://{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ hostname, did }) catch return;
+        self.url_len = url.len;
+    }
+};
+
+pub fn handleGetRepo(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
+    var did_buf: [256]u8 = undefined;
+    const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
+        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
+        return;
+    };
+    if (!std.mem.startsWith(u8, did, "did:")) {
+        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
+        return;
+    }
+
+    var req: GetRepoReq = .{};
+    @memcpy(req.did_buf[0..did.len], did);
+    req.did_len = did.len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
+        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
+        return;
+    }
+    if (req.db_err) {
+        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
+        return;
+    }
+    if (req.not_found) {
+        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
+        return;
+    }
+
+    h.respondRedirect(conn, req.url_buf[0..req.url_len]);
+}
+
+// --- getLatestCommit ---
+
+const GetLatestCommitReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    did_buf: [256]u8 = undefined,
+    did_len: usize = 0,
+    json_buf: [4096]u8 = undefined,
+    json_len: usize = 0,
+    not_found: bool = false,
+    db_err: bool = false,
+    status_err: ?[]const u8 = null,
+
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        const did = self.did_buf[0..self.did_len];
+
+        var row = (dp.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 {
+            self.db_err = true;
+            return;
+        }) orelse {
+            self.not_found = true;
+            return;
+        };
+        defer row.deinit() catch {};
+
+        const local_status = row.get([]const u8, 0);
+        const upstream_status = row.get([]const u8, 1);
+        const rev = row.get([]const u8, 2);
+        const cid = row.get([]const u8, 3);
+
+        const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status;
+
+        if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) {
+            self.status_err = "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}";
+            return;
+        } else if (std.mem.eql(u8, status, "deactivated")) {
+            self.status_err = "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}";
+            return;
+        } else if (std.mem.eql(u8, status, "deleted")) {
+            self.status_err = "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}";
+            return;
+        } else if (!std.mem.eql(u8, status, "active")) {
+            self.status_err = "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}";
+            return;
+        }
+
+        if (rev.len == 0 or cid.len == 0) {
+            self.not_found = true;
+            return;
+        }
+
+        var w: Io.Writer = .fixed(&self.json_buf);
+        w.writeAll("{\"cid\":\"") catch return;
+        w.writeAll(cid) catch return;
+        w.writeAll("\",\"rev\":\"") catch return;
         w.writeAll(rev) catch return;
-        w.writeAll("\"") catch return;
+        w.writeAll("\"}") catch return;
+        self.json_len = w.end;
     }
+};
 
-    w.writeByte('}') catch return;
-    h.respondJson(conn, .ok, w.buffered());
-}
-
-pub fn handleGetRepo(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
+pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
     var did_buf: [256]u8 = undefined;
     const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
         return;
     };
-
     if (!std.mem.startsWith(u8, did, "did:")) {
         h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
         return;
     }
 
-    const ev_db = persist.ensureEvDb() catch {
+    var req: GetLatestCommitReq = .{};
+    @memcpy(req.did_buf[0..did.len], did);
+    req.did_len = did.len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
         return;
-    };
-
-    // look up the PDS hostname for this account
-    var row = (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 {
+    }
+    if (req.db_err) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    }) orelse {
-        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
+    }
+    if (req.status_err) |err_body| {
+        h.respondJson(conn, .forbidden, err_body);
         return;
-    };
-    defer row.deinit() catch {};
+    }
+    if (req.not_found) {
+        if (req.json_len == 0) {
+            h.respondJson(conn, .not_found, "{\"error\":\"RepoNotSynchronized\",\"message\":\"relay has no repo data for this account\"}");
+        } else {
+            h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
+        }
+        return;
+    }
 
-    const hostname = row.get([]const u8, 0);
-
-    // build redirect URL: https://{hostname}/xrpc/com.atproto.sync.getRepo?did={did}
-    var url_buf: [512]u8 = undefined;
-    const url = std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ hostname, did }) catch return;
-
-    h.respondRedirect(conn, url);
+    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
 }
 
-pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
-    var did_buf: [256]u8 = undefined;
-    const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
-        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
-        return;
-    };
-
-    if (!std.mem.startsWith(u8, did, "did:")) {
-        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
-        return;
-    }
-
-    const ev_db = persist.ensureEvDb() catch {
-        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
-        return;
-    };
-
-    // look up account + repo state (includes both local and upstream status)
-    var row = (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 {
-        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
-        return;
-    }) orelse {
-        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
-        return;
-    };
-    defer row.deinit() catch {};
-
-    const local_status = row.get([]const u8, 0);
-    const upstream_status = row.get([]const u8, 1);
-    const rev = row.get([]const u8, 2);
-    const cid = row.get([]const u8, 3);
-
-    // combined status: local takes priority (Go relay: AccountStatus())
-    const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status;
-
-    // check account status (match Go relay behavior)
-    if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) {
-        h.respondJson(conn, .forbidden, "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}");
-        return;
-    } else if (std.mem.eql(u8, status, "deactivated")) {
-        h.respondJson(conn, .forbidden, "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}");
-        return;
-    } else if (std.mem.eql(u8, status, "deleted")) {
-        h.respondJson(conn, .forbidden, "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}");
-        return;
-    } else if (!std.mem.eql(u8, status, "active")) {
-        h.respondJson(conn, .forbidden, "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}");
-        return;
-    }
-
-    if (rev.len == 0 or cid.len == 0) {
-        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotSynchronized\",\"message\":\"relay has no repo data for this account\"}");
-        return;
-    }
-
-    var buf: [4096]u8 = undefined;
-    var w: Io.Writer = .fixed(&buf);
-
-    w.writeAll("{\"cid\":\"") catch return;
-    w.writeAll(cid) catch return;
-    w.writeAll("\",\"rev\":\"") catch return;
-    w.writeAll(rev) catch return;
-    w.writeAll("\"}") catch return;
-
-    h.respondJson(conn, .ok, w.buffered());
-}
+// --- listReposByCollection (no DB, uses collection index directly) ---
 
 pub fn handleListReposByCollection(conn: *h.Conn, query: []const u8, ci: *collection_index_mod.CollectionIndex) void {
     const collection = h.queryParam(query, "collection") orelse {
@@ -311,14 +423,12 @@
     var cursor_buf: [256]u8 = undefined;
     const cursor_did = h.queryParamDecoded(query, "cursor", &cursor_buf);
 
-    // scan collection index
     var did_buf: [65536]u8 = undefined;
     const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}");
         return;
     };
 
-    // build JSON response
     var buf: [65536]u8 = undefined;
     var w: Io.Writer = .fixed(&buf);
 
@@ -343,7 +453,64 @@
     h.respondJson(conn, .ok, w.buffered());
 }
 
-pub fn handleListHosts(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
+// --- listHosts ---
+
+const ListHostsReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    cursor_val: i64,
+    limit: i64,
+    json_buf: [65536]u8 = undefined,
+    json_len: usize = 0,
+    db_err: bool = false,
+
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        var w: Io.Writer = .fixed(&self.json_buf);
+
+        var result = dp.db.query(
+            "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2",
+            .{ self.cursor_val, self.limit },
+        ) catch {
+            self.db_err = true;
+            return;
+        };
+        defer result.deinit();
+
+        var count: i64 = 0;
+        var last_id: i64 = 0;
+
+        w.writeAll("{\"hosts\":[") catch return;
+
+        while (result.nextUnsafe() catch null) |row| {
+            if (count > 0) w.writeByte(',') catch return;
+
+            const id = row.get(i64, 0);
+            const hostname = row.get([]const u8, 1);
+            const status = row.get([]const u8, 2);
+            const seq = row.get(i64, 3);
+
+            w.writeAll("{\"hostname\":\"") catch return;
+            w.writeAll(hostname) catch return;
+            w.writeAll("\"") catch return;
+            w.print(",\"seq\":{d}", .{seq}) catch return;
+            w.writeAll(",\"status\":\"") catch return;
+            w.writeAll(status) catch return;
+            w.writeAll("\"}") catch return;
+
+            last_id = id;
+            count += 1;
+        }
+
+        w.writeByte(']') catch return;
+        if (count >= self.limit and count > 1) {
+            w.print(",\"cursor\":\"{d}\"", .{last_id}) catch return;
+        }
+        w.writeByte('}') catch return;
+        self.json_len = w.end;
+    }
+};
+
+pub fn handleListHosts(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
     const cursor_str = h.queryParam(query, "cursor") orelse "0";
     const limit_str = h.queryParam(query, "limit") orelse "200";
 
@@ -365,129 +532,123 @@
         return;
     }
 
-    const ev_db = persist.ensureEvDb() catch {
+    var req: ListHostsReq = .{ .cursor_val = cursor_val, .limit = limit };
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
         return;
-    };
-
-    var result = 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 {
+    }
+    if (req.db_err) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    };
-    defer result.deinit();
+    }
 
-    var buf: [65536]u8 = undefined;
-    var w: Io.Writer = .fixed(&buf);
+    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
+}
 
-    var count: i64 = 0;
-    var last_id: i64 = 0;
+// --- getHostStatus ---
 
-    w.writeAll("{\"hosts\":[") catch return;
+const GetHostStatusReq = struct {
+    base: DbRequest = .{ .callback = &execute },
+    hostname_buf: [256]u8 = undefined,
+    hostname_len: usize = 0,
+    json_buf: [4096]u8 = undefined,
+    json_len: usize = 0,
+    not_found: bool = false,
+    db_err: bool = false,
 
-    while (result.nextUnsafe() catch null) |row| {
-        if (count > 0) w.writeByte(',') catch return;
+    fn execute(b: *DbRequest, dp: *DiskPersist) void {
+        const self: *@This() = @fieldParentPtr("base", b);
+        const hostname = self.hostname_buf[0..self.hostname_len];
 
-        const id = row.get(i64, 0);
-        const hostname = row.get([]const u8, 1);
-        const status = row.get([]const u8, 2);
+        var row = (dp.db.rowUnsafe(
+            "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1",
+            .{hostname},
+        ) catch {
+            self.db_err = true;
+            return;
+        }) orelse {
+            self.not_found = true;
+            return;
+        };
+        defer row.deinit() catch {};
+
+        const host_id = row.get(i64, 0);
+        const host_name = row.get([]const u8, 1);
+        const raw_status = row.get([]const u8, 2);
         const seq = row.get(i64, 3);
 
+        const status = if (std.mem.eql(u8, raw_status, "blocked"))
+            "banned"
+        else if (std.mem.eql(u8, raw_status, "exhausted"))
+            "offline"
+        else
+            raw_status;
+
+        // count accounts on this host
+        const account_count: i64 = if (dp.db.rowUnsafe(
+            "SELECT COUNT(*) FROM account WHERE host_id = $1",
+            .{host_id},
+        ) catch null) |cnt_row| blk: {
+            var r = cnt_row;
+            defer r.deinit() catch {};
+            break :blk r.get(i64, 0);
+        } else 0;
+
+        var w: Io.Writer = .fixed(&self.json_buf);
         w.writeAll("{\"hostname\":\"") catch return;
-        w.writeAll(hostname) catch return;
+        w.writeAll(host_name) catch return;
         w.writeAll("\"") catch return;
-        w.print(",\"seq\":{d}", .{seq}) catch return;
+        w.print(",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return;
         w.writeAll(",\"status\":\"") catch return;
         w.writeAll(status) catch return;
         w.writeAll("\"}") catch return;
-
-        last_id = id;
-        count += 1;
+        self.json_len = w.end;
     }
+};
 
-    w.writeByte(']') catch return;
-
-    if (count >= limit and count > 1) {
-        w.print(",\"cursor\":\"{d}\"", .{last_id}) catch return;
-    }
-
-    w.writeByte('}') catch return;
-    h.respondJson(conn, .ok, w.buffered());
-}
-
-pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void {
+pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
     var hostname_buf: [256]u8 = undefined;
     const hostname = h.queryParamDecoded(query, "hostname", &hostname_buf) orelse {
         h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"hostname parameter required\"}");
         return;
     };
 
-    const ev_db = persist.ensureEvDb() catch {
+    var req: GetHostStatusReq = .{};
+    @memcpy(req.hostname_buf[0..hostname.len], hostname);
+    req.hostname_len = hostname.len;
+    ctx.db_queue.push(&req.base);
+    req.base.wait(ctx.io, ctx.shutdown);
+
+    if (req.base.err != null) {
         h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
         return;
-    };
-
-    // look up host
-    var row = (ev_db.rowUnsafe(
-        "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1",
-        .{hostname},
-    ) catch {
+    }
+    if (req.db_err) {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
         return;
-    }) orelse {
+    }
+    if (req.not_found) {
         h.respondJson(conn, .not_found, "{\"error\":\"HostNotFound\",\"message\":\"host not found\"}");
         return;
-    };
-    defer row.deinit() catch {};
+    }
 
-    const host_id = row.get(i64, 0);
-    const host_name = row.get([]const u8, 1);
-    const raw_status = row.get([]const u8, 2);
-    const seq = row.get(i64, 3);
-
-    // map internal status to lexicon hostStatus values
-    const status = if (std.mem.eql(u8, raw_status, "blocked"))
-        "banned"
-    else if (std.mem.eql(u8, raw_status, "exhausted"))
-        "offline"
-    else
-        raw_status; // active, idle pass through
-
-    // count accounts on this host
-    const account_count: i64 = if (ev_db.rowUnsafe(
-        "SELECT COUNT(*) FROM account WHERE host_id = $1",
-        .{host_id},
-    ) catch null) |cnt_row| blk: {
-        var r = cnt_row;
-        defer r.deinit() catch {};
-        break :blk r.get(i64, 0);
-    } else 0;
-
-    var buf: [4096]u8 = undefined;
-    var w: Io.Writer = .fixed(&buf);
-
-    w.writeAll("{\"hostname\":\"") catch return;
-    w.writeAll(host_name) catch return;
-    w.writeAll("\"") catch return;
-    w.print(",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return;
-    w.writeAll(",\"status\":\"") catch return;
-    w.writeAll(status) catch return;
-    w.writeAll("\"}") catch return;
-
-    h.respondJson(conn, .ok, w.buffered());
+    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
 }
 
-pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, slurper: *slurper_mod.Slurper) void {
-    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, slurper.allocator, body, .{ .ignore_unknown_fields = true }) catch {
+// --- requestCrawl ---
+
+pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, ctx: *HttpContext) void {
+    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
         h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"invalid JSON, expected {\\\"hostname\\\":\\\"...\\\"}\"}");
         return;
     };
     defer parsed.deinit();
 
-    // fast validation: hostname format (Go relay does this synchronously in handler)
-    const hostname = slurper_mod.validateHostname(slurper.allocator, parsed.value.hostname) catch |err| {
+    // fast validation: hostname format
+    const hostname = slurper_mod.validateHostname(ctx.persist.allocator, parsed.value.hostname) catch |err| {
         log.warn("requestCrawl rejected '{s}': {s}", .{ parsed.value.hostname, @errorName(err) });
         h.respondJson(conn, .bad_request, switch (err) {
             error.EmptyHostname => "{\"error\":\"InvalidRequest\",\"message\":\"empty hostname\"}",
@@ -501,17 +662,35 @@
         });
         return;
     };
-    defer slurper.allocator.free(hostname);
+    defer ctx.persist.allocator.free(hostname);
 
-    // fast validation: domain ban check (Evented fiber — use Ev pool)
-    if (slurper.persist.isDomainBannedEv(hostname) catch false) {
+    // domain ban check via DbRequestQueue
+    const DomainBanReq = struct {
+        base_req: DbRequest = .{ .callback = &execute },
+        hostname_buf: [256]u8 = undefined,
+        hostname_len: usize = 0,
+        banned: bool = false,
+
+        fn execute(b: *DbRequest, dp: *DiskPersist) void {
+            const self: *@This() = @fieldParentPtr("base_req", b);
+            self.banned = dp.isDomainBanned(self.hostname_buf[0..self.hostname_len]);
+        }
+    };
+    var ban_req: DomainBanReq = .{};
+    const copy_len = @min(hostname.len, ban_req.hostname_buf.len);
+    @memcpy(ban_req.hostname_buf[0..copy_len], hostname[0..copy_len]);
+    ban_req.hostname_len = copy_len;
+    ctx.db_queue.push(&ban_req.base_req);
+    ban_req.base_req.wait(ctx.io, ctx.shutdown);
+
+    if (ban_req.banned) {
         log.warn("requestCrawl rejected '{s}': domain banned", .{hostname});
         h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"domain is banned\"}");
         return;
     }
 
-    // enqueue for async processing (describeServer check happens in crawl processor)
-    slurper.addCrawlRequest(hostname) catch {
+    // enqueue for async processing
+    ctx.slurper.addCrawlRequest(hostname) catch {
         h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to store crawl request\"}");
         return;
     };