flow ci: drop temporal's sdk/testsuite in favor of actual integration testing

.github/workflows/flow.yml

@@ -12,22 +12,17 @@ jobs:
       matrix:
         runner: [ubicloud-standard-16-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
-    timeout-minutes: 30
+    timeout-minutes: 40
     services:
-      pg_cdc:
+      catalog:
         image: imresamu/postgis:15-3.4-alpine
         ports:
-          - 7132:5432
+          - 5432:5432
         env:
-          POSTGRES_USER: postgres
+          PGUSER: postgres
           POSTGRES_PASSWORD: postgres
           POSTGRES_DB: postgres
-        options: >-
-          --name pg_cdc
-          --health-cmd pg_isready
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
+
     steps:
       - name: checkout sources
         uses: actions/checkout@v4
@@ -43,18 +38,12 @@ jobs:
           go-version: "1.22"
           cache-dependency-path: flow/go.sum
 
-      - name: install gotestsum
-        run: |
-          go install gotest.tools/gotestsum@latest
-
       - name: install lib-geos
         run: |
           sudo apt-get update
           sudo apt-get install libgeos-dev
 
-      - name: download go modules
-        run: |
-          go mod download
+      - run: go mod download
         working-directory: ./flow
 
       - name: setup gcp service account
@@ -87,21 +76,27 @@ jobs:
 
       - name: create hstore extension, increase logical replication limits, and setup catalog database
         run: >
-          docker exec pg_cdc psql -h localhost -p 5432 -U postgres -c "CREATE EXTENSION hstore;"
+          docker exec "${{ job.services.catalog.id }}" psql -U postgres -c "CREATE EXTENSION hstore;"
           -c "ALTER SYSTEM SET wal_level=logical;"
           -c "ALTER SYSTEM SET max_replication_slots=192;"
           -c "ALTER SYSTEM SET max_wal_senders=256;"
           -c "ALTER SYSTEM SET max_connections=2048;" &&
-          (cat ../nexus/catalog/migrations/V{?,??}__* | docker exec -i pg_cdc psql -h localhost -p 5432 -U postgres) &&
-          docker restart pg_cdc
-        working-directory: ./flow
+          (cat ./nexus/catalog/migrations/V{?,??}__* | docker exec -i "${{ job.services.catalog.id }}" psql -U postgres) &&
+          docker restart "${{ job.services.catalog.id }}"
         env:
-          PG_CDC: empty
           PGPASSWORD: postgres
 
+      - name: Install Temporal CLI
+        uses: temporalio/setup-temporal@v0
+
       - name: run tests
         run: |
-          gotestsum --format testname -- -p 24 ./... -timeout 1200s
+          temporal server start-dev --namespace default --headless &
+          go build -ldflags="-s -w" -o peer-flow
+          temporal operator search-attribute create --name MirrorName --type Text --namespace default
+          ./peer-flow worker &
+          ./peer-flow snapshot-worker &
+          go test -p 32 ./... -timeout 1200s
         working-directory: ./flow
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
@@ -122,7 +117,7 @@ jobs:
           SQLSERVER_PASSWORD: ${{ secrets.SQLSERVER_PASSWORD }}
           SQLSERVER_DB: ${{ secrets.SQLSERVER_DB }}
           PEERDB_CATALOG_HOST: localhost
-          PEERDB_CATALOG_PORT: 7132
+          PEERDB_CATALOG_PORT: 5432
           PEERDB_CATALOG_USER: postgres
           PEERDB_CATALOG_PASSWORD: postgres
           PEERDB_CATALOG_DATABASE: postgres

flow/alerting/alerting.go

@@ -3,7 +3,6 @@ package alerting
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"log/slog"
 	"time"
@@ -51,14 +50,14 @@ func (a *Alerter) registerSendersFromPool(ctx context.Context) ([]*slackAlertSen
 }
 
 // doesn't take care of closing pool, needs to be done externally.
-func NewAlerter(catalogPool *pgxpool.Pool) (*Alerter, error) {
+func NewAlerter(catalogPool *pgxpool.Pool) *Alerter {
 	if catalogPool == nil {
-		return nil, errors.New("catalog pool is nil for Alerter")
+		panic("catalog pool is nil for Alerter")
 	}
 
 	return &Alerter{
 		catalogPool: catalogPool,
-	}, nil
+	}
 }
 
 func (a *Alerter) AlertIfSlotLag(ctx context.Context, peerName string, slotInfo *protos.SlotInfo) {

flow/cmd/api.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/cert.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"crypto/tls"

flow/cmd/handler.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/mirror_status.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/peer_data.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/snapshot_worker.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"
@@ -25,7 +25,7 @@ type SnapshotWorkerOptions struct {
 	TemporalKey       string
 }
 
-func SnapshotWorkerMain(opts *SnapshotWorkerOptions) error {
+func SnapshotWorkerMain(opts *SnapshotWorkerOptions) (client.Client, worker.Worker, error) {
 	clientOptions := client.Options{
 		HostPort:  opts.TemporalHostPort,
 		Namespace: opts.TemporalNamespace,
@@ -35,7 +35,7 @@ func SnapshotWorkerMain(opts *SnapshotWorkerOptions) error {
 	if opts.TemporalCert != "" && opts.TemporalKey != "" {
 		certs, err := Base64DecodeCertAndKey(opts.TemporalCert, opts.TemporalKey)
 		if err != nil {
-			return fmt.Errorf("unable to process certificate and key: %w", err)
+			return nil, nil, fmt.Errorf("unable to process certificate and key: %w", err)
 		}
 
 		connOptions := client.ConnectionOptions{
@@ -47,37 +47,29 @@ func SnapshotWorkerMain(opts *SnapshotWorkerOptions) error {
 		clientOptions.ConnectionOptions = connOptions
 	}
 
+	conn, err := utils.GetCatalogConnectionPoolFromEnv(context.Background())
+	if err != nil {
+		return nil, nil, fmt.Errorf("unable to create catalog connection pool: %w", err)
+	}
+
 	c, err := client.Dial(clientOptions)
 	if err != nil {
-		return fmt.Errorf("unable to create Temporal client: %w", err)
+		return nil, nil, fmt.Errorf("unable to create Temporal client: %w", err)
 	}
-	defer c.Close()
 
 	taskQueue := shared.GetPeerFlowTaskQueueName(shared.SnapshotFlowTaskQueue)
 	w := worker.New(c, taskQueue, worker.Options{
 		EnableSessionWorker: true,
+		OnFatalError: func(err error) {
+			slog.Error("Snapshot Worker failed", slog.Any("error", err))
+		},
 	})
 
-	conn, err := utils.GetCatalogConnectionPoolFromEnv(context.Background())
-	if err != nil {
-		return fmt.Errorf("unable to create catalog connection pool: %w", err)
-	}
-
-	alerter, err := alerting.NewAlerter(conn)
-	if err != nil {
-		return fmt.Errorf("unable to create alerter: %w", err)
-	}
-
 	w.RegisterWorkflow(peerflow.SnapshotFlowWorkflow)
 	w.RegisterActivity(&activities.SnapshotActivity{
 		SnapshotConnections: make(map[string]activities.SlotSnapshotSignal),
-		Alerter:             alerter,
+		Alerter:             alerting.NewAlerter(conn),
 	})
 
-	err = w.Run(worker.InterruptCh())
-	if err != nil {
-		return fmt.Errorf("worker run error: %w", err)
-	}
-
-	return nil
+	return c, w, nil
 }

flow/cmd/validate_mirror.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/validate_peer.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/version.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"

flow/cmd/worker.go

@@ -1,4 +1,4 @@
-package main
+package cmd
 
 import (
 	"context"
@@ -7,9 +7,7 @@ import (
 	"log"
 	"log/slog"
 	"os"
-	"os/signal"
 	"runtime"
-	"syscall"
 
 	"github.com/grafana/pyroscope-go"
 	"go.temporal.io/sdk/client"
@@ -74,22 +72,11 @@ func setupPyroscope(opts *WorkerOptions) {
 	}
 }
 
-func WorkerMain(opts *WorkerOptions) error {
+func WorkerMain(opts *WorkerOptions) (client.Client, worker.Worker, error) {
 	if opts.EnableProfiling {
 		setupPyroscope(opts)
 	}
 
-	go func() {
-		sigs := make(chan os.Signal, 1)
-		signal.Notify(sigs, syscall.SIGQUIT)
-		buf := make([]byte, 1<<20)
-		for {
-			<-sigs
-			stacklen := runtime.Stack(buf, true)
-			log.Printf("=== received SIGQUIT ===\n*** goroutine dump...\n%s\n*** end\n", buf[:stacklen])
-		}
-	}()
-
 	clientOptions := client.Options{
 		HostPort:  opts.TemporalHostPort,
 		Namespace: opts.TemporalNamespace,
@@ -100,7 +87,7 @@ func WorkerMain(opts *WorkerOptions) error {
 		slog.Info("Using temporal certificate/key for authentication")
 		certs, err := Base64DecodeCertAndKey(opts.TemporalCert, opts.TemporalKey)
 		if err != nil {
-			return fmt.Errorf("unable to process certificate and key: %w", err)
+			return nil, nil, fmt.Errorf("unable to process certificate and key: %w", err)
 		}
 		connOptions := client.ConnectionOptions{
 			TLS: &tls.Config{
@@ -113,37 +100,29 @@ func WorkerMain(opts *WorkerOptions) error {
 
 	conn, err := utils.GetCatalogConnectionPoolFromEnv(context.Background())
 	if err != nil {
-		return fmt.Errorf("unable to create catalog connection pool: %w", err)
+		return nil, nil, fmt.Errorf("unable to create catalog connection pool: %w", err)
 	}
 
 	c, err := client.Dial(clientOptions)
 	if err != nil {
-		return fmt.Errorf("unable to create Temporal client: %w", err)
+		return nil, nil, fmt.Errorf("unable to create Temporal client: %w", err)
 	}
 	slog.Info("Created temporal client")
-	defer c.Close()
 
 	taskQueue := shared.GetPeerFlowTaskQueueName(shared.PeerFlowTaskQueue)
 	w := worker.New(c, taskQueue, worker.Options{
 		EnableSessionWorker: true,
+		OnFatalError: func(err error) {
+			slog.Error("Peerflow Worker failed", slog.Any("error", err))
+		},
 	})
 	peerflow.RegisterFlowWorkerWorkflows(w)
 
-	alerter, err := alerting.NewAlerter(conn)
-	if err != nil {
-		return fmt.Errorf("unable to create alerter: %w", err)
-	}
-
 	w.RegisterActivity(&activities.FlowableActivity{
 		CatalogPool: conn,
-		Alerter:     alerter,
+		Alerter:     alerting.NewAlerter(conn),
 		CdcCache:    make(map[string]connectors.CDCPullConnector),
 	})
 
-	err = w.Run(worker.InterruptCh())
-	if err != nil {
-		return fmt.Errorf("worker run error: %w", err)
-	}
-
-	return nil
+	return c, w, nil
 }

flow/connectors/postgres/postgres_schema_delta_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/require"
 
+	"github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/e2eshared"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/model/qvalue"
@@ -24,13 +25,7 @@ type PostgresSchemaDeltaTestSuite struct {
 func SetupSuite(t *testing.T) PostgresSchemaDeltaTestSuite {
 	t.Helper()
 
-	connector, err := NewPostgresConnector(context.Background(), &protos.PostgresConfig{
-		Host:     "localhost",
-		Port:     7132,
-		User:     "postgres",
-		Password: "postgres",
-		Database: "postgres",
-	})
+	connector, err := NewPostgresConnector(context.Background(), utils.GetCatalogPostgresConfigFromEnv())
 	require.NoError(t, err)
 
 	setupTx, err := connector.conn.Begin(context.Background())

flow/connectors/postgres/qrep_bench_test.go

@@ -4,21 +4,14 @@ import (
 	"context"
 	"testing"
 
-	"github.com/PeerDB-io/peer-flow/generated/protos"
+	"github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 )
 
 func BenchmarkQRepQueryExecutor(b *testing.B) {
 	query := "SELECT * FROM bench.large_table"
 
 	ctx := context.Background()
-	connector, err := NewPostgresConnector(ctx,
-		&protos.PostgresConfig{
-			Host:     "localhost",
-			Port:     7132,
-			User:     "postgres",
-			Password: "postgres",
-			Database: "postgres",
-		})
+	connector, err := NewPostgresConnector(ctx, utils.GetCatalogPostgresConfigFromEnv())
 	if err != nil {
 		b.Fatalf("failed to create connection: %v", err)
 	}

flow/connectors/postgres/qrep_partition_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"go.temporal.io/sdk/log"
 
+	"github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
 	"github.com/PeerDB-io/peer-flow/shared"
 )
@@ -63,7 +64,7 @@ func newTestCaseForCTID(schema string, name string, rows uint32, expectedNum int
 }
 
 func TestGetQRepPartitions(t *testing.T) {
-	const connStr = "postgres://postgres:postgres@localhost:7132/postgres"
+	connStr := utils.GetCatalogConnectionStringFromEnv()
 
 	// Setup the DB
 	config, err := pgx.ParseConfig(connStr)