Skip to content

Commit 2a475ac

Browse files
tianzhouclaude
andauthored
fix: retry embedded PostgreSQL launch when the probed port is taken (#666)
* fix: retry embedded PostgreSQL launch when the probed port is taken The PG14 job of the v1.13.3 release run failed in TestApplyCommand_LockTimeoutRetryNotAppliedToConcurrentIndex with "pg_ctl: could not start server" and no further output. The port for the embedded server is reserved by probing it and closing the probe socket before pg_ctl binds it, so another process (such as a parallel test package starting its own instance) can take it in between. With log_min_messages=PANIC the postmaster's bind failure was never printed. StartEmbeddedPostgres now retries the launch up to three times, each on a fresh port and runtime directory, removes the data directory of a failed attempt, and runs the server at log_min_messages=LOG so a failed launch reports its cause in the error. The log still goes to the library's temp file, not to the user's terminal. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * fix: stop a leftover postmaster before cleaning up a failed launch When "pg_ctl start -w" gives up waiting, embedded-postgres returns an error but the server keeps running, and its Stop refuses to act because the instance was never marked started. Run "pg_ctl stop -m immediate" against the attempt's data directory before removing it, so a retry never deletes a running server's files. Also cover the retry bound: a start that fails on every attempt returns an error after three tries instead of retrying forever. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
1 parent 8703c27 commit 2a475ac

3 files changed

Lines changed: 185 additions & 4 deletions

File tree

‎internal/postgres/embedded.go‎

Lines changed: 48 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"io"
1111
"net"
1212
"os"
13+
"os/exec"
1314
"path/filepath"
1415
"sort"
1516
"strings"
@@ -97,8 +98,33 @@ func DetectPostgresVersionAndExtensionsFromDB(host string, port int, database, u
9798
return version, extensions, nil
9899
}
99100

100-
// StartEmbeddedPostgres starts a temporary embedded PostgreSQL instance
101+
// embeddedStartAttempts bounds how many times StartEmbeddedPostgres tries to
102+
// launch the server. The port is reserved by probing it and closing the probe
103+
// socket before pg_ctl binds it, so another process (for example a parallel
104+
// test package starting its own instance) can take the port in between and
105+
// the launch fails with "could not start server". Each attempt picks a fresh
106+
// port and runtime directory.
107+
const embeddedStartAttempts = 3
108+
109+
// StartEmbeddedPostgres starts a temporary embedded PostgreSQL instance.
101110
func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, error) {
111+
var err error
112+
for attempt := 1; attempt <= embeddedStartAttempts; attempt++ {
113+
var embeddedPG *EmbeddedPostgres
114+
embeddedPG, err = startEmbeddedPostgresOnce(config)
115+
if err == nil {
116+
return embeddedPG, nil
117+
}
118+
if attempt < embeddedStartAttempts {
119+
logger.Get().Debug("embedded PostgreSQL failed to start, retrying", "attempt", attempt, "error", err)
120+
}
121+
}
122+
return nil, err
123+
}
124+
125+
// startEmbeddedPostgresOnce performs a single launch attempt on a fresh port
126+
// and runtime directory.
127+
func startEmbeddedPostgresOnce(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, error) {
102128
// Create unique runtime path and schema name
103129
tempSchema := GenerateTempSchemaName()
104130
runtimePath := filepath.Join(os.TempDir(), tempSchema)
@@ -123,7 +149,7 @@ func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, e
123149
StartParameters(map[string]string{
124150
"logging_collector": "off", // Disable log collector
125151
"log_destination": "stderr", // Send logs to stderr (which we discard)
126-
"log_min_messages": "PANIC", // Only log PANIC level messages
152+
"log_min_messages": "LOG", // Keep startup failures (e.g. port in use) in the pg_ctl output
127153
"log_statement": "none", // Don't log SQL statements
128154
"log_min_duration_statement": "-1", // Don't log slow queries
129155
"unix_socket_directories": runtimePath, // Use a directory that is guaranteed to exist
@@ -133,6 +159,8 @@ func StartEmbeddedPostgres(config *EmbeddedPostgresConfig) (*EmbeddedPostgres, e
133159
// Create and start PostgreSQL instance
134160
instance := embeddedpostgres.NewDatabase(pgConfig)
135161
if err := instance.Start(); err != nil {
162+
stopOrphanedPostmaster(runtimePath)
163+
os.RemoveAll(runtimePath)
136164
return nil, fmt.Errorf("failed to start embedded PostgreSQL: %w", err)
137165
}
138166

@@ -389,8 +417,24 @@ func installUntilFixpoint(names []string, install func(name string) error) map[s
389417
return lastErr
390418
}
391419

392-
// findAvailablePort finds an available TCP port for PostgreSQL to use
393-
func findAvailablePort() (int, error) {
420+
// stopOrphanedPostmaster stops a postmaster that a failed launch may have left
421+
// running under runtimePath. When "pg_ctl start -w" gives up waiting, the
422+
// library returns an error but the server keeps running, and its own Stop
423+
// refuses to act because it never marked the instance as started. Best
424+
// effort: with no server running, pg_ctl stop fails fast and that is fine.
425+
func stopOrphanedPostmaster(runtimePath string) {
426+
binDir := binariesPath
427+
if binDir == "" {
428+
binDir = runtimePath // embedded-postgres extracts the binaries here by default
429+
}
430+
pgCtl := filepath.Join(binDir, "bin", "pg_ctl")
431+
_ = exec.Command(pgCtl, "stop", "-w", "-m", "immediate", "-D", filepath.Join(runtimePath, "data")).Run()
432+
}
433+
434+
// findAvailablePort finds an available TCP port for PostgreSQL to use.
435+
// It is a variable so tests can hand StartEmbeddedPostgres a port that is
436+
// already taken and exercise the retry.
437+
var findAvailablePort = func() (int, error) {
394438
listener, err := net.Listen("tcp", ":0")
395439
if err != nil {
396440
return 0, err

‎internal/postgres/embedded_test.go‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
package postgres_test
2+
3+
import (
4+
"fmt"
5+
"net"
6+
"os"
7+
"path/filepath"
8+
"strconv"
9+
"testing"
10+
11+
"github.com/pgplex/pgschema/internal/postgres"
12+
"github.com/pgplex/pgschema/testutil"
13+
"github.com/stretchr/testify/require"
14+
)
15+
16+
func testEmbeddedConfig(t *testing.T) *postgres.EmbeddedPostgresConfig {
17+
t.Helper()
18+
major := testutil.LatestPostgresVersion
19+
if v := os.Getenv("PGSCHEMA_POSTGRES_VERSION"); v != "" {
20+
var err error
21+
major, err = strconv.Atoi(v)
22+
require.NoError(t, err)
23+
}
24+
version, err := postgres.MapToEmbeddedPostgresVersion(major)
25+
require.NoError(t, err)
26+
return &postgres.EmbeddedPostgresConfig{
27+
Version: version,
28+
Database: "testdb",
29+
Username: "testuser",
30+
Password: "testpass",
31+
}
32+
}
33+
34+
// holdPort returns a port that stays occupied for the rest of the test.
35+
func holdPort(t *testing.T) int {
36+
t.Helper()
37+
held, err := net.Listen("tcp", "localhost:0")
38+
require.NoError(t, err)
39+
t.Cleanup(func() { held.Close() })
40+
return held.Addr().(*net.TCPAddr).Port
41+
}
42+
43+
// stubFindAvailablePort replaces the port picker for the test and returns a
44+
// counter of how many times it was called.
45+
func stubFindAvailablePort(t *testing.T, pick func(call int, orig func() (int, error)) (int, error)) *int {
46+
t.Helper()
47+
orig := *postgres.FindAvailablePort
48+
t.Cleanup(func() { *postgres.FindAvailablePort = orig })
49+
calls := 0
50+
*postgres.FindAvailablePort = func() (int, error) {
51+
calls++
52+
return pick(calls, orig)
53+
}
54+
return &calls
55+
}
56+
57+
// TestStartEmbeddedPostgres_RetriesWhenPortIsTaken hands the first launch
58+
// attempt a port that is already in use, as happens when another process grabs
59+
// the probed port before pg_ctl binds it, and expects the next attempt to come
60+
// up on a fresh port instead of failing the whole start.
61+
func TestStartEmbeddedPostgres_RetriesWhenPortIsTaken(t *testing.T) {
62+
if testing.Short() {
63+
t.Skip("Skipping integration test in short mode")
64+
}
65+
66+
taken := holdPort(t)
67+
calls := stubFindAvailablePort(t, func(call int, orig func() (int, error)) (int, error) {
68+
if call == 1 {
69+
return taken, nil
70+
}
71+
return orig()
72+
})
73+
74+
embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
75+
require.NoError(t, err, "second attempt should start on a fresh port")
76+
defer embeddedPG.Stop()
77+
78+
_, port, _, _, _ := embeddedPG.GetConnectionDetails()
79+
require.Equal(t, 2, *calls, "expected exactly one retry")
80+
require.NotEqual(t, taken, port)
81+
}
82+
83+
// TestStartEmbeddedPostgres_GivesUpAfterRetries keeps every attempt on the
84+
// taken port and expects the start to fail after the bounded number of
85+
// attempts rather than retry forever.
86+
func TestStartEmbeddedPostgres_GivesUpAfterRetries(t *testing.T) {
87+
if testing.Short() {
88+
t.Skip("Skipping integration test in short mode")
89+
}
90+
91+
taken := holdPort(t)
92+
calls := stubFindAvailablePort(t, func(int, func() (int, error)) (int, error) {
93+
return taken, nil
94+
})
95+
96+
embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
97+
require.Error(t, err)
98+
require.Nil(t, embeddedPG)
99+
require.Equal(t, 3, *calls)
100+
}
101+
102+
// TestStopOrphanedPostmaster verifies the fallback stop finds pg_ctl under the
103+
// runtime directory and brings a running server down, so a failed launch that
104+
// left a postmaster behind does not get its data directory deleted from under
105+
// it.
106+
func TestStopOrphanedPostmaster(t *testing.T) {
107+
if testing.Short() {
108+
t.Skip("Skipping integration test in short mode")
109+
}
110+
111+
embeddedPG, err := postgres.StartEmbeddedPostgres(testEmbeddedConfig(t))
112+
require.NoError(t, err)
113+
defer embeddedPG.Stop() // fails because the server is already down, but still removes the directory
114+
115+
runtimePath := embeddedPG.RuntimePath()
116+
_, port, _, _, _ := embeddedPG.GetConnectionDetails()
117+
_, err = os.Stat(filepath.Join(runtimePath, "data", "postmaster.pid"))
118+
require.NoError(t, err, "server should be running before the stop")
119+
120+
postgres.StopOrphanedPostmaster(runtimePath)
121+
122+
_, err = os.Stat(filepath.Join(runtimePath, "data", "postmaster.pid"))
123+
require.True(t, os.IsNotExist(err), "postmaster.pid should be gone after the stop")
124+
_, err = net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
125+
require.Error(t, err, "nothing should be listening on the server's port")
126+
}

‎internal/postgres/export_test.go‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package postgres
2+
3+
// Test-only hooks into unexported package state.
4+
var (
5+
FindAvailablePort = &findAvailablePort
6+
MapToEmbeddedPostgresVersion = mapToEmbeddedPostgresVersion
7+
StopOrphanedPostmaster = stopOrphanedPostmaster
8+
)
9+
10+
// RuntimePath exposes the instance's runtime directory to tests.
11+
func (ep *EmbeddedPostgres) RuntimePath() string { return ep.runtimePath }

0 commit comments

Comments
 (0)