diff --git a/docs/design.md b/docs/design.md
index 316a39d..a7691ce 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -83,7 +83,7 @@ additionally:
 | frame pool workers | 16 (env: `FRAME_WORKERS`) | 8 MB | CBOR decode, validation, DB persist, broadcast |
 | resolver threads | 4–8 (env: `RESOLVER_THREADS`) | 8 MB | DID document resolution, signing key extraction, cache population |
 | consumer write threads | 1 per downstream consumer | 8 MB | drain ring buffer → WebSocket write, ping/pong keepalive |
-| flush thread | 1 | 8 MB | batched fsync of event log (100ms or 400 events) |
+| flush thread | 1 | 8 MB | batched writes to event log (100ms or 400 events) |
 | GC thread | 1 | 8 MB | event log file cleanup every 10 minutes |
 | crawl queue thread | 1 | 8 MB | process `requestCrawl` — validate hostname, describeServer, spawn worker |
 | metrics server | 1 | 8 MB | HTTP on internal port, prometheus scrape |
@@ -152,6 +152,19 @@ files named `evts-{startSeq}`, rotated every 10K events. buffered writes
 flushed every 100ms or 400 events (whichever comes first). GC deletes files
 older than `RELAY_RETENTION_HOURS` (default: 72h).
 
+frame workers also flush accepted events before updating repo state or
+acknowledging upstream progress. each subscriber keeps a bounded queue of 256
+completion receipts: its cursor advances only through the successful prefix,
+even when workers finish different DIDs out of order. a processing failure
+causes reconnect from that prefix; connection teardown waits for outstanding
+receipts before freeing their storage. these writes protect process-restart
+replay; they do not add an fsync guarantee for host or power failure.
+
+shutdown stops readers and drains frame workers while DB and broadcast
+consumers remain running, then flushes the log and stops those consumers.
+rotation prepares and registers the replacement segment before releasing the
+active file, so allocation, file creation, or DB errors leave it usable.
+
 cursor replay: `playback(cursor)` binary-searches log files for the starting
 seq, then streams entries forward. the broadcaster tries disk first, falls
 back to in-memory ring buffer.
diff --git a/src/internal/broadcaster.zig b/src/internal/broadcaster.zig
index ddbb384..384d06d 100644
--- a/src/internal/broadcaster.zig
+++ b/src/internal/broadcaster.zig
@@ -573,6 +573,7 @@ pub const Broadcaster = struct {
     http_fallback_ctx: ?*anyopaque = null,
     io: Io,
     shutdown: *std.atomic.Value(bool),
+    producers_done: std.atomic.Value(bool) = .{ .raw = false },
 
     pub fn init(allocator: Allocator, io: Io, shutdown: *std.atomic.Value(bool)) Broadcaster {
         return .{
@@ -596,6 +597,7 @@ pub const Broadcaster = struct {
         self.consumers.deinit(self.allocator);
         self.history.deinit();
         if (self.error_frame) |ef| self.allocator.free(ef);
+        while (self.broadcast_queue.pop()) |item| self.allocator.free(@constCast(item.data));
     }
 
     pub fn addConsumer(self: *Broadcaster, conn: *websocket.Conn) !*Consumer {
@@ -804,7 +806,8 @@ pub const Broadcaster = struct {
     /// consumers_mutex / consumer.mutex / consumer.cond.
     pub fn runBroadcastLoop(self: *Broadcaster) void {
         log.info("broadcaster fiber started", .{});
-        while (!self.shutdown.load(.acquire)) {
+        while (true) {
+            const finished = self.producers_done.load(.acquire);
             // only emit frames that are durably on disk: hold any whose seq is
             // past the persister's committed watermark until the next flush
             // advances it. without a persister (degenerate no-persist mode),
@@ -822,14 +825,13 @@ pub const Broadcaster = struct {
                 self.allocator.free(@constCast(item.data));
                 drained += 1;
             }
+            if (finished) break;
             if (drained == 0) {
                 // nothing emittable — yield briefly so other fibers run
                 self.io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return;
             }
         }
-        // drain remaining items on shutdown
         while (self.broadcast_queue.pop()) |item| {
-            self.broadcast(item.seq, item.data);
             self.allocator.free(@constCast(item.data));
         }
         log.info("broadcaster fiber stopped", .{});
@@ -1943,3 +1945,21 @@ test "concurrent broadcast queue + drain produces monotonic sequences" {
         prev_seq = f.seq;
     }
 }
+
+test "shutdown keeps draining until the last producer finishes" {
+    var shutdown = std.atomic.Value(bool){ .raw = true };
+    var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown);
+    defer bc.deinit();
+    const drain = try std.Thread.spawn(.{}, Broadcaster.runBroadcastLoop, .{&bc});
+    defer drain.join();
+    defer bc.producers_done.store(true, .release);
+    for (1..10_001) |seq| {
+        const data = try std.testing.allocator.dupe(u8, "event");
+        bc.broadcast_queue.push(seq, data, &bc.stats);
+    }
+    bc.producers_done.store(true, .release);
+    while (bc.stats.seq.load(.acquire) != 10_000) {
+        try std.testing.io.sleep(Io.Duration.fromMilliseconds(1), .awake);
+    }
+    try std.testing.expectEqual(@as(u32, 0), bc.broadcast_queue.depth());
+}
diff --git a/src/internal/event_log.zig b/src/internal/event_log.zig
index 1a388c2..0311922 100644
--- a/src/internal/event_log.zig
+++ b/src/internal/event_log.zig
@@ -434,7 +434,7 @@ pub const DiskPersist = struct {
 
         // flush remaining
         self.mutex.lockUncancelable(self.io);
-        self.flushLocked() catch {};
+        self.flushLocked() catch |err| log.err("final event flush failed: {s}", .{@errorName(err)});
         self.mutex.unlock(self.io);
 
         // free write buffer
@@ -987,7 +987,7 @@ pub const DiskPersist = struct {
 
         // flush if threshold hit
         if (self.evtbuf.items.len >= default_flush_threshold) {
-            self.flushLocked() catch {};
+            self.flushLocked() catch |err| log.warn("event flush deferred: {s}", .{@errorName(err)});
         }
 
         return seq;
@@ -1367,15 +1367,18 @@ pub const DiskPersist = struct {
         self.event_counter = 0;
     }
 
+    pub fn flush(self: *DiskPersist) !void {
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        try self.flushLocked();
+    }
+
     fn flushLocked(self: *DiskPersist) !void {
         if (self.evtbuf.items.len == 0) return;
 
         // write buffered bytes to current file
         const file = self.current_file orelse return;
-        file.writePositionalAll(self.io, self.outbuf.items, self.current_file_pos) catch |err| {
-            log.err("flush: write failed: {s}", .{@errorName(err)});
-            return err;
-        };
+        try file.writePositionalAll(self.io, self.outbuf.items, self.current_file_pos);
         self.current_file_pos += self.outbuf.items.len;
 
         // clear write buffer (bytes are now durable)
@@ -1407,7 +1410,7 @@ pub const DiskPersist = struct {
             // sleep for flush interval (replaces timedWait — Io.Condition has no timed variant)
             self.io.sleep(Io.Duration.fromMilliseconds(@intCast(default_flush_interval_ms)), .awake) catch {};
             self.mutex.lockUncancelable(self.io);
-            self.flushLocked() catch {};
+            self.flushLocked() catch |err| log.err("event flush failed: {s}", .{@errorName(err)});
             self.mutex.unlock(self.io);
         }
     }
@@ -1816,6 +1819,72 @@ test "persist and playback" {
     try std.testing.expectEqualStrings("payload-three", entries.items[2].data);
 }
 
+test "worker failure preserves upstream cursor and successful retry reaches disk" {
+    const cbor = @import("zat").cbor;
+    const broadcaster = @import("broadcaster.zig");
+    const validator_mod = @import("validator.zig");
+    const worker = @import("frame_worker.zig");
+    const ingest_ack = @import("ingest_ack.zig");
+    var tdb = try TestDb.create(try requireDatabaseUrl());
+    defer tdb.destroy();
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const dir_path = try tmpDirRealPath(std.testing.allocator, tmp);
+    defer std.testing.allocator.free(dir_path);
+    var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io);
+    defer dp.deinit();
+    var shutdown = std.atomic.Value(bool){ .raw = false };
+    var bc = broadcaster.Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown);
+    defer bc.deinit();
+    var validator = validator_mod.Validator.init(std.testing.allocator, &bc.stats, std.testing.io);
+    defer validator.deinit();
+    var progress: ingest_ack.Progress = .{};
+    var cursor: ?u64 = 10;
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+    const header = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{
+        .{ .key = "op", .value = .{ .unsigned = 1 } },
+        .{ .key = "t", .value = .{ .text = "#identity" } },
+    } });
+    const payload = try cbor.encodeAlloc(arena.allocator(), .{ .map = &.{
+        .{ .key = "did", .value = .{ .text = "did:plc:cursor-test" } },
+        .{ .key = "seq", .value = .{ .unsigned = 11 } },
+    } });
+
+    for (0..2) |attempt| {
+        var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 });
+        dp.allocator = if (attempt == 0) failing.allocator() else std.testing.allocator;
+        defer dp.allocator = std.testing.allocator;
+        var work: worker.FrameWork = .{
+            .data = try std.mem.concat(std.testing.allocator, u8, &.{ header, payload }),
+            .hostname = try std.testing.allocator.dupe(u8, "example.com"),
+            .host_id = 1,
+            .allocator = std.testing.allocator,
+            .io = std.testing.io,
+            .bc = &bc,
+            .validator = &validator,
+            .persist = &dp,
+            .collection_index = null,
+            .resyncer = null,
+            .receipt = try progress.reserve(std.testing.io, 11, &cursor),
+        };
+        bc.stats.pool_queued_bytes.store(work.data.len, .monotonic);
+        worker.processFrame(&work);
+        if (attempt == 0) {
+            try std.testing.expectError(error.FrameProcessingFailed, progress.drain(&cursor, true));
+            try std.testing.expectEqual(@as(?u64, 10), cursor);
+            try std.testing.expectEqual(@as(u64, 1), dp.cur_seq);
+            progress.reset();
+        } else {
+            try progress.drain(&cursor, true);
+            try std.testing.expectEqual(@as(?u64, 11), cursor);
+            try std.testing.expectEqual(@as(u64, 1), dp.committed_seq.load(.acquire));
+            const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1);
+            try std.testing.expectEqual(@as(?u64, 1), scan.last_seq);
+        }
+    }
+}
+
 test "persist allocation failures preserve sequence and buffered records" {
     const Case = struct {
         fn run(allocator: Allocator) !void {
@@ -1853,6 +1922,40 @@ test "persist allocation failures preserve sequence and buffered records" {
     try std.testing.checkAllAllocationFailures(std.testing.allocator, Case.run, .{});
 }
 
+test "failed threshold write retains ownership and bounds pending events until retry" {
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true });
+    defer file.close(std.testing.io);
+    const read_only = try tmp.dir.openFile(std.testing.io, "events", .{});
+    defer read_only.close(std.testing.io);
+    var dp: DiskPersist = .{
+        .allocator = std.testing.allocator,
+        .dir_path = "",
+        .dir = tmp.dir,
+        .db = undefined,
+        .current_file = read_only,
+        .did_cache = undefined,
+        .host_id_cache = undefined,
+        .io = std.testing.io,
+    };
+    defer {
+        for (dp.evtbuf.items) |job| std.testing.allocator.free(job.data);
+        dp.evtbuf.deinit(std.testing.allocator);
+        dp.outbuf.deinit(std.testing.allocator);
+    }
+    for (0..default_flush_threshold) |_| _ = try dp.persist(.commit, 1, "payload");
+    try std.testing.expectEqual(@as(u64, 0), dp.committed_seq.load(.acquire));
+    try std.testing.expectError(error.NotOpenForWriting, dp.persist(.commit, 1, "next"));
+    try std.testing.expectEqual(default_flush_threshold, dp.evtbuf.items.len);
+    try std.testing.expectEqual(default_flush_threshold + 1, dp.cur_seq);
+    dp.current_file = file;
+    try dp.flush();
+    const scan = try scanForLastSeq(file, std.testing.io, 1);
+    try std.testing.expectEqual(@as(?u64, default_flush_threshold), scan.last_seq);
+    try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len);
+}
+
 test "failed rotation preserves the active file and retries after database recovery" {
     var tdb = try TestDb.create(try requireDatabaseUrl());
     defer tdb.destroy();
diff --git a/src/internal/frame_worker.zig b/src/internal/frame_worker.zig
index 0cbbde4..4981f7f 100644
--- a/src/internal/frame_worker.zig
+++ b/src/internal/frame_worker.zig
@@ -18,6 +18,7 @@ const resync_mod = @import("collection_index/resync.zig");
 const status_checker_mod = @import("status_checker.zig");
 const thread_pool = @import("util/thread_pool.zig");
 const util = @import("util/util.zig");
+const ingest_ack = @import("ingest_ack.zig");
 
 const Allocator = std.mem.Allocator;
 const log = std.log.scoped(.relay);
@@ -36,13 +37,24 @@ pub const FrameWork = struct {
     collection_index: ?*collection_index_mod.CollectionIndex,
     resyncer: ?*resync_mod.Resyncer,
     status_checker: ?*status_checker_mod.StatusChecker = null,
+    receipt: ?*ingest_ack.Receipt = null,
 };
 
 pub fn processFrame(work: *FrameWork) void {
+    var success = false;
+    defer if (work.receipt) |receipt| receipt.complete(success);
     _ = work.bc.stats.pool_queued_bytes.fetchSub(work.data.len, .monotonic);
     defer work.allocator.free(work.data);
     defer work.allocator.free(work.hostname);
 
+    processFrameInner(work) catch |err| {
+        log.warn("frame processing failed for host {s}: {s}", .{ work.hostname, @errorName(err) });
+        return;
+    };
+    success = true;
+}
+
+fn processFrameInner(work: *FrameWork) !void {
     var arena = std.heap.ArenaAllocator.init(work.allocator);
     defer arena.deinit();
     const alloc = arena.allocator();
@@ -51,6 +63,7 @@ pub fn processFrame(work: *FrameWork) void {
 
     // re-decode header (cheap — ~1-2μs)
     const header_result = zat.cbor.decode(alloc, data) catch |err| {
+        if (err == error.OutOfMemory) return err;
         log.debug("worker: frame header decode failed: {s} (len={d})", .{ @errorName(err), data.len });
         _ = work.bc.stats.decode_errors.fetchAdd(1, .monotonic);
         return;
@@ -63,6 +76,7 @@ pub fn processFrame(work: *FrameWork) void {
 
     const frame_type = header.getString("t") orelse return;
     const payload = zat.cbor.decodeAll(alloc, payload_data) catch |err| {
+        if (err == error.OutOfMemory) return err;
         log.debug("worker: frame payload decode failed: {s} (type={s})", .{ @errorName(err), frame_type });
         _ = work.bc.stats.decode_errors.fetchAdd(1, .monotonic);
         return;
@@ -90,7 +104,7 @@ pub fn processFrame(work: *FrameWork) void {
     // resolve DID → numeric UID for event header (host-aware)
     const uid: u64 = if (work.persist) |dp| blk: {
         if (did) |d| {
-            const result = dp.uidForDidFromHost(d, work.host_id) catch break :blk @as(u64, 0);
+            const result = try dp.uidForDidFromHost(d, work.host_id);
 
             // host authority enforcement: verify DID doc confirms this host.
             // #identity events are exempt — any PDS can emit them (same as indigo).
@@ -109,7 +123,7 @@ pub fn processFrame(work: *FrameWork) void {
                 })) {
                     .migrate => {
                         // DID doc confirms the host — update and continue
-                        dp.setAccountHostId(result.uid, work.host_id) catch {};
+                        try dp.setAccountHostId(result.uid, work.host_id);
                         if (result.host_changed) {
                             log.info("host {s}: account migrated uid={d} did={s}", .{
                                 work.hostname, result.uid, d,
@@ -170,8 +184,7 @@ pub fn processFrame(work: *FrameWork) void {
         // drop events for inactive accounts.
         if (work.persist) |dp| {
             if (uid > 0) {
-                const activity = dp.accountActivity(uid) catch
-                    event_log_mod.DiskPersist.AccountActivity{ .local_ok = true, .upstream_ok = true };
+                const activity = try dp.accountActivity(uid);
                 if (!activity.active()) {
                     // a repo event from the account's own confirmed host while
                     // upstream_status is non-active is self-contradicting — the
@@ -208,7 +221,7 @@ pub fn processFrame(work: *FrameWork) void {
                 }
 
                 if (work.persist) |dp| {
-                    if (dp.getAccountState(uid, alloc) catch null) |prev| {
+                    if (try dp.getAccountState(uid, alloc)) |prev| {
                         // stale rev check
                         if (std.mem.order(u8, incoming_rev, prev.rev) != .gt) {
                             log.debug("host {s}: dropping stale commit uid={d} rev={s} <= {s}", .{
@@ -244,7 +257,7 @@ pub fn processFrame(work: *FrameWork) void {
                         if (work.validator.config.verify_commit_diff) {
                             if (payload.get("prevData")) |pd| {
                                 if (pd == .cid) {
-                                    const prev_data_encoded = zat.multibase.encode(alloc, .base32lower, pd.cid.raw) catch "";
+                                    const prev_data_encoded = try zat.multibase.encode(alloc, .base32lower, pd.cid.raw);
                                     if (prev_data_encoded.len > 0 and prev.data_cid.len > 0 and
                                         !std.mem.eql(u8, prev_data_encoded, prev.data_cid))
                                     {
@@ -314,23 +327,28 @@ pub fn processFrame(work: *FrameWork) void {
         const relay_seq = dp.persist(kind, uid, data) catch |err| {
             work.bc.persist_order.store(0, .release);
             log.warn("persist failed: {s}", .{@errorName(err)});
-            return;
+            return err;
         };
         work.bc.stats.relay_seq.store(relay_seq, .release);
 
-        const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse data;
-        const owned = work.allocator.dupe(u8, broadcast_data) catch {
+        const broadcast_data = broadcaster.resequenceFrame(alloc, data, relay_seq) orelse {
             work.bc.persist_order.store(0, .release);
-            return;
+            return error.FrameResequenceFailed;
+        };
+        const owned = work.allocator.dupe(u8, broadcast_data) catch |err| {
+            work.bc.persist_order.store(0, .release);
+            return err;
         };
         work.bc.broadcast_queue.push(relay_seq, owned, &work.bc.stats);
         work.bc.persist_order.store(0, .release);
 
+        try dp.flush();
+
         // update per-DID state outside the ordering lock (Postgres round-trip)
         if ((is_commit or is_sync) and uid > 0) {
             if (commit_rev) |rev| {
                 const cid_str: []const u8 = if (commit_data_cid) |cid_raw|
-                    zat.multibase.encode(alloc, .base32lower, cid_raw) catch ""
+                    try zat.multibase.encode(alloc, .base32lower, cid_raw)
                 else
                     "";
                 if (dp.updateAccountState(uid, rev, cid_str)) |updated| {
@@ -341,12 +359,13 @@ pub fn processFrame(work: *FrameWork) void {
                     }
                 } else |err| {
                     log.debug("account state update failed: {s}", .{@errorName(err)});
+                    return err;
                 }
             }
         }
     } else {
         const upstream_seq = payload.getUint("seq") orelse 0;
-        const owned = work.allocator.dupe(u8, data) catch return;
+        const owned = try work.allocator.dupe(u8, data);
         work.bc.broadcast_queue.push(upstream_seq, owned, &work.bc.stats);
     }
 }
diff --git a/src/internal/ingest_ack.zig b/src/internal/ingest_ack.zig
new file mode 100644
index 0000000..efe25af
--- /dev/null
+++ b/src/internal/ingest_ack.zig
@@ -0,0 +1,125 @@
+const std = @import("std");
+const Io = std.Io;
+
+pub const Receipt = struct {
+    io: Io = undefined,
+    seq: ?u64 = null,
+    mutex: Io.Mutex = .init,
+    condition: Io.Condition = .init,
+    result: ?bool = null,
+    retry_needed: ?*std.atomic.Value(bool) = null,
+
+    pub fn complete(self: *Receipt, success: bool) void {
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        self.result = success;
+        if (!success) {
+            if (self.retry_needed) |retry| retry.store(true, .release);
+        }
+        self.condition.signal(self.io);
+    }
+
+    fn read(self: *Receipt, wait: bool) ?bool {
+        self.mutex.lockUncancelable(self.io);
+        defer self.mutex.unlock(self.io);
+        while (wait and self.result == null) self.condition.waitUncancelable(self.io, &self.mutex);
+        return self.result;
+    }
+};
+
+pub const Progress = struct {
+    const capacity = 256;
+    receipts: [capacity]Receipt = [_]Receipt{.{}} ** capacity,
+    head: usize = 0,
+    count: usize = 0,
+    failed: bool = false,
+    retry_needed: std.atomic.Value(bool) = .{ .raw = false },
+
+    pub fn reserve(self: *Progress, io: Io, seq: ?u64, cursor: *?u64) !*Receipt {
+        try self.drain(cursor, false);
+        if (self.count == capacity) {
+            _ = self.receipts[self.head].read(true);
+            try self.drain(cursor, false);
+        }
+        const receipt = &self.receipts[(self.head + self.count) % capacity];
+        receipt.* = .{ .io = io, .seq = seq, .retry_needed = &self.retry_needed };
+        self.count += 1;
+        return receipt;
+    }
+
+    pub fn drain(self: *Progress, cursor: *?u64, wait: bool) !void {
+        while (self.count > 0) {
+            const receipt = &self.receipts[self.head];
+            const success = receipt.read(wait) orelse break;
+            if (!success) self.failed = true;
+            if (!self.failed) {
+                if (receipt.seq) |seq| cursor.* = seq;
+            }
+            self.head = (self.head + 1) % capacity;
+            self.count -= 1;
+        }
+        if (self.failed) return error.FrameProcessingFailed;
+    }
+
+    pub fn reset(self: *Progress) void {
+        std.debug.assert(self.count == 0);
+        self.failed = false;
+        self.retry_needed.store(false, .release);
+    }
+};
+
+test "cursor follows completion order rather than worker order" {
+    var progress: Progress = .{};
+    var cursor: ?u64 = 10;
+    const first = try progress.reserve(std.testing.io, 11, &cursor);
+    const second = try progress.reserve(std.testing.io, 14, &cursor);
+    second.complete(true);
+    try progress.drain(&cursor, false);
+    try std.testing.expectEqual(@as(?u64, 10), cursor);
+    first.complete(true);
+    try progress.drain(&cursor, false);
+    try std.testing.expectEqual(@as(?u64, 14), cursor);
+}
+
+test "failed frame blocks cursor advancement across later successes" {
+    var progress: Progress = .{};
+    var cursor: ?u64 = 10;
+    const first = try progress.reserve(std.testing.io, 11, &cursor);
+    const second = try progress.reserve(std.testing.io, 12, &cursor);
+    const third = try progress.reserve(std.testing.io, 13, &cursor);
+    first.complete(true);
+    second.complete(false);
+    third.complete(true);
+    try std.testing.expectError(error.FrameProcessingFailed, progress.drain(&cursor, true));
+    try std.testing.expectEqual(@as(?u64, 11), cursor);
+    try std.testing.expectEqual(@as(usize, 0), progress.count);
+    try std.testing.expectError(error.FrameProcessingFailed, progress.reserve(std.testing.io, 14, &cursor));
+}
+
+test "receipt drain waits for a worker before releasing its storage" {
+    var progress: Progress = .{};
+    var cursor: ?u64 = null;
+    const receipt = try progress.reserve(std.testing.io, 1, &cursor);
+    const worker = try std.Thread.spawn(.{}, Receipt.complete, .{ receipt, true });
+    defer worker.join();
+    try progress.drain(&cursor, true);
+    try std.testing.expectEqual(@as(?u64, 1), cursor);
+}
+
+test "canceled zio reader waits for native worker completion" {
+    var runtime: @import("zio").ZioBackend = undefined;
+    try runtime.init(std.testing.allocator, .{});
+    defer runtime.deinit();
+    try runtime.start();
+    var receipt: Receipt = .{ .io = runtime.io() };
+    var reader = try runtime.io().concurrent(Receipt.read, .{ &receipt, true });
+    const Worker = struct {
+        fn run(target: *Receipt) void {
+            std.testing.io.sleep(Io.Duration.fromMilliseconds(20), .awake) catch {};
+            target.complete(true);
+        }
+    };
+    const worker = try std.Thread.spawn(.{}, Worker.run, .{&receipt});
+    defer worker.join();
+    try std.testing.expectEqual(@as(?bool, true), reader.cancel(runtime.io()));
+}
diff --git a/src/internal/slurper.zig b/src/internal/slurper.zig
index 4659e2f..2b5e6fb 100644
--- a/src/internal/slurper.zig
+++ b/src/internal/slurper.zig
@@ -688,22 +688,17 @@ pub const Slurper = struct {
         self.crawl_cond.signal(self.io);
         if (self.crawl_future) |*f| f.cancel(self.io);
 
-        // collect futures to cancel (can't cancel while holding workers_mutex)
-        var futures_to_cancel: std.ArrayListUnmanaged(Io.Future(void)) = .empty;
-        defer futures_to_cancel.deinit(self.allocator);
-
-        {
+        while (true) {
             self.workers_mutex.lockUncancelable(self.io);
-            defer self.workers_mutex.unlock(self.io);
             var it = self.workers.iterator();
-            while (it.next()) |entry| {
-                futures_to_cancel.append(self.allocator, entry.value_ptr.future) catch {};
-            }
+            var future = if (it.next()) |entry| entry.value_ptr.future else {
+                self.workers_mutex.unlock(self.io);
+                break;
+            };
+            self.workers_mutex.unlock(self.io);
+            future.cancel(self.io);
         }
 
-        // cancel all subscriber tasks FIRST (they stop submitting to pool)
-        for (futures_to_cancel.items) |*f| f.cancel(self.io);
-
         // then drain + join pool workers (processes remaining queued frames)
         if (self.frame_pool) |*fp| {
             fp.shutdown();
diff --git a/src/internal/subscriber.zig b/src/internal/subscriber.zig
index a5ba3d4..4854334 100644
--- a/src/internal/subscriber.zig
+++ b/src/internal/subscriber.zig
@@ -17,6 +17,7 @@ const collection_index_mod = @import("collection_index/index.zig");
 const resync_mod = @import("collection_index/resync.zig");
 const status_checker_mod = @import("status_checker.zig");
 const frame_worker_mod = @import("frame_worker.zig");
+const ingest_ack = @import("ingest_ack.zig");
 const host_ops_mod = @import("host_ops.zig");
 const ws_ping_mod = @import("ws_ping.zig");
 const util = @import("util/util.zig");
@@ -203,6 +204,7 @@ pub const Subscriber = struct {
     resyncer: ?*resync_mod.Resyncer = null,
     status_checker: ?*status_checker_mod.StatusChecker = null,
     pool: ?*frame_worker_mod.FramePool = null,
+    progress: ingest_ack.Progress = .{},
     /// dedicated Threaded io for frame workers — safe from plain OS threads
     pool_io: ?Io = null,
     /// host ops queue — pushes rare DB ops to a background thread
@@ -355,6 +357,11 @@ pub const Subscriber = struct {
     }
 
     fn connectAndRead(self: *Subscriber) !void {
+        defer {
+            self.progress.drain(&self.last_upstream_seq, true) catch {};
+            self.flushCursor();
+            self.progress.reset();
+        }
         var path_buf: [256]u8 = undefined;
         var w: std.Io.Writer = .fixed(&path_buf);
 
@@ -489,7 +496,7 @@ pub const Subscriber = struct {
         // stagger by host so a cold start's simultaneous connects don't
         // produce a synchronized fleet-wide ping wave every interval
         var jitter: u64 = self.options.host_id % @max(1, cfg.interval_sec.load(.acquire));
-        while (jitter > 0 and !conn.read_done.load(.acquire) and !self.shouldStop()) : (jitter -= 1) {
+        while (jitter > 0 and !conn.read_done.load(.acquire) and !self.shouldStop() and !self.progress.retry_needed.load(.acquire)) : (jitter -= 1) {
             self.io.sleep(Io.Duration.fromSeconds(1), .awake) catch {};
         }
 
@@ -499,7 +506,7 @@ pub const Subscriber = struct {
             // are noticed promptly (bounds reconnect latency to ~1s)
             var slept: u64 = 0;
             while (slept < @max(1, cfg.interval_sec.load(.acquire))) : (slept += 1) {
-                if (conn.read_done.load(.acquire) or self.shouldStop()) break :outer;
+                if (conn.read_done.load(.acquire) or self.shouldStop() or self.progress.retry_needed.load(.acquire)) break :outer;
                 self.io.sleep(Io.Duration.fromSeconds(1), .awake) catch {};
             }
             // a ping is answered iff ANY frame arrived after it was sent.
@@ -585,6 +592,7 @@ const FrameHandler = struct {
         const alloc = arena.allocator();
 
         const header_result = zat.cbor.decode(alloc, data) catch |err| {
+            if (err == error.OutOfMemory) return err;
             log.debug("frame header decode failed: {s} (len={d})", .{ @errorName(err), data.len });
             _ = sub.bc.stats.decode_errors.fetchAdd(1, .monotonic);
             return;
@@ -620,6 +628,7 @@ const FrameHandler = struct {
 
         const frame_type = header.getString("t") orelse return;
         const payload = zat.cbor.decodeAll(alloc, payload_data) catch |err| {
+            if (err == error.OutOfMemory) return err;
             log.debug("frame payload decode failed: {s} (type={s})", .{ @errorName(err), frame_type });
             _ = sub.bc.stats.decode_errors.fetchAdd(1, .monotonic);
             return;
@@ -630,8 +639,9 @@ const FrameHandler = struct {
         const now_ms = milliTimestamp(io);
         sub.last_progress_ms.store(now_ms, .release);
 
-        // extract seq for cursor tracking (deferred until after pool accepts)
+        // advance only across the completed prefix of this upstream stream
         const upstream_seq = payload.getUint("seq");
+        try sub.progress.drain(&sub.last_upstream_seq, false);
 
         // time-based cursor flush (Go relay: every 4 seconds)
         {
@@ -669,7 +679,10 @@ const FrameHandler = struct {
         if (!is_commit and !is_sync and !is_account and !is_identity) {
             // advance cursor for intentionally skipped frames —
             // we won't process these on reconnect either
-            if (upstream_seq) |s| sub.last_upstream_seq = s;
+            if (sub.pool != null) {
+                const receipt = try sub.progress.reserve(io, upstream_seq, &sub.last_upstream_seq);
+                receipt.complete(true);
+            } else if (upstream_seq) |s| sub.last_upstream_seq = s;
             return;
         }
 
@@ -681,16 +694,17 @@ const FrameHandler = struct {
                 const d = payload.getString("repo") orelse payload.getString("did");
                 break :blk if (d) |s| std.hash.Wyhash.hash(0, s) else sub.options.host_id;
             };
-            const duped = sub.allocator.dupe(u8, data) catch return;
+            const duped = try sub.allocator.dupe(u8, data);
+            errdefer sub.allocator.free(duped);
             // dupe hostname per-frame: subscriber teardown (slurper.runWorker)
             // frees sub.options.hostname after sub.run() returns, but FrameWorks
             // can still be queued in the pool. borrowing the slice would be a
             // use-after-free (corrupt hostnames in chain-break logs, etc.).
-            const hostname_dup = sub.allocator.dupe(u8, sub.options.hostname) catch {
-                sub.allocator.free(duped);
-                return;
-            };
+            const hostname_dup = try sub.allocator.dupe(u8, sub.options.hostname);
+            errdefer sub.allocator.free(hostname_dup);
+            const receipt = try sub.progress.reserve(io, upstream_seq, &sub.last_upstream_seq);
             const t0 = nanoTimestamp(io);
+            _ = sub.bc.stats.pool_queued_bytes.fetchAdd(duped.len, .monotonic);
             if (pool.submit(did_key, .{
                 .data = duped,
                 .host_id = sub.options.host_id,
@@ -703,10 +717,8 @@ const FrameHandler = struct {
                 .collection_index = sub.collection_index,
                 .resyncer = sub.resyncer,
                 .status_checker = sub.status_checker,
+                .receipt = receipt,
             }, sub.shutdown)) {
-                // pool accepted — advance cursor past this frame
-                _ = sub.bc.stats.pool_queued_bytes.fetchAdd(duped.len, .monotonic);
-                if (upstream_seq) |s| sub.last_upstream_seq = s;
                 if (nanoTimestamp(io) - t0 > 1_000_000) { // >1ms = had to wait
                     _ = sub.bc.stats.pool_backpressure.fetchAdd(1, .monotonic);
                 }
@@ -714,6 +726,8 @@ const FrameHandler = struct {
                 // shutdown requested — don't advance cursor so reconnect replays this frame
                 sub.allocator.free(duped);
                 sub.allocator.free(hostname_dup);
+                _ = sub.bc.stats.pool_queued_bytes.fetchSub(duped.len, .monotonic);
+                receipt.complete(false);
             }
             return;
         }
diff --git a/src/main.zig b/src/main.zig
index 49520f1..c06016d 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -397,8 +397,9 @@ fn runRelay() !void {
     // DbRequestQueue — routes DB requests through pool_io worker threads.
     // originally needed to bridge Evented fibers to Threaded pg.Pool; now
     // redundant under all-Threaded but still functional. cleanup candidate.
+    var queue_shutdown = std.atomic.Value(bool){ .raw = false };
     var db_queue: event_log_mod.DbRequestQueue = .{
-        .shutdown = &shutdown_flag,
+        .shutdown = &queue_shutdown,
         .persist = &dp,
     };
 
@@ -461,7 +462,7 @@ fn runRelay() !void {
     var host_ops_queue: host_ops_mod.HostOpsQueue = .{
         .persist = &dp,
         .cursor_map = &cursor_map,
-        .shutdown = &shutdown_flag,
+        .shutdown = &queue_shutdown,
         .max_consecutive_failures = 15,
         .bc = &bc,
     };
@@ -487,7 +488,8 @@ fn runRelay() !void {
         io,
         pool_io,
     );
-    defer slurper.deinit();
+    var slurper_alive = true;
+    defer if (slurper_alive) slurper.deinit();
     slurper.collection_index = &ci;
     slurper.resyncer = &resyncer;
     slurper.status_checker = &status_checker;
@@ -611,7 +613,14 @@ fn runRelay() !void {
     // 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
+    metrics_srv.server.deinit(io);
+    metrics_future.cancel(io);
+
+    slurper.deinit();
+    slurper_alive = false;
+    queue_shutdown.store(true, .release);
+
+    // join db queue workers after their producers have stopped
     db_worker_1.join();
     db_worker_2.join();
 
@@ -622,13 +631,11 @@ fn runRelay() !void {
     backfiller.waitForCompletion();
     cleaner.waitForCompletion();
 
-    // cancel broadcaster fiber (shutdown flag already set, it will drain remaining)
-    broadcast_future.cancel(io);
-
-    // close metrics listener to unblock accept(), then cancel task
-    metrics_srv.server.deinit(io);
-    metrics_future.cancel(io);
+    const flush_result = dp.flush();
+    bc.producers_done.store(true, .release);
+    broadcast_future.await(io);
 
+    try flush_result;
     log.info("relay stopped cleanly", .{});
 }
 
diff --git a/src/tests.zig b/src/tests.zig
index cc13862..4fd30e8 100644
--- a/src/tests.zig
+++ b/src/tests.zig
@@ -17,6 +17,7 @@ test {
     _ = @import("internal/ingest_liveness.zig");
     _ = @import("internal/ws_ping.zig");
     _ = @import("internal/frame_worker.zig");
+    _ = @import("internal/ingest_ack.zig");
     _ = @import("internal/status_checker.zig");
     _ = @import("internal/collection_index/index.zig");
     _ = @import("internal/collection_index/backfill.zig");