fix: lint and test

Author: Amogh-Bharadwaj

flow/e2e/pg_schema_dump_test.go

@@ -220,13 +220,13 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_And_CDC() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load parent", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + dstParent).Scan(&count)
+			"SELECT COUNT(*) FROM "+dstParent).Scan(&count)
 		return err == nil && count == 5
 	})
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load child", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + dstChild).Scan(&count)
+			"SELECT COUNT(*) FROM "+dstChild).Scan(&count)
 		return err == nil && count == 10
 	})
 
@@ -306,29 +306,29 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_And_CDC() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc parent rows", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + dstParent).Scan(&count)
+			"SELECT COUNT(*) FROM "+dstParent).Scan(&count)
 		return err == nil && count == 8
 	})
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc child rows", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + dstChild).Scan(&count)
+			"SELECT COUNT(*) FROM "+dstChild).Scan(&count)
 		return err == nil && count == 15
 	})
 
 	// verify data integrity: compare actual row content
 	// query source and destination and compare
 	var srcParentCount, dstParentCount int64
-	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + srcParent).Scan(&srcParentCount)
+	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM "+srcParent).Scan(&srcParentCount)
 	require.NoError(s.t, err)
-	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + dstParent).Scan(&dstParentCount)
+	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM "+dstParent).Scan(&dstParentCount)
 	require.NoError(s.t, err)
 	require.Equal(s.t, srcParentCount, dstParentCount, "parent table row counts should match")
 
 	var srcChildCount, dstChildCount int64
-	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + srcChild).Scan(&srcChildCount)
+	err = srcConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM "+srcChild).Scan(&srcChildCount)
 	require.NoError(s.t, err)
-	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM " + dstChild).Scan(&dstChildCount)
+	err = dstConn.QueryRow(s.t.Context(), "SELECT COUNT(*) FROM "+dstChild).Scan(&dstChildCount)
 	require.NoError(s.t, err)
 	require.Equal(s.t, srcChildCount, dstChildCount, "child table row counts should match")
 
@@ -437,7 +437,7 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_No_Owner_No_Privileges() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "initial load owned_tbl", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + qualified).Scan(&count)
+			"SELECT COUNT(*) FROM "+qualified).Scan(&count)
 		return err == nil && count == 5
 	})
 
@@ -452,7 +452,7 @@ func (s PeerFlowE2ETestSuitePG) Test_PG_Schema_Dump_No_Owner_No_Privileges() {
 	EnvWaitFor(s.t, env, 3*time.Minute, "cdc owned_tbl", func() bool {
 		var count int64
 		err := dstConn.QueryRow(s.t.Context(),
-			"SELECT COUNT(*) FROM " + qualified).Scan(&count)
+			"SELECT COUNT(*) FROM "+qualified).Scan(&count)
 		return err == nil && count == 10
 	})
 

flow/e2e/test_utils.go

@@ -8,12 +8,14 @@ import (
 	"log/slog"
 	"slices"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/api/enums/v1"
 	"go.temporal.io/sdk/client"
@@ -724,11 +726,26 @@ func (env WorkflowRun) Query(ctx context.Context, queryType string, args ...any)
 	return env.c.QueryWorkflow(ctx, env.GetID(), "", queryType, args...)
 }
 
+// catalogTestAccessPool is a pgxpool with application_name="catalog_test_access"
+// so test-side catalog reads are not collateral damage when other tests issue
+// pg_terminate_backend WHERE application_name='peerdb' (see api_test.go).
+var catalogTestAccessPool = sync.OnceValues(func() (*pgxpool.Pool, error) {
+	ctx := context.Background()
+	connStr := internal.GetCatalogConnectionStringFromEnv(ctx)
+	cfg, err := pgxpool.ParseConfig(connStr)
+	if err != nil {
+		return nil, err
+	}
+	cfg.ConnConfig.RuntimeParams["application_name"] = "catalog_test_access"
+	cfg.MaxConns = 3
+	return pgxpool.NewWithConfig(ctx, cfg)
+})
+
 func (env WorkflowRun) GetFlowStatus(t *testing.T) protos.FlowStatus {
 	t.Helper()
-	pool, err := internal.GetCatalogConnectionPoolFromEnv(t.Context())
+	pool, err := catalogTestAccessPool()
 	EnvNoError(t, env, err)
-	status, err := internal.GetWorkflowStatus(t.Context(), pool, env.GetID())
+	status, err := internal.GetWorkflowStatus(t.Context(), shared.CatalogPool{Pool: pool}, env.GetID())
 	EnvNoError(t, env, err)
 	return status
 }

flow/internal/dynamicconf.go

@@ -445,8 +445,9 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		TargetForSetting: protos.DynconfTarget_ALL,
 	},
 	{
-		Name:             "PEERDB_PG_AUTOMATED_SCHEMA_DUMP",
-		Description:      "For PG-to-PG mirrors, run pg_dump --schema-only from source into psql on destination during setup so destination schema/tables/indexes match the source.",
+		Name: "PEERDB_PG_AUTOMATED_SCHEMA_DUMP",
+		Description: "For PG-to-PG mirrors, run pg_dump --schema-only from source into psql on destination " +
+			"during setup so destination schema/tables/indexes match the source.",
 		DefaultValue:     "false",
 		ValueType:        protos.DynconfValueType_BOOL,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,