fix: retry embedded PostgreSQL launch when the probed port is taken

internal/postgres/embedded.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"sort"
 	"strings"
@@ -97,8 +98,33 @@ func DetectPostgresVersionAndExtensionsFromDB(host string, port int, database, u
 	return version, extensions, nil
 }
 
-// StartEmbeddedPostgres starts a temporary embedded PostgreSQL instance
+// embeddedStartAttempts bounds how many times StartEmbeddedPostgres tries to
+// launch the server. The port is reserved by probing it and closing the probe
+// socket before pg_ctl binds it, so another process (for example a parallel
+// test package starting its own instance) can take the port in between and
+// the launch fails with "could not start server". Each attempt picks a fresh
+// port and runtime directory.
+const embeddedStartAttempts = 3
+
+// StartEmbeddedPostgres starts a temporary embedded PostgreSQL instance.
 func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, error) {
+	var err error
+	for attempt := 1; attempt <= embeddedStartAttempts; attempt++ {
+		var embeddedPG *EmbeddedPostgres
+		embeddedPG, err = startEmbeddedPostgresOnce(config)
+		if err == nil {
+			return embeddedPG, nil
+		}
+		if attempt < embeddedStartAttempts {
+			logger.Get().Debug("embedded PostgreSQL failed to start, retrying", "attempt", attempt, "error", err)
+		}
+	}
+	return nil, err
+}
+
+// startEmbeddedPostgresOnce performs a single launch attempt on a fresh port
+// and runtime directory.
+func startEmbeddedPostgresOnce(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, error) {
 	// Create unique runtime path and schema name
 	tempSchema := GenerateTempSchemaName()
 	runtimePath := filepath.Join(os.TempDir(), tempSchema)
@@ -123,7 +149,7 @@ func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, e
 		StartParameters(map[string]string{
 			"logging_collector":          "off",       // Disable log collector
 			"log_destination":            "stderr",    // Send logs to stderr (which we discard)
-			"log_min_messages":           "PANIC",     // Only log PANIC level messages
+			"log_min_messages":           "LOG",       // Keep startup failures (e.g. port in use) in the pg_ctl output
 			"log_statement":              "none",      // Don't log SQL statements
 			"log_min_duration_statement": "-1",        // Don't log slow queries
 			"unix_socket_directories":    runtimePath, // Use a directory that is guaranteed to exist
@@ -133,6 +159,8 @@ func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, e
 	// Create and start PostgreSQL instance
 	instance := embeddedpostgres.NewDatabase(pgConfig)
 	if err := instance.Start(); err != nil {
+		stopOrphanedPostmaster(runtimePath)
+		os.RemoveAll(runtimePath)
 		return nil, fmt.Errorf("failed to start embedded PostgreSQL: %w", err)
 	}
 
@@ -389,8 +417,24 @@ func installUntilFixpoint(names []string, install func(name string) error) map[s
 	return lastErr
 }
 
-// findAvailablePort finds an available TCP port for PostgreSQL to use
-func findAvailablePort() (int, error) {
+// stopOrphanedPostmaster stops a postmaster that a failed launch may have left
+// running under runtimePath. When "pg_ctl start -w" gives up waiting, the
+// library returns an error but the server keeps running, and its own Stop
+// refuses to act because it never marked the instance as started. Best
+// effort: with no server running, pg_ctl stop fails fast and that is fine.
+func stopOrphanedPostmaster(runtimePath string) {
+	binDir := binariesPath
+	if binDir == "" {
+		binDir = runtimePath // embedded-postgres extracts the binaries here by default
+	}
+	pgCtl := filepath.Join(binDir, "bin", "pg_ctl")
+	_ = exec.Command(pgCtl, "stop", "-w", "-m", "immediate", "-D", filepath.Join(runtimePath, "data")).Run()
+}
+
+// findAvailablePort finds an available TCP port for PostgreSQL to use.
+// It is a variable so tests can hand StartEmbeddedPostgres a port that is
+// already taken and exercise the retry.
+var findAvailablePort = func() (int, error) {
 	listener, err := net.Listen("tcp", ":0")
 	if err != nil {
 		return 0, err

internal/postgres/embedded_test.go

@@ -0,0 +1,126 @@
+package postgres_test
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"strconv"
+	"testing"
+
+	"github.com/pgplex/pgschema/internal/postgres"
+	"github.com/pgplex/pgschema/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func testEmbeddedConfig(t *testing.T) *postgres.EmbeddedPostgresConfig {
+	t.Helper()
+	major := testutil.LatestPostgresVersion
+	if v := os.Getenv("PGSCHEMA_POSTGRES_VERSION"); v != "" {
+		var err error
+		major, err = strconv.Atoi(v)
+		require.NoError(t, err)
+	}
+	version, err := postgres.MapToEmbeddedPostgresVersion(major)
+	require.NoError(t, err)
+	return &postgres.EmbeddedPostgresConfig{
+		Version:  version,
+		Database: "testdb",
+		Username: "testuser",
+		Password: "testpass",
+	}
+}
+
+// holdPort returns a port that stays occupied for the rest of the test.
+func holdPort(t *testing.T) int {
+	t.Helper()
+	held, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { held.Close() })
+	return held.Addr().(*net.TCPAddr).Port
+}
+
+// stubFindAvailablePort replaces the port picker for the test and returns a
+// counter of how many times it was called.
+func stubFindAvailablePort(t *testing.T, pick func(call int, orig func() (int, error)) (int, error)) *int {
+	t.Helper()
+	orig := *postgres.FindAvailablePort
+	t.Cleanup(func() { *postgres.FindAvailablePort = orig })
+	calls := 0
+	*postgres.FindAvailablePort = func() (int, error) {
+		calls++
+		return pick(calls, orig)
+	}
+	return &calls
+}
+
+// TestStartEmbeddedPostgres_RetriesWhenPortIsTaken hands the first launch
+// attempt a port that is already in use, as happens when another process grabs
+// the probed port before pg_ctl binds it, and expects the next attempt to come
+// up on a fresh port instead of failing the whole start.
+func TestStartEmbeddedPostgres_RetriesWhenPortIsTaken(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	taken := holdPort(t)
+	calls := stubFindAvailablePort(t, func(call int, orig func() (int, error)) (int, error) {
+		if call == 1 {
+			return taken, nil
+		}
+		return orig()
+	})
+
+	embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
+	require.NoError(t, err, "second attempt should start on a fresh port")
+	defer embeddedPG.Stop()
+
+	_, port, _, _, _ := embeddedPG.GetConnectionDetails()
+	require.Equal(t, 2, *calls, "expected exactly one retry")
+	require.NotEqual(t, taken, port)
+}
+
+// TestStartEmbeddedPostgres_GivesUpAfterRetries keeps every attempt on the
+// taken port and expects the start to fail after the bounded number of
+// attempts rather than retry forever.
+func TestStartEmbeddedPostgres_GivesUpAfterRetries(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	taken := holdPort(t)
+	calls := stubFindAvailablePort(t, func(int, func() (int, error)) (int, error) {
+		return taken, nil
+	})
+
+	embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
+	require.Error(t, err)
+	require.Nil(t, embeddedPG)
+	require.Equal(t, 3, *calls)
+}
+
+// TestStopOrphanedPostmaster verifies the fallback stop finds pg_ctl under the
+// runtime directory and brings a running server down, so a failed launch that
+// left a postmaster behind does not get its data directory deleted from under
+// it.
+func TestStopOrphanedPostmaster(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
+	require.NoError(t, err)
+	defer embeddedPG.Stop() // fails because the server is already down, but still removes the directory
+
+	runtimePath := embeddedPG.RuntimePath()
+	_, port, _, _, _ := embeddedPG.GetConnectionDetails()
+	_, err = os.Stat(filepath.Join(runtimePath, "data", "postmaster.pid"))
+	require.NoError(t, err, "server should be running before the stop")
+
+	postgres.StopOrphanedPostmaster(runtimePath)
+
+	_, err = os.Stat(filepath.Join(runtimePath, "data", "postmaster.pid"))
+	require.True(t, os.IsNotExist(err), "postmaster.pid should be gone after the stop")
+	_, err = net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
+	require.Error(t, err, "nothing should be listening on the server's port")
+}

internal/postgres/export_test.go

@@ -0,0 +1,11 @@
+package postgres
+
+// Test-only hooks into unexported package state.
+var (
+	FindAvailablePort            = &findAvailablePort
+	MapToEmbeddedPostgresVersion = mapToEmbeddedPostgresVersion
+	StopOrphanedPostmaster       = stopOrphanedPostmaster
+)
+
+// RuntimePath exposes the instance's runtime directory to tests.
+func (ep *EmbeddedPostgres) RuntimePath() string { return ep.runtimePath }