Fix flaky snapshot+stream race in integration test

The test compared the stream output against an atomic counter that races
with the actual database commits. The writer goroutine does INSERT then
count.Add(1), but Stop() can land between those two operations, causing
the count to be off by one from what the database (and thus the stream)
actually sees.

Fix: read the committed row count from the database after stopping the
writer, instead of trusting the in-process atomic counter.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 2 people

internal/impl/postgresql/integration_test.go

@@ -916,11 +916,9 @@ read_until:
 	}))
 
 	// Continuously write so there is a chance we skip data between snapshot and stream hand off.
-	var count atomic.Int64
 	writer := asyncroutine.NewPeriodic(time.Microsecond, func() {
 		_, err := db.Exec("INSERT INTO seq DEFAULT VALUES")
 		require.NoError(t, err)
-		count.Add(1)
 	})
 	writer.Start()
 	t.Cleanup(writer.Stop)
@@ -948,8 +946,15 @@ read_until:
 		require.Fail(t, "stream did not complete in time")
 	}
 	require.NoError(t, streamOut.StopWithin(10*time.Second))
+
+	// Read the actual committed count from the database rather than
+	// relying on the atomic counter, which can race with the last
+	// INSERT commit.
+	var dbCount int64
+	require.NoError(t, db.QueryRow("SELECT COUNT(*) FROM seq").Scan(&dbCount))
+
 	expected := []int64{}
-	for i := range count.Load() {
+	for i := range dbCount {
 		expected = append(expected, i+1)
 	}
 	batchMu.Lock()

internal/impl/postgresql/pglogicalstream/pgtype_compat.go

@@ -10,8 +10,9 @@ package pglogicalstream
 
 import "strings"
 
-// sanitizeTsrange strips escaped quotes from Postgres tsrange text
-// representations. Postgres may return ranges like:
+// sanitizeTsrange strips quoting from Postgres tsrange text representations.
+//
+// Postgres quotes range bounds containing spaces, producing:
 //
 //	["2024-01-01 00:00:00","2024-12-31 00:00:00")
 //
@@ -20,7 +21,15 @@ import "strings"
 //
 //	[2024-01-01 00:00:00,2024-12-31 00:00:00)
 //
-// This function replicates that behavior without the old pgtype dependency.
+// This function replicates that behavior by stripping all double quotes.
+// This is safe for tsrange because timestamp bound values never contain
+// literal double quotes — they consist only of digits, dashes, colons,
+// spaces, and decimal points.
+//
+// NOTE: This function is NOT suitable for arbitrary range types whose
+// bound values may contain literal double quotes (e.g. text ranges).
+// For such types, a proper range parser that handles quoting and escaping
+// (like the old pgtype.ParseUntypedTextRange) would be needed.
 func sanitizeTsrange(s string) string {
 	return strings.ReplaceAll(s, `"`, "")
 }

internal/impl/postgresql/pglogicalstream/snapshotter.go

@@ -253,6 +253,7 @@ type tuple struct {
 	elements []any
 }
 
+//nolint:stylecheck // This is implementing the squirrel.Sqlizer interface
 func (t *tuple) ToSql() (sql string, args []any, err error) {
 	sql = "(" + strings.Join(slices.Repeat([]string{"?"}, len(t.elements)), ", ") + ")"
 	args = t.elements