diff --git a/src/config/config.zig b/src/config/config.zig index 94c3f41..d400697 100644 --- a/src/config/config.zig +++ b/src/config/config.zig @@ -40,6 +40,10 @@ pub const PostgresSource = struct { connection_env: []const u8, slot_name: []const u8, publication_name: []const u8, + // Load-testing aid: when false, the slot is never advanced past the first LSN, + // so a backlog generated once can be replayed by every restart. Keep it true in + // production, where advancing the slot is what makes delivery at-least-once. + confirm_lsn: bool = true, }; pub const SourceConfig = struct { diff --git a/src/main.zig b/src/main.zig index 610aad6..0c6750b 100644 --- a/src/main.zig +++ b/src/main.zig @@ -118,6 +118,10 @@ fn run(init: std.process.Init) !void { printStatus("Using PostgreSQL streaming replication (pgoutput protocol)\n", .{}); var source = PostgresSource.init(allocator, postgres.slot_name, postgres.publication_name); + source.confirm_lsn = postgres.confirm_lsn; + if (!postgres.confirm_lsn) { + std.log.warn("confirm_lsn is off: the replication slot will not advance (load-testing mode); restarts replay the same backlog and delivery is not durable", .{}); + } // NOTE: source will be deinit'd by processor.deinit() printStatus("Connecting to PostgreSQL streaming replication...\n", .{}); diff --git a/src/source/postgres/integration_test.zig b/src/source/postgres/integration_test.zig index 1882ccc..d507141 100644 --- a/src/source/postgres/integration_test.zig +++ b/src/source/postgres/integration_test.zig @@ -189,6 +189,86 @@ test "Streaming source: receive and convert INSERT messages to ChangeEvents" { std.log.info("Integration test passed!", .{}); } +test "Streaming source: confirm_lsn=false pins the slot at the first LSN" { + const allocator = testing.allocator; + + var prng = std.Random.DefaultPrng.init(@intCast(test_helpers.nowMicros(std.testing.io))); + const random_suffix = prng.random().int(u32); + const timestamp = test_helpers.nowSeconds(std.testing.io); + const table_name = try std.fmt.allocPrint(allocator, "pin_test_{d}_{d}", .{ timestamp, random_suffix }); + defer allocator.free(table_name); + const slot_name = try std.fmt.allocPrint(allocator, "slot_pin_{d}_{d}", .{ timestamp, random_suffix }); + defer allocator.free(slot_name); + const pub_name = try std.fmt.allocPrint(allocator, "pub_pin_{d}_{d}", .{ timestamp, random_suffix }); + defer allocator.free(pub_name); + + const setup_conn = try createSetupConnection(allocator); + defer c.PQfinish(setup_conn); + defer cleanupTestEnvironment(allocator, setup_conn, table_name, slot_name, pub_name); + + const create_table_sql = try test_helpers.formatSqlZ(allocator, "CREATE TABLE {s} (id SERIAL PRIMARY KEY, name TEXT)", .{table_name}); + defer allocator.free(create_table_sql); + try execSQL(setup_conn, create_table_sql); + + const create_pub_sql = try test_helpers.formatSqlZ(allocator, "CREATE PUBLICATION {s} FOR TABLE {s}", .{ pub_name, table_name }); + defer allocator.free(create_pub_sql); + try execSQL(setup_conn, create_pub_sql); + + const create_slot_sql = try test_helpers.formatSqlZ(allocator, "SELECT pg_create_logical_replication_slot('{s}', 'pgoutput')", .{slot_name}); + defer allocator.free(create_slot_sql); + try execSQL(setup_conn, create_slot_sql); + + const lsn_result = c.PQexec(setup_conn, "SELECT pg_current_wal_lsn()"); + defer c.PQclear(lsn_result); + const start_lsn = try allocator.dupeZ(u8, std.mem.span(c.PQgetvalue(lsn_result, 0, 0))); + defer allocator.free(start_lsn); + + // Several rows in one transaction so the first change LSN differs from the last. + const insert_sql = try test_helpers.formatSqlZ(allocator, "INSERT INTO {s} (name) VALUES ('r1'),('r2'),('r3'),('r4'),('r5')", .{table_name}); + defer allocator.free(insert_sql); + try execSQL(setup_conn, insert_sql); + try execSQL(setup_conn, "SELECT pg_switch_wal()"); + + const conn_str = try getTestConnectionString(allocator); + defer allocator.free(conn_str); + + var source = PostgresSource.init(allocator, slot_name, pub_name); + source.confirm_lsn = false; + defer source.deinit(); + + try source.connect(conn_str, start_lsn); + + const batch = try source.receiveBatch(std.testing.io, allocator, 10); + defer { + var mut_batch = batch; + mut_batch.deinit(); + } + + try testing.expect(source.first_lsn != 0); + try testing.expect(source.first_lsn < batch.last_lsn); + + // Feedback carries the batch's last LSN, but confirm_lsn=false must report the + // first LSN, so the slot pins there instead of advancing to the end. + try source.sendFeedback(std.testing.io, batch.last_lsn); + + const query = try test_helpers.formatSqlZ(allocator, "SELECT (confirmed_flush_lsn - '0/0'::pg_lsn)::text FROM pg_replication_slots WHERE slot_name = '{s}'", .{slot_name}); + defer allocator.free(query); + + // The walsender applies feedback asynchronously, so poll for the pinned value. + var confirmed: u64 = 0; + var attempt: usize = 0; + while (attempt < 100) : (attempt += 1) { + const r = c.PQexec(setup_conn, query.ptr); + defer c.PQclear(r); + confirmed = try std.fmt.parseInt(u64, std.mem.span(c.PQgetvalue(r, 0, 0)), 10); + if (confirmed == source.first_lsn) break; + std.testing.io.sleep(.fromMilliseconds(20), .awake) catch {}; + } + + try testing.expectEqual(source.first_lsn, confirmed); + try testing.expect(confirmed < batch.last_lsn); +} + test "ReplicationProtocol: mixed-case slot and publication names are idempotent across restarts" { const allocator = testing.allocator; diff --git a/src/source/postgres/source.zig b/src/source/postgres/source.zig index d0cd735..dbced73 100644 --- a/src/source/postgres/source.zig +++ b/src/source/postgres/source.zig @@ -82,6 +82,10 @@ 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, + // When false, feedback always confirms first_lsn instead of advancing, pinning + // the slot so a load-test backlog can be replayed on every restart (see config). + confirm_lsn: bool, + first_lsn: u64, // LSN of the first change seen; the pin point when confirm_lsn is false const Self = @This(); @@ -98,6 +102,8 @@ pub const PostgresSource = struct { .converter = Converter.init(allocator), .last_lsn = 0, .last_commit_time = 0, + .confirm_lsn = true, + .first_lsn = 0, }; } @@ -186,6 +192,7 @@ pub const PostgresSource = struct { const msg_lsn = try self.extractChangeFromMessage(batch_allocator, msg, &changes); last_confirmed_lsn = msg_lsn; // Update LSN (always > 0 on success) + if (self.first_lsn == 0 and msg_lsn != 0) self.first_lsn = msg_lsn; // Step 2: DRAIN all buffered messages (non-blocking) while (changes.items.len < limit) { @@ -199,6 +206,7 @@ pub const PostgresSource = struct { const buffered_lsn = try self.extractChangeFromMessage(batch_allocator, buffered_msg, &changes); last_confirmed_lsn = buffered_lsn; // Update LSN (always > 0 on success) + if (self.first_lsn == 0 and buffered_lsn != 0) self.first_lsn = buffered_lsn; } } @@ -278,10 +286,16 @@ pub const PostgresSource = struct { /// Send LSN feedback to PostgreSQL (confirm processing) pub fn sendFeedback(self: *Self, io: std.Io, lsn: u64) PostgresSourceError!void { + // With confirm_lsn off, pin the slot at the first LSN seen: feedback (and its + // keepalive) keeps flowing, but the slot never advances, so a load-test + // backlog replays on every restart. first_lsn is 0 until the first change, + // where reporting 0 is a harmless keepalive that also does not advance. + const feedback_lsn = if (self.confirm_lsn) lsn else self.first_lsn; + const status = StandbyStatusUpdate{ - .wal_write_position = lsn, - .wal_flush_position = lsn, - .wal_apply_position = lsn, + .wal_write_position = feedback_lsn, + .wal_flush_position = feedback_lsn, + .wal_apply_position = feedback_lsn, .client_time = std.Io.Timestamp.now(io, .real).toSeconds(), // Request an immediate keepalive back. Our regular feedback keeps the // walsender quiet (it only probes after wal_sender_timeout/2 of client diff --git a/tests/load/README.md b/tests/load/README.md index 03178d7..59ff751 100644 --- a/tests/load/README.md +++ b/tests/load/README.md @@ -18,7 +18,7 @@ What each command does: - `make infra` starts PostgreSQL, Kafka, Kafka UI, Prometheus, Grafana, exporters, and cAdvisor. It stops Debezium/Outboxx if they exist and creates logical replication slots before any workload. - `make load` brings infra and slots up if needed, then generates PostgreSQL writes. It leaves any running readers in place, so lag and throughput can be watched live while the load runs. For the readers-down backlog scenario (WAL accumulating behind the slots), run `make infra` first. - `make start-debezium` starts only Debezium. Outboxx is stopped/removed first. -- `make start-outboxx` starts only Outboxx. Debezium and connector-init are stopped/removed first. +- `make start-outboxx` starts only Outboxx. Debezium and connector-init are stopped/removed first. Its config sets `confirm_lsn = false`, so the slot never advances: generate the backlog once with `make load`, then rerun `make start-outboxx` as many times as you like and each run replays the same data from the start. - `make start-all` starts Debezium and Outboxx together. Debezium is registered automatically. - `make reset` stops the stand and removes this stand's PostgreSQL, Kafka, Prometheus, and Grafana volumes. diff --git a/tests/load/outboxx/Dockerfile b/tests/load/outboxx/Dockerfile index 9d236c7..bc4ddf4 100644 --- a/tests/load/outboxx/Dockerfile +++ b/tests/load/outboxx/Dockerfile @@ -12,7 +12,7 @@ RUN nix develop --command echo "deps cached" # -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 -Dlog_level=debug +RUN nix develop --command zig build -Doptimize=ReleaseFast -Dlog_level=info WORKDIR /app EXPOSE 9464 diff --git a/tests/load/outboxx/config.toml b/tests/load/outboxx/config.toml index a014ee0..f7fc3dd 100644 --- a/tests/load/outboxx/config.toml +++ b/tests/load/outboxx/config.toml @@ -8,6 +8,9 @@ type = "postgres" connection_env = "POSTGRES_URL" slot_name = "outboxx_benchmark_slot" publication_name = "outboxx_benchmark_publication" +# Never advance the slot, so `make load` once then `make start-outboxx` repeatedly +# replays the same backlog from the start on every run. Not for production. +confirm_lsn = false [sink] type = "kafka"