fix: Resolve nightly test timeout in shared/pkg/saga (#693)

Two issues caused the nightly "Slow Integration Tests" job to fail:

1. Container proliferation timeout: ~60 integration tests each created
   their own CockroachDB testcontainer (~15s startup each), exhausting
   the 15-minute per-package timeout. Fixed by sharing a single container
   via TestMain with sync.Once lazy initialization and per-test table
   cleanup via DELETE.

2. FOR UPDATE SKIP LOCKED in timeout_worker.go: CockroachDB's serializable
   isolation silently returns empty results for SKIP LOCKED, preventing
   ProcessExpiredSuspensions from finding expired sagas. Fixed by removing
   SKIP LOCKED and splitting the CTE+UPDATE into separate SELECT and UPDATE
   steps (CockroachDB requires FOR UPDATE at the top level of SELECT).

   This is the same class of issue fixed in claiming.go by PR #689.

Also guards RunSagaMigrations against re-invocation on CockroachDB, where
GORM's AutoMigrate fails with SQLSTATE 42704 when checking existing
unique constraints (naming mismatch between GORM and CockroachDB).

Test results: saga package runs in ~22s (previously >15m).

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

shared/pkg/saga/migrations.go

@@ -23,10 +23,17 @@ import (
 //	    }
 //	}
 func RunSagaMigrations(db *gorm.DB) error {
-	// Run AutoMigrate for the saga models
-	// This creates the tables and standard indexes defined in struct tags
-	if err := db.AutoMigrate(&SagaInstance{}, &SagaStepResult{}); err != nil {
-		return fmt.Errorf("failed to auto-migrate saga models: %w", err)
+	// Only run AutoMigrate when tables don't yet exist.
+	// GORM's AutoMigrate is not idempotent on CockroachDB: re-running it
+	// fails with SQLSTATE 42704 because CockroachDB names unique constraints
+	// differently than what GORM expects when checking existing schema.
+	// The partial indexes and composite constraint below are already
+	// idempotent (IF NOT EXISTS / information_schema checks).
+	migrator := db.Migrator()
+	if !migrator.HasTable(&SagaInstance{}) || !migrator.HasTable(&SagaStepResult{}) {
+		if err := db.AutoMigrate(&SagaInstance{}, &SagaStepResult{}); err != nil {
+			return fmt.Errorf("failed to auto-migrate saga models: %w", err)
+		}
 	}
 
 	// Create the partial index for orphan detection

shared/pkg/saga/test_helpers_test.go

@@ -1,15 +1,114 @@
 package saga
 
 import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
 	"testing"
+	"time"
 
-	"github.com/meridianhub/meridian/shared/platform/testdb"
+	"github.com/testcontainers/testcontainers-go/modules/cockroachdb"
+	gormpg "gorm.io/driver/postgres"
 	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
 )
 
-// setupTestPostgres creates a CockroachDB testcontainer for integration testing.
-// Named setupTestPostgres for historical compatibility - CockroachDB is wire-compatible
-// with PostgreSQL and is our production database.
+// sharedDB holds the shared CockroachDB connection for all integration tests.
+// Lazily initialized on first use via sync.Once to avoid starting the container
+// when only unit tests run (e.g. -short mode).
+var (
+	sharedDB      *gorm.DB
+	sharedOnce    sync.Once
+	sharedInitErr error
+	sharedCleanup func()
+)
+
+func TestMain(m *testing.M) {
+	code := m.Run()
+
+	if sharedCleanup != nil {
+		sharedCleanup()
+	}
+	os.Exit(code)
+}
+
+func initSharedContainer() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	crdbContainer, err := cockroachdb.Run(ctx,
+		"cockroachdb/cockroach:v24.3.0",
+		cockroachdb.WithDatabase("test_db"),
+		cockroachdb.WithUser("root"),
+		cockroachdb.WithInsecure(),
+	)
+	if err != nil {
+		return fmt.Errorf("start container: %w", err)
+	}
+
+	connConfig, err := crdbContainer.ConnectionConfig(ctx)
+	if err != nil {
+		_ = crdbContainer.Terminate(ctx)
+		return fmt.Errorf("connection config: %w", err)
+	}
+
+	db, err := gorm.Open(gormpg.Open(connConfig.ConnString()), &gorm.Config{
+		Logger: logger.Default.LogMode(logger.Silent),
+	})
+	if err != nil {
+		_ = crdbContainer.Terminate(ctx)
+		return fmt.Errorf("gorm open: %w", err)
+	}
+
+	// Run migrations once — subsequent RunSagaMigrations calls in tests
+	// are idempotent no-ops (tables/indexes already exist).
+	if err := RunSagaMigrations(db); err != nil {
+		_ = crdbContainer.Terminate(ctx)
+		return fmt.Errorf("migrations: %w", err)
+	}
+
+	sharedDB = db
+	sharedCleanup = func() {
+		cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cleanupCancel()
+
+		sqlDB, _ := db.DB()
+		if sqlDB != nil {
+			_ = sqlDB.Close()
+		}
+		_ = crdbContainer.Terminate(cleanupCtx)
+	}
+
+	return nil
+}
+
+// setupTestPostgres returns the shared CockroachDB connection with per-test
+// data cleanup. Named setupTestPostgres for historical compatibility —
+// CockroachDB is wire-compatible with PostgreSQL and is our production database.
+//
+// Previously each test started its own CockroachDB container (~15s startup).
+// With ~60 integration tests the package exceeded the 15-minute timeout.
+// Sharing one container reduces total overhead from ~15 minutes to ~15 seconds.
 func setupTestPostgres(t *testing.T) (*gorm.DB, func()) {
-	return testdb.SetupCockroachDB(t, nil)
+	t.Helper()
+
+	sharedOnce.Do(func() {
+		sharedInitErr = initSharedContainer()
+	})
+	if sharedInitErr != nil {
+		t.Fatalf("shared CockroachDB setup failed: %v", sharedInitErr)
+	}
+
+	// Clean tables before each test (FK order: children first).
+	if err := sharedDB.Exec("DELETE FROM saga_step_results").Error; err != nil {
+		t.Fatalf("Failed to clean saga_step_results: %v", err)
+	}
+	if err := sharedDB.Exec("DELETE FROM saga_instances").Error; err != nil {
+		t.Fatalf("Failed to clean saga_instances: %v", err)
+	}
+
+	return sharedDB, func() {
+		// No-op: container lifecycle managed by TestMain.
+	}
 }

shared/pkg/saga/timeout_worker.go

@@ -107,44 +107,51 @@ func (w *TimeoutWorker) processExpiredSuspensions(ctx context.Context) error {
 	var expired []expiredSaga
 
 	err := w.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		// Find and transition expired suspensions atomically
-		// Use CTE to capture idempotency_key before clearing suspend_data
-		// The JSONB query extracts the timeout_at field and compares with NOW()
-		result := tx.Raw(`
-			WITH expired_sagas AS (
-				SELECT id, correlation_id, suspend_data->>'idempotency_key' as idempotency_key
-				FROM saga_instances
-				WHERE status = ?
-				  AND (suspend_data->>'timeout_at')::timestamptz < ?
-				FOR UPDATE SKIP LOCKED
-				LIMIT ?
-			)
-			UPDATE saga_instances si
-			SET
-				status = ?,
-				error_message = ?,
-				error_category = ?,
-				updated_at = ?,
-				suspend_reason = NULL,
-				suspend_data = NULL
-			FROM expired_sagas es
-			WHERE si.id = es.id
-			RETURNING si.id, es.correlation_id, es.idempotency_key
+		// Step 1: Find expired suspensions with row-level locking.
+		// CockroachDB requires FOR UPDATE at the top level of a SELECT (not in
+		// CTEs or subqueries). SKIP LOCKED is omitted because CockroachDB's
+		// serializable isolation silently returns empty results when it's used.
+		// Serializable isolation provides equivalent concurrency safety.
+		if err := tx.Raw(`
+			SELECT id, correlation_id, suspend_data->>'idempotency_key' as idempotency_key
+			FROM saga_instances
+			WHERE status = ?
+			  AND (suspend_data->>'timeout_at')::timestamptz < ?
+			ORDER BY id
+			FOR UPDATE
+			LIMIT ?
 		`,
 			SagaStatusWaitingForEvent,
 			now,
 			w.config.BatchSize,
-			SagaStatusFailed,
-			"Suspend timeout exceeded - external event not received within deadline",
-			string(ErrorCategoryFatal),
-			now,
-		).Scan(&expired)
+		).Scan(&expired).Error; err != nil {
+			return fmt.Errorf("failed to find expired suspensions: %w", err)
+		}
 
-		if result.Error != nil {
-			return result.Error
+		if len(expired) == 0 {
+			return nil
+		}
+
+		// Step 2: Transition expired sagas to FAILED.
+		// Capture idempotency_key in step 1 before clearing suspend_data here.
+		ids := make([]uuid.UUID, len(expired))
+		for i, s := range expired {
+			ids[i] = s.ID
+		}
+		if err := tx.Model(&SagaInstance{}).
+			Where("id IN ?", ids).
+			Updates(map[string]interface{}{
+				"status":         SagaStatusFailed,
+				"error_message":  "Suspend timeout exceeded - external event not received within deadline",
+				"error_category": string(ErrorCategoryFatal),
+				"updated_at":     now,
+				"suspend_reason": nil,
+				"suspend_data":   nil,
+			}).Error; err != nil {
+			return fmt.Errorf("failed to update expired sagas: %w", err)
 		}
 
-		// Also update the corresponding step results to FAILED
+		// Step 3: Update the corresponding step results to FAILED.
 		for _, saga := range expired {
 			stepUpdate := tx.Model(&SagaStepResult{}).
 				Where("saga_instance_id = ? AND status = ?", saga.ID, StepStatusSuspended).