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