From 3f3e124530cf9e119d6540a134389df7a7d5843e Mon Sep 17 00:00:00 2001 From: dand Date: Tue, 7 Jul 2026 15:37:41 -0400 Subject: [PATCH] Isolate integration replication state --- pkg/stream/integration/helper_test.go | 77 +++++++++++++++++-- .../integration/pg_kafka_integration_test.go | 2 +- .../pg_pg_batch_coalesce_integration_test.go | 6 +- .../integration/pg_pg_integration_test.go | 12 +-- .../pg_pg_integration_transformer_test.go | 2 +- .../integration/pg_search_integration_test.go | 6 +- .../pg_webhook_integration_test.go | 2 +- .../snapshot_pg_integration_test.go | 12 +-- 8 files changed, 90 insertions(+), 29 deletions(-) diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go index 5c11f1f9..af365402 100644 --- a/pkg/stream/integration/helper_test.go +++ b/pkg/stream/integration/helper_test.go @@ -10,6 +10,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "sync/atomic" "testing" "time" @@ -50,6 +51,8 @@ const ( withGeneratedColumn = true ) +var integrationReplicationSlotCounter uint64 + type mockProcessor struct { eventChan chan *wal.Event skipEventFn func(event *wal.Event) bool @@ -93,19 +96,39 @@ func (m *mockWebhookServer) close() { } func runStream(t *testing.T, ctx context.Context, cfg *stream.Config) { - // start the configured stream listener/processor + t.Helper() + + done := make(chan error, 1) go func() { - err := stream.Run(ctx, testLogger(), cfg, false, nil) - require.NoError(t, err) + done <- stream.Run(ctx, testLogger(), cfg, false, nil) }() + + t.Cleanup(func() { + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Error("timeout waiting for stream to stop") + } + }) } func runSnapshot(t *testing.T, ctx context.Context, cfg *stream.Config) { - // start the configured stream listener/processor + t.Helper() + + done := make(chan error, 1) go func() { - err := stream.Snapshot(ctx, testLogger(), cfg, nil) - require.NoError(t, err) + done <- stream.Snapshot(ctx, testLogger(), cfg, nil) }() + + t.Cleanup(func() { + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Error("timeout waiting for snapshot to stop") + } + }) } func initStream(t *testing.T, ctx context.Context, url string) { @@ -135,17 +158,55 @@ func testLogger() loglib.Logger { })) } -func testPostgresListenerCfg() stream.ListenerConfig { +func testPostgresListenerCfg(t *testing.T) stream.ListenerConfig { + t.Helper() + + slotName := initReplicationSlotForTest(t, pgurl) return stream.ListenerConfig{ Postgres: &stream.PostgresListenerConfig{ URL: pgurl, Replication: pgreplication.Config{ - PostgresURL: pgurl, + PostgresURL: pgurl, + ReplicationSlotName: slotName, }, }, } } +func initReplicationSlotForTest(t *testing.T, url string) string { + t.Helper() + + slotName := fmt.Sprintf("pgstream_it_%d_%d", time.Now().UnixNano(), atomic.AddUint64(&integrationReplicationSlotCounter, 1)) + require.NoError(t, stream.Init(context.Background(), &stream.InitConfig{ + PostgresURL: url, + ReplicationSlotName: slotName, + })) + + t.Cleanup(func() { + dropReplicationSlotForTest(t, url, slotName) + }) + + return slotName +} + +func dropReplicationSlotForTest(t *testing.T, url, slotName string) { + t.Helper() + + require.Eventually(t, func() bool { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + conn, err := pglib.NewConn(ctx, url) + if err != nil { + return false + } + defer conn.Close(ctx) + + _, err = conn.Exec(ctx, `SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1`, slotName) + return err == nil + }, 10*time.Second, 200*time.Millisecond, "replication slot %s was not dropped", slotName) +} + func testPostgresListenerCfgWithSnapshot(sourceURL, targetURL string, tables []string) stream.ListenerConfig { return stream.ListenerConfig{ Postgres: &stream.PostgresListenerConfig{ diff --git a/pkg/stream/integration/pg_kafka_integration_test.go b/pkg/stream/integration/pg_kafka_integration_test.go index fd51b421..70f06b33 100644 --- a/pkg/stream/integration/pg_kafka_integration_test.go +++ b/pkg/stream/integration/pg_kafka_integration_test.go @@ -26,7 +26,7 @@ func Test_PostgresToKafka(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testKafkaProcessorCfg(), } diff --git a/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go b/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go index 64084e13..cd7032da 100644 --- a/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go +++ b/pkg/stream/integration/pg_pg_batch_coalesce_integration_test.go @@ -27,7 +27,7 @@ func Test_PostgresToPostgres_BatchCoalesce(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: stream.ProcessorConfig{ Postgres: &stream.PostgresProcessorConfig{ BatchWriter: postgres.Config{ @@ -195,7 +195,7 @@ func Test_PostgresToPostgres_BatchCoalesce_WithCompositeKey(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: stream.ProcessorConfig{ Postgres: &stream.PostgresProcessorConfig{ BatchWriter: postgres.Config{ @@ -270,7 +270,7 @@ func Test_PostgresToPostgres_BatchCoalesce_OnConflict(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: stream.ProcessorConfig{ Postgres: &stream.PostgresProcessorConfig{ BatchWriter: postgres.Config{ diff --git a/pkg/stream/integration/pg_pg_integration_test.go b/pkg/stream/integration/pg_pg_integration_test.go index ac4c70c5..7bff9758 100644 --- a/pkg/stream/integration/pg_pg_integration_test.go +++ b/pkg/stream/integration/pg_pg_integration_test.go @@ -39,7 +39,7 @@ func Test_PostgresToPostgres(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } @@ -315,7 +315,7 @@ func Test_PostgresToPostgres_IndexesAndConstraints(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } @@ -423,7 +423,7 @@ func Test_PostgresToPostgres_IdentityColumns(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } @@ -571,7 +571,7 @@ func Test_PostgresToPostgres_AlwaysIdentityUpdate(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } @@ -653,7 +653,7 @@ func Test_PostgresToPostgres_Sequences(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } @@ -962,7 +962,7 @@ func Test_PostgresToPostgres_LargeIntegerPrecisionWAL(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfg(), } diff --git a/pkg/stream/integration/pg_pg_integration_transformer_test.go b/pkg/stream/integration/pg_pg_integration_transformer_test.go index 52b68087..4d09b27c 100644 --- a/pkg/stream/integration/pg_pg_integration_transformer_test.go +++ b/pkg/stream/integration/pg_pg_integration_transformer_test.go @@ -53,7 +53,7 @@ func Test_PostgresToPostgres_Transformer(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testPostgresProcessorCfgWithTransformer(pgurl), } diff --git a/pkg/stream/integration/pg_search_integration_test.go b/pkg/stream/integration/pg_search_integration_test.go index 1da59153..63fa9b19 100644 --- a/pkg/stream/integration/pg_search_integration_test.go +++ b/pkg/stream/integration/pg_search_integration_test.go @@ -131,7 +131,7 @@ func Test_PostgresToSearch(t *testing.T) { t.Run("postgres to opensearch", func(t *testing.T) { cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testSearchProcessorCfg(store.Config{ OpenSearchURL: opensearchURL, }), @@ -145,7 +145,7 @@ func Test_PostgresToSearch(t *testing.T) { t.Run("postgres to elasticsearch", func(t *testing.T) { cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testSearchProcessorCfg(store.Config{ ElasticsearchURL: elasticsearchURL, }), @@ -162,7 +162,7 @@ func Test_PostgresToSearch(t *testing.T) { cfg := testSearchProcessorCfg(store.Config{ElasticsearchURL: elasticsearchURL}) cfg.Search.Indexer = search.IndexerConfig{HashDocIDs: true} - runStream(t, ctx, &stream.Config{Listener: testPostgresListenerCfg(), Processor: cfg}) + runStream(t, ctx, &stream.Config{Listener: testPostgresListenerCfg(t), Processor: cfg}) client, err := elasticsearch.NewClient(elasticsearchURL) require.NoError(t, err) diff --git a/pkg/stream/integration/pg_webhook_integration_test.go b/pkg/stream/integration/pg_webhook_integration_test.go index 085cd37b..54c45bd1 100644 --- a/pkg/stream/integration/pg_webhook_integration_test.go +++ b/pkg/stream/integration/pg_webhook_integration_test.go @@ -26,7 +26,7 @@ func Test_PostgresToWebhook(t *testing.T) { } cfg := &stream.Config{ - Listener: testPostgresListenerCfg(), + Listener: testPostgresListenerCfg(t), Processor: testWebhookProcessorCfg(), } diff --git a/pkg/stream/integration/snapshot_pg_integration_test.go b/pkg/stream/integration/snapshot_pg_integration_test.go index c443cba3..99ce61ad 100644 --- a/pkg/stream/integration/snapshot_pg_integration_test.go +++ b/pkg/stream/integration/snapshot_pg_integration_test.go @@ -39,7 +39,7 @@ func Test_SnapshotToPostgres(t *testing.T) { execQueryWithURL(t, ctx, snapshotPGURL, fmt.Sprintf("grant select on %s to test_role_%s", testTable, testTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{testTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL) @@ -204,7 +204,7 @@ func Test_SnapshotToPostgres_IdentityOnlyTable(t *testing.T) { `INSERT INTO %s(id, parent_id, name) VALUES (1, 100, 'a'),(2, 200, 'b'),(3, 300, 'c')`, childTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{parentTable, childTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL) @@ -306,7 +306,7 @@ func Test_SnapshotToPostgres_CubeColumns(t *testing.T) { testTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{testTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL) @@ -421,7 +421,7 @@ func Test_SnapshotToPostgres_LtreeColumns(t *testing.T) { testTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{testTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL) @@ -706,7 +706,7 @@ func Test_SnapshotToPostgres_IdentityAndGeneratedColumns(t *testing.T) { `INSERT INTO %s(id, name) VALUES (10, 'alpha'),(20, 'beta'),(30, 'gamma')`, testTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{testTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL) @@ -797,7 +797,7 @@ func Test_SnapshotToPostgres_LargeIntegerPrecision(t *testing.T) { testTable)) cfg := &stream.Config{ - Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{"*.*"}), + Listener: testPostgresListenerCfgWithSnapshot(snapshotPGURL, targetPGURL, []string{testTable}), Processor: testPostgresProcessorCfg(opts...), } initStream(t, ctx, snapshotPGURL)