From 1775e5074cd5eb338216d1582a54ed3a05af763a Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Sat, 11 Jul 2026 16:36:47 +0200 Subject: [PATCH 1/3] Keep the replication connection single-threaded The flush/commit worker ran on its own thread and called both producer.flush() (Kafka) and source.sendFeedback() (libpq) on the same PGconn the main receive loop reads from. libpq forbids using one connection from two threads; under load the race wedged the connection, so a Kafka queue-full stall hung the graceful teardown in PQflush until SIGINT (EINTR) broke it loose, instead of failing fast. Keep the worker for the slow Kafka flush only: it records how far it has flushed in an atomic flushed_lsn and never touches libpq. The main thread sends standby feedback -- receiveBatch now takes the confirmed LSN and reports it before reading more WAL, so feedback rides the same single-threaded replication loop as the read, like the native walreceiver. Graceful shutdown confirms the worker's final flush. The flush interval becomes a Processor field (default from the constant). Add an e2e test asserting confirmed_flush_lsn advances when a receive finds no new changes. --- src/processor/processor.zig | 64 ++++++++----- src/source/postgres/integration_test.zig | 12 +-- src/source/postgres/source.zig | 25 ++++- tests/e2e/cdc_test.zig | 116 +++++++++++++++++++++++ 4 files changed, 184 insertions(+), 33 deletions(-) diff --git a/src/processor/processor.zig b/src/processor/processor.zig index a25c380..a88d608 100644 --- a/src/processor/processor.zig +++ b/src/processor/processor.zig @@ -47,14 +47,18 @@ 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 @@ -62,7 +66,7 @@ fn flushCommitWorker( io.sleep(.fromSeconds(1), .awake) catch break; iterations += 1; - if (iterations < flush_interval_iterations) { + if (iterations < flush_interval_sec) { continue; } @@ -78,10 +82,7 @@ fn flushCommitWorker( continue; } - source.sendFeedback(io, lsn) catch |err| { - std.log.err("Background LSN commit failed: {}", .{err}); - continue; - }; + flushed_lsn.store(lsn, .release); } producer.flush(constants.CDC.KAFKA_FLUSH_TIMEOUT_MS) catch |err| { @@ -90,12 +91,10 @@ fn flushCommitWorker( 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", .{}); + std.log.debug("Flush worker stopped", .{}); } /// CDC Processor that works with PostgreSQL streaming replication @@ -108,7 +107,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,6 +128,8 @@ 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), }; } @@ -132,7 +140,10 @@ pub const Processor = struct { /// 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 @@ -223,7 +234,7 @@ 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. @@ -231,16 +242,18 @@ pub const Processor = struct { 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. + // On a receive error, stop the worker (wakes its sleep, awaits) before the + // error propagates. The worker touches no libpq, so this can't wedge on it. errdefer flush_future.cancel(io); while (!stop_signal.load(.monotonic)) { @@ -254,9 +267,16 @@ 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: cancel and await the worker (runs its final flush), then + // confirm the LSN it flushed on the way out. The loop's per-batch feedback + // (via receiveBatch) can't cover this last flush, so send it explicitly. flush_future.cancel(io); + const final_lsn = self.flushed_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", .{}); } }; diff --git a/src/source/postgres/integration_test.zig b/src/source/postgres/integration_test.zig index 98f0582..8642893 100644 --- a/src/source/postgres/integration_test.zig +++ b/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(); diff --git a/src/source/postgres/source.zig b/src/source/postgres/source.zig index 58069a3..2114cc9 100644 --- a/src/source/postgres/source.zig +++ b/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| { diff --git a/tests/e2e/cdc_test.zig b/tests/e2e/cdc_test.zig index dec37c1..2377820 100644 --- a/tests/e2e/cdc_test.zig +++ b/tests/e2e/cdc_test.zig @@ -431,3 +431,119 @@ test "E2E: DELETE operation - full pipeline verification" { std.debug.print("✓ 2 DELETE operations resulted in 2 DELETE messages\n", .{}); std.debug.print("✓ CDC pipeline captured all changes\n", .{}); } + +// Read a single text cell from a query, or null if no row / SQL NULL. Caller owns the result. +fn querySingleText(conn: *c.PGconn, allocator: std.mem.Allocator, sql: [:0]const u8) !?[]u8 { + const res = c.PQexec(conn, sql.ptr); + defer c.PQclear(res); + if (c.PQresultStatus(res) != c.PGRES_TUPLES_OK) return error.QueryFailed; + if (c.PQntuples(res) < 1) return null; + if (c.PQgetisnull(res, 0, 0) == 1) return null; + return try allocator.dupe(u8, std.mem.span(c.PQgetvalue(res, 0, 0))); +} + +test "E2E: a receive with no new changes still confirms the flushed LSN to the slot" { + const allocator = testing.allocator; + + const conn_str = try test_helpers.getTestConnectionString(allocator); + defer allocator.free(conn_str); + const conn_str_z = try allocator.dupeZ(u8, conn_str); + defer allocator.free(conn_str_z); + const conn = c.PQconnectdb(conn_str_z.ptr) orelse return error.ConnectionFailed; + defer c.PQfinish(conn); + if (c.PQstatus(conn) != c.CONNECTION_OK) return error.ConnectionFailed; + + const timestamp = test_helpers.nowSeconds(std.testing.io); + const table_name = try std.fmt.allocPrint(allocator, "idle_commit_{d}", .{timestamp}); + defer allocator.free(table_name); + const topic_name = try std.fmt.allocPrint(allocator, "topic.idle.commit.{d}", .{timestamp}); + defer allocator.free(topic_name); + const slot_name = try std.fmt.allocPrint(allocator, "idle_commit_slot_{d}", .{timestamp}); + defer allocator.free(slot_name); + const pub_name = try std.fmt.allocPrint(allocator, "idle_commit_pub_{d}", .{timestamp}); + defer allocator.free(pub_name); + + try test_helpers.createTestTable(conn, allocator, table_name); + const replica_sql = try test_helpers.formatSqlZ(allocator, "ALTER TABLE {s} REPLICA IDENTITY FULL;", .{table_name}); + defer allocator.free(replica_sql); + _ = c.PQexec(conn, replica_sql.ptr); + + const stream_config = try test_helpers.createTestStreamConfig(allocator, table_name, topic_name); + defer allocator.free(stream_config.name); + + // NOTE: source is moved into the processor, which deinits it (releasing the slot). + var source = PostgresSource.init(allocator, slot_name, pub_name); + defer { + const drop_tmp = std.fmt.allocPrint(allocator, "SELECT pg_drop_replication_slot('{s}');", .{slot_name}) catch unreachable; + defer allocator.free(drop_tmp); + const drop = allocator.dupeZ(u8, drop_tmp) catch unreachable; + defer allocator.free(drop); + _ = c.PQexec(conn, drop.ptr); + } + try source.connect(conn_str, "0/0"); + + const streams = try allocator.alloc(Stream, 1); + defer allocator.free(streams); + streams[0] = stream_config; + + var producer = try KafkaProducer.init(allocator, "localhost:9092", null); + try producer.testConnection(); + + var obs = Observability.noop(); + var processor = Processor.init(allocator, source, producer, streams, &obs); + defer processor.deinit(); + + // Baseline: the slot's confirmed position right after creation. + const baseline_sql = try test_helpers.formatSqlZ(allocator, "SELECT confirmed_flush_lsn::text FROM pg_replication_slots WHERE slot_name = '{s}';", .{slot_name}); + defer allocator.free(baseline_sql); + const baseline = (try querySingleText(conn, allocator, baseline_sql)) orelse try allocator.dupe(u8, "0/0"); + defer allocator.free(baseline); + + // Produce some WAL and receive it, so pending_lsn advances past the baseline. + const insert = try test_helpers.formatSqlZ(allocator, "INSERT INTO {s} (name, value) VALUES ('Alice', 1), ('Bob', 2);", .{table_name}); + defer allocator.free(insert); + _ = c.PQexec(conn, insert.ptr); + test_helpers.sleepNs(std.testing.io, 200_000_000); + + { + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + try processor.processChangesToKafka(std.testing.io, arena.allocator(), 100); + } + + // Simulate the flush worker: flush Kafka, then publish how far we flushed. + try processor.producer.flush(5000); + processor.flushed_lsn.store(processor.pending_lsn.load(.acquire), .release); + + // Now idle -- no new changes. The next receive must still send the confirmed + // LSN as standby feedback (receiveBatch sends it before reading), advancing + // the slot. + { + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + try processor.processChangesToKafka(std.testing.io, arena.allocator(), 100); + } + + // The slot's confirmed_flush_lsn must have advanced past the baseline. + const advanced_sql = try test_helpers.formatSqlZ( + allocator, + "SELECT (confirmed_flush_lsn > '{s}'::pg_lsn) FROM pg_replication_slots WHERE slot_name = '{s}';", + .{ baseline, slot_name }, + ); + defer allocator.free(advanced_sql); + + var advanced = false; + var attempts: usize = 0; + while (attempts < 25) : (attempts += 1) { // up to ~5s for the walsender to record it + test_helpers.sleepNs(std.testing.io, 200_000_000); + if (try querySingleText(conn, allocator, advanced_sql)) |val| { + defer allocator.free(val); + if (std.mem.eql(u8, val, "t")) { + advanced = true; + break; + } + } + } + + try testing.expect(advanced); +} From 0888c397413dee9a6ed42cd1e243774043bae714 Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Sat, 11 Jul 2026 20:53:11 +0200 Subject: [PATCH 2/3] Reply to keepalives and don't block shutdown on a Kafka flush Two failures surfaced under load: Postgres killed the walsender with 'terminating walsender due to replication timeout', and outboxx then hung in teardown instead of exiting, so it never restarted. Standby feedback only went out from receiveBatch when flushed_lsn advanced, and keepalives were never answered. During a startup backlog, where the worker's first successful flush + feedback did not land within wal_sender_timeout (60s), no status update was sent and Postgres dropped the connection. Now reply to a keepalive's reply_requested with the last durably-confirmed position (last_feedback_lsn, never the received LSN, so at-least-once holds), which resets the timeout even when flushed_lsn is not advancing yet. The teardown hung because errdefer flush_future.cancel awaited the worker's final producer.flush() into a wedged/backlogged broker. Drop the worker's final flush so cancel returns promptly; do the final flush on the main thread in the graceful path and confirm pending_lsn (durable after a full flush). On a fatal error the process now exits fast instead of hanging forever. --- src/processor/processor.zig | 28 +++++++++++++--------------- src/source/postgres/source.zig | 18 ++++++++++-------- 2 files changed, 23 insertions(+), 23 deletions(-) diff --git a/src/processor/processor.zig b/src/processor/processor.zig index a88d608..8ca71d0 100644 --- a/src/processor/processor.zig +++ b/src/processor/processor.zig @@ -85,15 +85,9 @@ fn flushWorker( flushed_lsn.store(lsn, .release); } - 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) { - flushed_lsn.store(lsn, .release); - } - + // 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", .{}); } @@ -252,8 +246,9 @@ pub const Processor = struct { &self.flushed_lsn, self.flush_interval_sec, }); - // On a receive error, stop the worker (wakes its sleep, awaits) before the - // error propagates. The worker touches no libpq, so this can't wedge on it. + // 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 flush_future.cancel(io); while (!stop_signal.load(.monotonic)) { @@ -267,11 +262,14 @@ 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), then - // confirm the LSN it flushed on the way out. The loop's per-batch feedback - // (via receiveBatch) can't cover this last flush, so send it explicitly. + // 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); - const final_lsn = self.flushed_lsn.load(.acquire); + 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}); diff --git a/src/source/postgres/source.zig b/src/source/postgres/source.zig index 2114cc9..0696f33 100644 --- a/src/source/postgres/source.zig +++ b/src/source/postgres/source.zig @@ -266,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; }, } From 03d4d930dc7ef49b5666bd7c92d4a2036f333467 Mon Sep 17 00:00:00 2001 From: Yegor Lukash Date: Sun, 12 Jul 2026 08:32:01 +0200 Subject: [PATCH 3/3] Add a log_level build flag and teardown tracing (#94) --- build.zig | 10 ++++++++++ src/constants.zig | 4 ++++ src/kafka/producer.zig | 3 +++ src/main.zig | 12 +++++++++++- src/processor/processor.zig | 23 +++++++++-------------- tests/load/outboxx/Dockerfile | 6 ++++-- 6 files changed, 41 insertions(+), 17 deletions(-) diff --git a/build.zig b/build.zig index 28e8941..2c5ea1a 100644 --- a/build.zig +++ b/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(.{ diff --git a/src/constants.zig b/src/constants.zig index 24b8196..1eb14bc 100644 --- a/src/constants.zig +++ b/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"; diff --git a/src/kafka/producer.zig b/src/kafka/producer.zig index feb0869..fac7be3 100644 --- a/src/kafka/producer.zig +++ b/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", .{}); } } diff --git a/src/main.zig b/src/main.zig index 2ec55a5..0616330 100644 --- a/src/main.zig +++ b/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); diff --git a/src/processor/processor.zig b/src/processor/processor.zig index 8ca71d0..c0bd70d 100644 --- a/src/processor/processor.zig +++ b/src/processor/processor.zig @@ -128,8 +128,11 @@ pub const Processor = struct { } 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. @@ -166,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; } @@ -182,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; }; @@ -191,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, - }); } } @@ -249,7 +240,11 @@ pub const Processor = struct { // 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 flush_future.cancel(io); + 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); diff --git a/tests/load/outboxx/Dockerfile b/tests/load/outboxx/Dockerfile index df28118..9d236c7 100644 --- a/tests/load/outboxx/Dockerfile +++ b/tests/load/outboxx/Dockerfile @@ -8,9 +8,11 @@ WORKDIR /src COPY flake.nix flake.lock ./ RUN nix develop --command echo "deps cached" -# Build outboxx from the local checkout (build context is the repo root) +# Build outboxx from the local checkout (build context is the repo root). +# -Dlog_level=debug keeps ReleaseFast perf but compiles in std.log.debug so the +# load stand can trace teardown / fail-fast behavior. COPY . . -RUN nix develop --command zig build -Doptimize=ReleaseFast +RUN nix develop --command zig build -Doptimize=ReleaseFast -Dlog_level=debug WORKDIR /app EXPOSE 9464