Check the slot's database and plugin before reusing it

src/source/postgres/integration_test.zig

@@ -816,3 +816,34 @@ test "Streaming source: created slot captures the consistent point and streams f
     // The captured point is a valid start LSN: streaming begins there.
     try source.startReplication(start_lsn);
 }
+
+test "Slot check: a same-named slot with another output plugin is rejected" {
+    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 slot_name = try std.fmt.allocPrint(allocator, "slot_mismatch_{d}_{d}", .{ timestamp, random_suffix });
+    defer allocator.free(slot_name);
+
+    const setup_conn = try createSetupConnection(allocator);
+    defer c.PQfinish(setup_conn);
+
+    const create_slot_sql = try test_helpers.formatSqlZ(allocator, "SELECT pg_create_logical_replication_slot('{s}', 'test_decoding')", .{slot_name});
+    defer allocator.free(create_slot_sql);
+    try execSQL(setup_conn, create_slot_sql);
+
+    const drop_slot_sql = try test_helpers.formatSqlZ(allocator, "SELECT pg_drop_replication_slot('{s}')", .{slot_name});
+    defer allocator.free(drop_slot_sql);
+    defer execSQL(setup_conn, drop_slot_sql) catch {};
+
+    const conn_str = try getTestConnectionString(allocator);
+    defer allocator.free(conn_str);
+
+    var protocol = ReplicationProtocol.init(allocator, slot_name, "unused_pub");
+    defer protocol.deinit();
+    try protocol.connect(conn_str);
+
+    try testing.expectError(error.SlotMismatch, protocol.slotExists());
+}

src/source/postgres/replication_protocol.zig

@@ -1,8 +1,13 @@
 const std = @import("std");
 const c = @import("c"); // C bindings (build-system translate-c)
 
+// The logical decoding plugin this source speaks: slots are created with it and
+// an existing slot is only usable if it runs it.
+const OUTPUT_PLUGIN = "pgoutput";
+
 pub const ReplicationError = error{
     ConnectionFailed,
+    SlotMismatch,
     StartReplicationFailed,
     ReceiveFailed,
     SendFeedbackFailed,
@@ -39,6 +44,7 @@ pub const StandbyStatusUpdate = struct {
     wal_write_position: u64,
     wal_flush_position: u64,
     wal_apply_position: u64,
+    /// Microseconds since 2000-01-01, the epoch the protocol uses on the wire.
     client_time: i64,
     reply_requested: bool,
 };
@@ -213,12 +219,46 @@ pub const ReplicationProtocol = struct {
         try self.execSimple(sql.ptr, "DROP PUBLICATION (snapshot marker)");
     }
 
-    /// Whether the replication slot already exists.
+    /// Whether the replication slot already exists and is usable by this run.
+    /// Slot names are global to the cluster, so a same-named slot can belong to
+    /// another database or another output plugin; that one is rejected here with
+    /// SlotMismatch, because START_REPLICATION would otherwise fail on every start
+    /// and turn a naming clash into a restart loop.
     pub fn slotExists(self: *Self) ReplicationError!bool {
         if (self.connection == null) return ReplicationError.ConnectionFailed;
         const slot_name = try self.lowerName(self.slot_name);
         defer self.allocator.free(slot_name);
-        return self.objectExists("pg_replication_slots", "slot_name", slot_name);
+
+        const sql = std.fmt.allocPrintSentinel(
+            self.allocator,
+            "SELECT database, plugin, current_database() FROM pg_replication_slots WHERE slot_name = '{s}'",
+            .{slot_name},
+            0,
+        ) catch return ReplicationError.OutOfMemory;
+        defer self.allocator.free(sql);
+
+        const result = c.PQexec(self.connection, sql.ptr);
+        defer c.PQclear(result);
+        if (c.PQresultStatus(result) != c.PGRES_TUPLES_OK) {
+            std.log.warn("Slot check failed: {s}", .{c.PQresultErrorMessage(result)});
+            return ReplicationError.ConnectionFailed;
+        }
+        if (c.PQntuples(result) == 0) return false;
+
+        // A physical slot reports both columns as NULL.
+        const database = columnText(result, 0) orelse "(none)";
+        const plugin = columnText(result, 1) orelse "(none)";
+        const current_database = columnText(result, 2) orelse "";
+
+        if (!std.mem.eql(u8, database, current_database) or !std.mem.eql(u8, plugin, OUTPUT_PLUGIN)) {
+            std.log.warn(
+                "Replication slot '{s}' already exists on database '{s}' with plugin '{s}'; this run needs database '{s}' with plugin '{s}'",
+                .{ slot_name, database, plugin, current_database, OUTPUT_PLUGIN },
+            );
+            return ReplicationError.SlotMismatch;
+        }
+
+        return true;
     }
 
     /// Create the replication slot and capture its consistent point and exported
@@ -240,7 +280,7 @@ pub const ReplicationProtocol = struct {
             self.snapshot_name = null;
         }
 
-        const sql = std.fmt.allocPrintSentinel(self.allocator, "CREATE_REPLICATION_SLOT {s} LOGICAL pgoutput", .{slot_name}, 0) catch return ReplicationError.OutOfMemory;
+        const sql = std.fmt.allocPrintSentinel(self.allocator, "CREATE_REPLICATION_SLOT {s} LOGICAL {s}", .{ slot_name, OUTPUT_PLUGIN }, 0) catch return ReplicationError.OutOfMemory;
         defer self.allocator.free(sql);
 
         std.log.info("Creating replication slot: {s}", .{slot_name});
@@ -325,6 +365,12 @@ pub const ReplicationProtocol = struct {
         return c.PQntuples(result) > 0;
     }
 
+    // Text of a column in the first row, null when the value is SQL NULL.
+    fn columnText(result: ?*c.PGresult, column: c_int) ?[]const u8 {
+        if (c.PQgetisnull(result, 0, column) != 0) return null;
+        return std.mem.span(c.PQgetvalue(result, 0, column));
+    }
+
     // Run a command that returns no rows (CREATE/DROP), failing on a non-OK status.
     // Accepts TUPLES_OK too, since some replication commands report a result set.
     fn execSimple(self: *Self, sql: [*:0]const u8, ctx: []const u8) ReplicationError!void {

src/source/postgres/source.zig

@@ -61,6 +61,9 @@ pub const Batch = struct {
 
 pub const PostgresSourceError = error{
     ConnectionFailed,
+    // A slot with the configured name exists but belongs to another database or
+    // output plugin, so this run must not use it (see slotExists).
+    SlotMismatch,
     ReplicationFailed,
     DecodeFailed,
     ConversionFailed,
@@ -176,7 +179,10 @@ pub const PostgresSource = struct {
     // "no slot, marker present", which reads as a fresh bootstrap, never a false
     // "completed".
     fn ensureSlot(self: *Self, bootstrap: Bootstrap) PostgresSourceError!void {
-        const slot_exists = self.protocol.slotExists() catch return PostgresSourceError.ConnectionFailed;
+        const slot_exists = self.protocol.slotExists() catch |err| switch (err) {
+            error.SlotMismatch => return PostgresSourceError.SlotMismatch,
+            else => return PostgresSourceError.ConnectionFailed,
+        };
 
         if (bootstrap == .stream_only) {
             self.protocol.dropSnapshotMarker() catch return PostgresSourceError.ConnectionFailed;
@@ -357,7 +363,13 @@ pub const PostgresSource = struct {
 
             // Step 2: DRAIN all buffered messages (non-blocking)
             while (changes.items.len < limit) {
-                const next_msg = self.protocol.receiveMessage(io, 0) catch break; // 0ms wait time = non-blocking
+                // 0ms wait time = non-blocking. A failure here is the same broken
+                // connection the blocking receive above reports, so it propagates
+                // instead of ending the batch as if the buffer had run dry.
+                const next_msg = self.protocol.receiveMessage(io, 0) catch |err| {
+                    std.log.warn("Failed to drain buffered replication message: {}", .{err});
+                    return PostgresSourceError.ReplicationFailed;
+                };
                 if (next_msg == null) break; // No more buffered data
 
                 var buffered_msg = next_msg.?;
@@ -450,7 +462,7 @@ pub const PostgresSource = struct {
             .wal_write_position = lsn,
             .wal_flush_position = lsn,
             .wal_apply_position = lsn,
-            .client_time = std.Io.Timestamp.now(io, .real).toSeconds(),
+            .client_time = std.Io.Timestamp.now(io, .real).toMicroseconds() - POSTGRES_EPOCH_UNIX_SECONDS * std.time.us_per_s,
             // Request an immediate keepalive back. Our regular feedback keeps the
             // walsender quiet (it only probes after wal_sender_timeout/2 of client
             // silence), so an idle stream would carry no inbound traffic and trip