From 439c678a910d06bc64776017a7ba9db593be0d14 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sat, 04 Apr 2026 04:27:30 +0000 Subject: [PATCH] fix startup deadlock: run resyncer on plain thread, not Evented fiber the previous fix (6674812) correctly identified that Evented Io.Mutex from a plain thread causes SIGSEGV, but the fix used Threaded futex from within an Evented fiber. Threaded futexWait blocks the Uring OS thread, preventing it from processing io_uring completions for other fibers — deadlocking the event loop during CA bundle scan. fix: the resyncer now runs entirely on pool_io (Threaded) via a plain std.Thread. no Evented io involvement at all. this is correct because: - enqueue() from frame workers: Threaded futex on plain thread ✓ - dequeue() in worker: Threaded futex on plain thread ✓ - HTTP client: blocking I/O on plain thread ✓ the fundamental constraint: Io.Mutex cannot be shared across Io types. Threaded futex on Evented fiber → blocks Uring thread → deadlock. Evented futex on plain thread → NULL threadlocal → SIGSEGV. Co-Authored-By: Claude Opus 4.6 --- src/main.zig | 8 +++++--- src/resync.zig | 54 ++++++++++++++++++++++++------------------------------ 2 file(s) changed, 29 insertion(s)(+), 33 deletion(s)(-) diff --git a/src/main.zig b/src/main.zig --- a/src/main.zig +++ b/src/main.zig @@ -251,9 +251,11 @@ var cleaner = cleaner_mod.Cleaner.init(allocator, io, &ci, dp.db); // init resyncer (updates collection index on #sync events) - // queue_io = pool_io: enqueue() is called from frame worker threads (plain std.Thread), - // so mutex/cond operations must use Threaded io, not Evented. - var resyncer = resync_mod.Resyncer.init(allocator, io, pool_io, &ci); + // runs entirely on pool_io (Threaded) — enqueue() is called from frame worker + // threads, and the background worker is a plain std.Thread. MUST NOT use + // Evented io (Threaded futex on Evented fiber blocks Uring thread → deadlock; + // Evented futex on plain thread → NULL threadlocal → SIGSEGV). + var resyncer = resync_mod.Resyncer.init(allocator, pool_io, &ci); try resyncer.start(); defer resyncer.deinit(); diff --git a/src/resync.zig b/src/resync.zig --- a/src/resync.zig +++ b/src/resync.zig @@ -5,7 +5,10 @@ //! describeRepo from the PDS to get the current collection list, then //! replaces the index entries for that DID. //! -//! modeled on cleaner.zig — bounded queue, single background worker thread. +//! runs entirely on pool_io (Threaded) — enqueue() is called from frame worker +//! threads (plain std.Thread), and the background worker is also a plain thread. +//! MUST NOT use Evented io: Threaded futex on an Evented fiber blocks the Uring +//! thread, and Evented futex on a plain thread dereferences a NULL threadlocal. const std = @import("std"); const Io = std.Io; @@ -34,11 +37,9 @@ pub const Resyncer = struct { allocator: Allocator, + /// pool_io (Threaded) — used for all synchronization AND the HTTP client. + /// this struct must never touch the Evented io. io: Io, - /// pool_io (Threaded) for queue synchronization — enqueue() is called from - /// frame worker threads (plain std.Thread), so mutex/cond ops must use an Io - /// whose futexWait/futexWake work outside of Io.Uring fibers. - queue_io: Io, collection_index: *collection_index_mod.CollectionIndex, // bounded ring buffer queue @@ -50,7 +51,7 @@ cond: Io.Condition, running: std.atomic.Value(bool), - future: ?Io.Future(void), + thread: ?std.Thread, // stats processed: std.atomic.Value(u64), @@ -60,13 +61,11 @@ pub fn init( allocator: Allocator, io: Io, - queue_io: Io, collection_index: *collection_index_mod.CollectionIndex, ) Resyncer { return .{ .allocator = allocator, .io = io, - .queue_io = queue_io, .collection_index = collection_index, .queue = undefined, .head = 0, @@ -75,7 +74,7 @@ .mutex = Io.Mutex.init, .cond = Io.Condition.init, .running = .{ .raw = false }, - .future = null, + .thread = null, .processed = .{ .raw = 0 }, .failed = .{ .raw = 0 }, .dropped = .{ .raw = 0 }, @@ -85,16 +84,16 @@ /// start the background worker thread. pub fn start(self: *Resyncer) !void { self.running.store(true, .release); - self.future = try self.io.concurrent(run, .{self}); + self.thread = try std.Thread.spawn(.{}, run, .{self}); } /// enqueue a DID for resync. non-blocking, drops if queue full or inputs too long. - /// called from frame worker threads (plain std.Thread) — uses queue_io (Threaded). + /// called from frame worker threads (plain std.Thread). pub fn enqueue(self: *Resyncer, did: []const u8, hostname: []const u8) void { if (did.len == 0 or did.len > 128 or hostname.len == 0 or hostname.len > 256) return; - self.mutex.lockUncancelable(self.queue_io); - defer self.mutex.unlock(self.queue_io); + self.mutex.lockUncancelable(self.io); + defer self.mutex.unlock(self.io); if (self.len >= queue_capacity) { _ = self.dropped.fetchAdd(1, .monotonic); @@ -113,18 +112,16 @@ self.queue[self.tail] = item; self.tail = (self.tail + 1) % queue_capacity; self.len += 1; - self.cond.signal(self.queue_io); + self.cond.signal(self.io); } /// dequeue one item. blocks until available or shutdown. - /// runs in the Evented fiber (run()), but uses queue_io for mutex/cond - /// since the mutex is shared with enqueue() which runs on plain threads. fn dequeue(self: *Resyncer) ?ResyncItem { - self.mutex.lockUncancelable(self.queue_io); - defer self.mutex.unlock(self.queue_io); + self.mutex.lockUncancelable(self.io); + defer self.mutex.unlock(self.io); while (self.len == 0 and self.running.load(.acquire)) { - self.cond.waitUncancelable(self.queue_io, &self.mutex); + self.cond.waitUncancelable(self.io, &self.mutex); } if (self.len == 0) return null; @@ -225,20 +222,20 @@ } pub fn queueDepth(self: *Resyncer) usize { - self.mutex.lockUncancelable(self.queue_io); - defer self.mutex.unlock(self.queue_io); + self.mutex.lockUncancelable(self.io); + defer self.mutex.unlock(self.io); return self.len; } pub fn stop(self: *Resyncer) void { self.running.store(false, .release); - self.cond.signal(self.queue_io); + self.cond.signal(self.io); } pub fn deinit(self: *Resyncer) void { self.stop(); - if (self.future) |*f| f.cancel(self.io); - self.future = null; + if (self.thread) |t| t.join(); + self.thread = null; } const DescribeRepoResponse = struct { @@ -253,7 +250,6 @@ var r: Resyncer = .{ .allocator = std.testing.allocator, .io = std.testing.io, - .queue_io = std.testing.io, .collection_index = undefined, // not used in this test .queue = undefined, .head = 0, @@ -262,7 +258,7 @@ .mutex = Io.Mutex.init, .cond = Io.Condition.init, .running = .{ .raw = true }, - .future = null, + .thread = null, .processed = .{ .raw = 0 }, .failed = .{ .raw = 0 }, .dropped = .{ .raw = 0 }, @@ -282,7 +278,6 @@ var r: Resyncer = .{ .allocator = std.testing.allocator, .io = std.testing.io, - .queue_io = std.testing.io, .collection_index = undefined, .queue = undefined, .head = 0, @@ -291,7 +286,7 @@ .mutex = Io.Mutex.init, .cond = Io.Condition.init, .running = .{ .raw = true }, - .future = null, + .thread = null, .processed = .{ .raw = 0 }, .failed = .{ .raw = 0 }, .dropped = .{ .raw = 0 }, @@ -305,7 +300,6 @@ var r: Resyncer = .{ .allocator = std.testing.allocator, .io = std.testing.io, - .queue_io = std.testing.io, .collection_index = undefined, .queue = undefined, .head = 0, @@ -314,7 +308,7 @@ .mutex = Io.Mutex.init, .cond = Io.Condition.init, .running = .{ .raw = true }, - .future = null, + .thread = null, .processed = .{ .raw = 0 }, .failed = .{ .raw = 0 }, .dropped = .{ .raw = 0 }, -- tangled.sh