Keep the replication connection single-threaded

build.zig

@@ -17,8 +17,18 @@ pub fn build(b: *std.Build) void {
         "Version string embedded into the outboxx binary",
     ) orelse "0.2.0";
 
+    // Minimum log level compiled in. Default info keeps Release/prod output clean
+    // (lower levels are filtered at comptime); the load stand builds with
+    // -Dlog_level=debug to trace teardown / fail-fast behavior.
+    const log_level = b.option(
+        std.log.Level,
+        "log_level",
+        "Minimum log level compiled in: err, warn, info, debug (default: info)",
+    ) orelse .info;
+
     const build_options = b.addOptions();
     build_options.addOption([]const u8, "version", version);
+    build_options.addOption(std.log.Level, "log_level", log_level);
 
     // Constants module (application-wide constants)
     const constants_module = b.createModule(.{

src/constants.zig

@@ -1,7 +1,11 @@
+const std = @import("std");
 const builtin = @import("builtin");
 const build_options = @import("build_options");
 
 pub const VERSION = build_options.version;
+/// Minimum log level compiled in; set with -Dlog_level=debug|info|warn|err.
+/// addOption emits its own copy of the enum, so map it back onto std.log.Level by tag.
+pub const LOG_LEVEL: std.log.Level = std.meta.stringToEnum(std.log.Level, @tagName(build_options.log_level)).?;
 pub const APP_NAME = "Outboxx";
 pub const DESCRIPTION = "PostgreSQL Change Data Capture with Kafka";
 

src/kafka/producer.zig

@@ -150,8 +150,11 @@ pub const KafkaProducer = struct {
         self.topics.deinit();
 
         if (self.producer) |producer| {
+            std.log.debug("producer.deinit: flush (timeout {d}ms)", .{constants.CDC.KAFKA_FLUSH_TIMEOUT_MS});
             _ = c.rd_kafka_flush(producer, constants.CDC.KAFKA_FLUSH_TIMEOUT_MS);
+            std.log.debug("producer.deinit: flush returned, destroying", .{});
             c.rd_kafka_destroy(producer);
+            std.log.debug("producer.deinit: destroyed", .{});
         }
     }
 

src/main.zig

@@ -12,6 +12,12 @@ const constants = @import("constants");
 const observability = @import("observability");
 const Observability = observability.Observability;
 
+// Log level is a build option (-Dlog_level=..., default info), so lower levels
+// are compiled out of Release/prod. The load stand builds with debug.
+pub const std_options: std.Options = .{
+    .log_level = constants.LOG_LEVEL,
+};
+
 pub const CliError = error{
     NoConfigPath,
 };
@@ -139,7 +145,11 @@ fn run(init: std.process.Init) !void {
         var metrics_future = try init.io.concurrent(observability.serve, .{
             init.io, &obs, obs_cfg.address, obs_cfg.port, &shutdown_requested,
         });
-        defer metrics_future.cancel(init.io);
+        defer {
+            std.log.debug("main: cancelling metrics server", .{});
+            metrics_future.cancel(init.io);
+            std.log.debug("main: metrics server cancelled", .{});
+        }
         try processor.startStreaming(init.io, &shutdown_requested);
     } else {
         try processor.startStreaming(init.io, &shutdown_requested);

src/processor/processor.zig

@@ -47,22 +47,26 @@ pub fn matchStreams(allocator: std.mem.Allocator, streams: []const Stream, table
     return matched;
 }
 
-fn flushCommitWorker(
+// Background worker: flush Kafka off the hot path and record how far we have
+// flushed in flushed_lsn. It must NOT touch the Postgres connection -- libpq
+// forbids using one PGconn from two threads, and the receive loop already owns
+// it. The main thread reads flushed_lsn and confirms it to Postgres.
+fn flushWorker(
     io: std.Io,
     producer: *KafkaProducer,
-    source: *PostgresSource,
     pending_lsn: *std.atomic.Value(u64),
+    flushed_lsn: *std.atomic.Value(u64),
+    flush_interval_sec: u32,
 ) void {
     var iterations: u32 = 0;
-    const flush_interval_iterations: u32 = @intCast(constants.CDC.KAFKA_FLUSH_INTERVAL_SEC);
 
     while (true) {
         // The worker's only cancelation point: on shutdown the future is
         // canceled, this returns error.Canceled, and we break to the final flush.
         io.sleep(.fromSeconds(1), .awake) catch break;
         iterations += 1;
 
-        if (iterations < flush_interval_iterations) {
+        if (iterations < flush_interval_sec) {
             continue;
         }
 
@@ -78,24 +82,13 @@ fn flushCommitWorker(
             continue;
         }
 
-        source.sendFeedback(io, lsn) catch |err| {
-            std.log.err("Background LSN commit failed: {}", .{err});
-            continue;
-        };
-    }
-
-    producer.flush(constants.CDC.KAFKA_FLUSH_TIMEOUT_MS) catch |err| {
-        std.log.warn("Final background flush failed: {}", .{err});
-    };
-
-    const lsn = pending_lsn.load(.acquire);
-    if (lsn > 0) {
-        source.sendFeedback(io, lsn) catch |err| {
-            std.log.warn("Final background LSN commit failed: {}", .{err});
-        };
+        flushed_lsn.store(lsn, .release);
     }
 
-    std.log.debug("Flush/commit worker stopped", .{});
+    // No final flush on cancel: awaiting a flush into a wedged broker is what
+    // hangs the shutdown. The graceful path flushes on the main thread instead;
+    // on a fatal error we just exit.
+    std.log.debug("Flush worker stopped", .{});
 }
 
 /// CDC Processor that works with PostgreSQL streaming replication
@@ -108,7 +101,14 @@ pub const Processor = struct {
     obs: *Observability,
 
     events_processed: usize,
+    // Staged by the main thread after producing a batch to Kafka's local queue.
     pending_lsn: std.atomic.Value(u64),
+    // Set by the flush worker after a successful Kafka flush; read by the main
+    // thread, which confirms it to Postgres. Keeps all libpq access single-threaded.
+    flushed_lsn: std.atomic.Value(u64),
+    // Seconds between background Kafka flushes. Defaults to the constant; tests
+    // shorten it to exercise the periodic flush/commit path quickly.
+    flush_interval_sec: u32,
 
     const Self = @This();
 
@@ -122,17 +122,25 @@ pub const Processor = struct {
             .obs = obs,
             .events_processed = 0,
             .pending_lsn = std.atomic.Value(u64).init(0),
+            .flushed_lsn = std.atomic.Value(u64).init(0),
+            .flush_interval_sec = @intCast(constants.CDC.KAFKA_FLUSH_INTERVAL_SEC),
         };
     }
 
     pub fn deinit(self: *Self) void {
+        std.log.debug("processor.deinit: deinit producer", .{});
         self.producer.deinit();
+        std.log.debug("processor.deinit: producer done, deinit source", .{});
         self.source.deinit();
+        std.log.debug("processor.deinit: done", .{});
     }
 
     /// Receive one batch, route each change to its streams, and stage the batch LSN for commit.
     pub fn processChangesToKafka(self: *Self, io: std.Io, batch_allocator: std.mem.Allocator, limit: u32) !void {
-        var batch = try self.source.receiveBatch(io, batch_allocator, limit);
+        // Confirm how far the flush worker has flushed to Kafka; the source sends
+        // it as standby feedback before reading the next batch.
+        const confirmed_lsn = self.flushed_lsn.load(.acquire);
+        var batch = try self.source.receiveBatch(io, batch_allocator, limit, confirmed_lsn);
         defer batch.deinit();
 
         // A successful receive (even an empty batch) means we are connected and
@@ -161,11 +169,6 @@ pub const Processor = struct {
             defer matched.deinit(batch_allocator);
 
             if (matched.items.len == 0) {
-                std.log.debug("No matching streams for {s}.{s} ({s})", .{
-                    change_event.meta.schema,
-                    change_event.meta.resource,
-                    change_event.op,
-                });
                 continue;
             }
 
@@ -177,6 +180,7 @@ pub const Processor = struct {
 
                 producer.sendMessage(topic_name, partition_key, json_bytes) catch |err| {
                     self.obs.recordProduceError();
+                    std.log.debug("processChangesToKafka: sendMessage failed, propagating {} (fail-fast)", .{err});
                     return err;
                 };
 
@@ -186,14 +190,6 @@ pub const Processor = struct {
                 if (self.events_processed % 10000 == 0) {
                     std.log.info("Processed {} CDC events", .{self.events_processed});
                 }
-
-                std.log.debug("Sent {s} message for {s}.{s} to topic '{s}' (key: {s})", .{
-                    change_event.op,
-                    change_event.meta.schema,
-                    change_event.meta.resource,
-                    topic_name,
-                    partition_key,
-                });
             }
         }
 
@@ -223,25 +219,32 @@ pub const Processor = struct {
         return try allocator.dupe(u8, change_event.meta.resource);
     }
 
-    /// Run the batch loop until stop_signal is set, with a background flush/commit worker.
+    /// Run the batch loop until stop_signal is set, with a background Kafka flush worker.
     pub fn startStreaming(self: *Self, io: std.Io, stop_signal: *std.atomic.Value(bool)) !void {
         // The source is connected and validated before we get here, so readiness's
         // connection signal goes up now; markStreaming follows on the first batch.
         self.obs.markConnected(true);
 
         const producer = &self.producer;
 
-        // Background flush/commit loop. `concurrent` not `async`: it must run
-        // alongside the receive loop, and `async` may defer the call until await.
-        var flush_future = try io.concurrent(flushCommitWorker, .{
+        // Background flush loop. `concurrent` not `async`: it must run alongside
+        // the receive loop, and `async` may defer the call until await. It only
+        // touches Kafka; LSN feedback stays on this thread (see commitFlushed).
+        var flush_future = try io.concurrent(flushWorker, .{
             io,
             producer,
-            &self.source,
             &self.pending_lsn,
+            &self.flushed_lsn,
+            self.flush_interval_sec,
         });
-        // On a receive error, still stop the worker (wakes its sleep for the
-        // final flush, and awaits) before the error propagates.
-        errdefer flush_future.cancel(io);
+        // On a receive error, stop the worker before the error propagates (it uses
+        // the producer, which processor.deinit destroys). The worker touches no
+        // libpq and no longer flushes on cancel, so this returns promptly.
+        errdefer {
+            std.log.debug("startStreaming: error path, cancelling flush worker", .{});
+            flush_future.cancel(io);
+            std.log.debug("startStreaming: flush worker cancelled, propagating error", .{});
+        }
 
         while (!stop_signal.load(.monotonic)) {
             self.obs.heartbeat(io);
@@ -254,9 +257,19 @@ pub const Processor = struct {
             try self.processChangesToKafka(io, batch_alloc, constants.CDC.BATCH_SIZE);
         }
 
-        // Graceful stop: cancel and await the worker (runs its final flush/commit)
-        // before reporting the stream stopped.
+        // Graceful stop: stop the worker, then flush everything staged and confirm
+        // it, both on this thread. After a full flush pending_lsn is durable, so it
+        // is safe to confirm (and covers the last batch the loop just produced).
         flush_future.cancel(io);
+        producer.flush(constants.CDC.KAFKA_FLUSH_TIMEOUT_MS) catch |err| {
+            std.log.warn("Final flush failed: {}", .{err});
+        };
+        const final_lsn = self.pending_lsn.load(.acquire);
+        if (final_lsn > 0) {
+            self.source.sendFeedback(io, final_lsn) catch |err| {
+                std.log.warn("Final LSN feedback failed: {}", .{err});
+            };
+        }
         std.log.info("Streaming stopped gracefully", .{});
     }
 };

src/source/postgres/integration_test.zig

@@ -162,7 +162,7 @@ test "Streaming source: receive and convert INSERT messages to ChangeEvents" {
 
     std.log.info("Streaming source connected, receiving batch...", .{});
 
-    const batch = try source.receiveBatch(std.testing.io, allocator, 10);
+    const batch = try source.receiveBatch(std.testing.io, allocator, 10, 0);
     defer {
         var mut_batch = batch;
         mut_batch.deinit();
@@ -262,7 +262,7 @@ test "Streaming source: UPDATE operation E2E with old and new tuples" {
 
     try source.connect(conn_str, start_lsn);
 
-    const batch = try source.receiveBatch(std.testing.io, allocator, 10);
+    const batch = try source.receiveBatch(std.testing.io, allocator, 10, 0);
     defer {
         var mut_batch = batch;
         mut_batch.deinit();
@@ -390,7 +390,7 @@ test "Streaming source: unchanged TOAST column becomes the placeholder" {
 
     try source.connect(conn_str, start_lsn);
 
-    const batch = try source.receiveBatch(std.testing.io, allocator, 10);
+    const batch = try source.receiveBatch(std.testing.io, allocator, 10, 0);
     defer {
         var mut_batch = batch;
         mut_batch.deinit();
@@ -504,7 +504,7 @@ test "Streaming source: DELETE operation E2E" {
 
     try source.connect(conn_str, start_lsn);
 
-    const batch = try source.receiveBatch(std.testing.io, allocator, 10);
+    const batch = try source.receiveBatch(std.testing.io, allocator, 10, 0);
     defer {
         var mut_batch = batch;
         mut_batch.deinit();
@@ -619,7 +619,7 @@ test "Streaming source: Multiple batches with limit parameter" {
     var last_lsn: u64 = 0;
 
     while (total_changes < 500) {
-        const batch = try source.receiveBatch(std.testing.io, allocator, 100);
+        const batch = try source.receiveBatch(std.testing.io, allocator, 100, 0);
         defer {
             var mut_batch = batch;
             mut_batch.deinit();
@@ -707,7 +707,7 @@ test "Streaming source: Timeout behavior with no data" {
 
     const start_time = test_helpers.nowMillis(std.testing.io);
 
-    const batch = try source.receiveBatchWithWaitTime(std.testing.io, allocator, 10, 1000);
+    const batch = try source.receiveBatchWithWaitTime(std.testing.io, allocator, 10, 0, 1000);
     defer {
         var mut_batch = batch;
         mut_batch.deinit();

src/source/postgres/source.zig

@@ -78,6 +78,8 @@ pub const PostgresSource = struct {
     // The stream does not expose the server WAL head during a backlog, so lag is
     // measured as wall-clock time behind this commit, like Debezium.
     last_commit_time: i64,
+    // Last LSN sent to Postgres as standby feedback, to skip redundant updates.
+    last_feedback_lsn: u64,
 
     const Self = @This();
 
@@ -94,6 +96,7 @@ pub const PostgresSource = struct {
             .converter = Converter.init(allocator),
             .last_lsn = 0,
             .last_commit_time = 0,
+            .last_feedback_lsn = 0,
         };
     }
 
@@ -129,16 +132,28 @@ pub const PostgresSource = struct {
         std.log.info("Streaming replication started from LSN: {s}", .{start_lsn});
     }
 
-    /// Receive batch of changes from PostgreSQL (default wait time from constants)
-    /// Wrapper for compatibility with polling source API
-    pub fn receiveBatch(self: *Self, io: std.Io, batch_allocator: std.mem.Allocator, limit: usize) PostgresSourceError!Batch {
-        return self.receiveBatchWithWaitTime(io, batch_allocator, limit, constants.CDC.BATCH_WAIT_MS);
+    /// Receive a batch of changes, first confirming to Postgres how far we have
+    /// durably delivered downstream. confirmed_lsn is what the Kafka flush worker
+    /// has flushed; feedback rides the same replication loop as the read.
+    pub fn receiveBatch(self: *Self, io: std.Io, batch_allocator: std.mem.Allocator, limit: usize, confirmed_lsn: u64) PostgresSourceError!Batch {
+        return self.receiveBatchWithWaitTime(io, batch_allocator, limit, confirmed_lsn, constants.CDC.BATCH_WAIT_MS);
     }
 
     /// Receive batch of changes from PostgreSQL (with wait time)
     /// limit: desired batch size (soft limit)
     /// wait_time_ms: max time to wait for batch
-    pub fn receiveBatchWithWaitTime(self: *Self, io: std.Io, batch_allocator: std.mem.Allocator, limit: usize, wait_time_ms: i32) PostgresSourceError!Batch {
+    pub fn receiveBatchWithWaitTime(self: *Self, io: std.Io, batch_allocator: std.mem.Allocator, limit: usize, confirmed_lsn: u64, wait_time_ms: i32) PostgresSourceError!Batch {
+        // Standby feedback rides the receive loop: before asking for more WAL,
+        // confirm how far we have durably delivered. Sent only when it advances;
+        // a failure is non-fatal here (the read below surfaces a dead link).
+        if (confirmed_lsn > self.last_feedback_lsn) {
+            if (self.sendFeedback(io, confirmed_lsn)) |_| {
+                self.last_feedback_lsn = confirmed_lsn;
+            } else |err| {
+                std.log.warn("Failed to send standby status update: {}", .{err});
+            }
+        }
+
         var changes = std.ArrayList(ChangeEvent).empty;
         errdefer {
             for (changes.items) |*change| {
@@ -251,14 +266,16 @@ pub const PostgresSource = struct {
                 return xlog.server_wal_end;
             },
             .keepalive => |keepalive| {
-                // Do NOT send reply here (even if reply_requested=true)
-                // Reason: We must maintain at-least-once guarantee
-                // - Sending reply here would confirm LSN before Kafka flush
-                // - If Kafka flush fails, data would be lost
-                // - Processor will send feedback after successful Kafka flush
-                //
-                // Note: PostgreSQL may wait for reply, but our batch wait time (~6 sec)
-                // is much shorter than wal_sender_timeout (default 60 sec)
+                // Reply when Postgres asks, reporting the last DURABLY confirmed
+                // position (last_feedback_lsn), never the received LSN -- confirming
+                // unflushed data would break at-least-once. Without this reply, a
+                // backlog where flushed_lsn is not advancing yet sends no status
+                // update, and Postgres drops the walsender on wal_sender_timeout.
+                if (keepalive.reply_requested) {
+                    self.sendFeedback(io, self.last_feedback_lsn) catch |err| {
+                        std.log.warn("Failed to reply to keepalive: {}", .{err});
+                    };
+                }
                 return keepalive.server_wal_end;
             },
         }