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 a25c380..c0bd70d 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,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,7 +219,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,17 +227,24 @@ 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. - 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", .{}); } }; 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..0696f33 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| { @@ -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; }, } 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); +} 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