Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions internal/generate/cloud_engines_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,31 @@ func TestCloudEngineTemplatesLint(t *testing.T) {
}
}

// TestLocalStackHealthGatesOnAvailable guards the fix for the "shared-localstack
// unhealthy after 1 attempt" bug. LocalStack 3.x reports each configured SERVICE
// as "available" on startup — a service only flips to "running" after its first
// request. A healthcheck that greps solely for "running" therefore NEVER passes
// (nothing is running until traffic arrives), the container stays unhealthy, and
// the up saga aborts. The gate must accept "available".
func TestLocalStackHealthGatesOnAvailable(t *testing.T) {
src := template.NewFSSource(templates.FS)
res, err := template.Resolve(src, "localstack", nil)
if err != nil {
t.Fatal(err)
}
compose, err := LintResolved("localstack", res)
if err != nil {
t.Fatal(err)
}
s := string(compose)
if !strings.Contains(s, "available") {
t.Errorf("localstack healthcheck must accept the \"available\" state, not gate solely on \"running\":\n%s", s)
}
if strings.Contains(s, "grep -q running") {
t.Error("localstack healthcheck still greps solely for \"running\" — the deadlock bug")
}
}

// TestRabbitMQSecretIsValueless asserts RABBITMQ_DEFAULT_PASS is emitted as a
// valueless env key (no plaintext) — the §7.5 secret coupling for broker creds.
func TestRabbitMQSecretIsValueless(t *testing.T) {
Expand Down
107 changes: 107 additions & 0 deletions internal/orchestrate/connect_retry.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
package orchestrate

import (
"context"
"strings"
"time"

"github.com/open-source-cloud/devstack/internal/provision"
)

// This file hardens the host-side Postgres admin connection against the
// readiness race that surfaced as `db create` failing with "connect to shared
// postgres on 127.0.0.1:<port>: read: connection reset by peer".
//
// Why the race exists: the imperative resource path (and the up provision phase)
// publishes the shared engine's host port via an up-time compose overlay, then
// `docker compose up -d <inst>` applies it. Adding a published port RECREATES the
// container, so Postgres restarts; for a second or two afterwards Docker's
// userland proxy accepts the TCP connection on 127.0.0.1:<port> but the backend
// isn't listening yet, so it RSTs the handshake ("connection reset by peer",
// "failed to receive message", EOF). A single immediate connect loses the race.
//
// The fix mirrors how any client should treat a just-(re)started server: retry
// the connect with backoff for a bounded window. Idempotent and safe — a healthy
// server connects on the first try, so this only ever adds latency on the race.

const (
// connectRetryBudget bounds how long we retry a transient connect before
// giving up and surfacing the real error (Postgres genuinely down / wrong
// creds fail fast because those errors are not transient).
connectRetryBudget = 30 * time.Second
// connectRetryStart is the initial backoff; it doubles up to connectRetryMax.
connectRetryStart = 200 * time.Millisecond
connectRetryMax = 2 * time.Second
)

// transientConnErr reports whether a Postgres connect error is the engine still
// coming up after a port-overlay recreate (retry) rather than a permanent
// failure like bad credentials or an unknown database (fail fast).
func transientConnErr(err error) bool {
if err == nil {
return false
}
s := strings.ToLower(err.Error())
for _, m := range []string{
"connection reset by peer",
"connection refused",
"failed to receive message",
"the database system is starting up",
"broken pipe",
"unexpected eof",
"eof",
"i/o timeout",
"no route to host",
"server closed the connection unexpectedly",
} {
if strings.Contains(s, m) {
return true
}
}
return false
}

// sleepFn is indirected so tests can drive the backoff without real time.
var sleepFn = func(ctx context.Context, d time.Duration) error {
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-t.C:
return nil
}
}

// nowFn is indirected for tests.
var nowFn = time.Now

// retryingPgConnect wraps a PgConnector so a transient connect error (the engine
// just restarted to bind its host port) is retried with capped backoff for
// connectRetryBudget. A nil connector passes through nil (the default connector
// is substituted downstream). Non-transient errors and a cancelled context
// return immediately.
func retryingPgConnect(connect PgConnector) PgConnector {
if connect == nil {
return nil
}
return func(ctx context.Context, dsn string) (provision.Conn, func() error, error) {
deadline := nowFn().Add(connectRetryBudget)
backoff := connectRetryStart
for {
conn, closeFn, err := connect(ctx, dsn)
if err == nil {
return conn, closeFn, nil
}
if !transientConnErr(err) || !nowFn().Before(deadline) || ctx.Err() != nil {
return nil, nil, err
}
if serr := sleepFn(ctx, backoff); serr != nil {
return nil, nil, err
}
if backoff < connectRetryMax {
backoff *= 2
}
}
}
}
139 changes: 139 additions & 0 deletions internal/orchestrate/connect_retry_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
package orchestrate

import (
"context"
"errors"
"testing"
"time"

"github.com/open-source-cloud/devstack/internal/provision"
)

func TestTransientConnErr(t *testing.T) {
transient := []string{
"connect to shared postgres: read tcp 127.0.0.1:46820->127.0.0.1:45432: read: connection reset by peer",
"failed to receive message: EOF",
"dial tcp 127.0.0.1:45432: connect: connection refused",
"the database system is starting up",
}
for _, m := range transient {
if !transientConnErr(errors.New(m)) {
t.Errorf("expected transient: %q", m)
}
}
permanent := []string{
"password authentication failed for user \"devstack\"",
"database \"nope\" does not exist",
"",
}
for _, m := range permanent {
if transientConnErr(errors.New(m)) {
t.Errorf("expected permanent: %q", m)
}
}
if transientConnErr(nil) {
t.Error("nil is not transient")
}
}

// fakeConn is a throwaway provision.Conn for connector return values.
type fakeConn struct{}

func (fakeConn) Exec(context.Context, string, ...any) error { return nil }
func (fakeConn) Exists(context.Context, string, ...any) (bool, error) { return false, nil }

func withNoSleep(t *testing.T) {
t.Helper()
orig := sleepFn
sleepFn = func(ctx context.Context, _ time.Duration) error {
if ctx.Err() != nil {
return ctx.Err()
}
return nil
}
t.Cleanup(func() { sleepFn = orig })
}

func TestRetryingPgConnect_EventuallySucceeds(t *testing.T) {
withNoSleep(t)
calls := 0
base := PgConnector(func(context.Context, string) (provision.Conn, func() error, error) {
calls++
if calls < 3 {
return nil, nil, errors.New("read: connection reset by peer")
}
return fakeConn{}, func() error { return nil }, nil
})
conn, closeFn, err := retryingPgConnect(base)(context.Background(), "dsn")
if err != nil {
t.Fatalf("want success after retries, got %v", err)
}
if conn == nil || closeFn == nil {
t.Fatal("want a live conn + close on success")
}
if calls != 3 {
t.Errorf("calls = %d, want 3 (two transient failures then success)", calls)
}
}

func TestRetryingPgConnect_FailsFastOnPermanent(t *testing.T) {
withNoSleep(t)
calls := 0
base := PgConnector(func(context.Context, string) (provision.Conn, func() error, error) {
calls++
return nil, nil, errors.New("password authentication failed")
})
_, _, err := retryingPgConnect(base)(context.Background(), "dsn")
if err == nil {
t.Fatal("want the permanent error surfaced")
}
if calls != 1 {
t.Errorf("calls = %d, want 1 (no retry on a permanent error)", calls)
}
}

func TestRetryingPgConnect_BudgetBounded(t *testing.T) {
withNoSleep(t)
// Drive a synthetic clock so the 30s budget elapses without real waiting.
origNow := nowFn
tick := time.Unix(0, 0)
nowFn = func() time.Time { tick = tick.Add(5 * time.Second); return tick }
t.Cleanup(func() { nowFn = origNow })

calls := 0
base := PgConnector(func(context.Context, string) (provision.Conn, func() error, error) {
calls++
return nil, nil, errors.New("connection refused")
})
_, _, err := retryingPgConnect(base)(context.Background(), "dsn")
if err == nil {
t.Fatal("want the transient error surfaced after the budget elapses")
}
if calls < 2 {
t.Errorf("calls = %d, want ≥2 (retried before giving up)", calls)
}
}

func TestRetryingPgConnect_HonorsContextCancel(t *testing.T) {
withNoSleep(t)
ctx, cancel := context.WithCancel(context.Background())
cancel()
calls := 0
base := PgConnector(func(context.Context, string) (provision.Conn, func() error, error) {
calls++
return nil, nil, errors.New("connection reset by peer")
})
_, _, err := retryingPgConnect(base)(ctx, "dsn")
if err == nil {
t.Fatal("want an error when ctx is already cancelled")
}
if calls != 1 {
t.Errorf("calls = %d, want 1 (a cancelled ctx stops the retry loop)", calls)
}
}

func TestRetryingPgConnect_NilPassthrough(t *testing.T) {
if retryingPgConnect(nil) != nil {
t.Error("retryingPgConnect(nil) must be nil so the default connector is used downstream")
}
}
4 changes: 4 additions & 0 deletions internal/orchestrate/provision.go
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,10 @@ func provisionPhase(d UpDeps, targets []provTarget) Phase {
if connect == nil {
connect = defaultPgConnect
}
// Retry a transient connect: the host-port overlay may have just
// recreated the engine, so the proxy RSTs the handshake until Postgres
// relistens (the "connection reset by peer" race).
connect = retryingPgConnect(connect)
byInst := map[string][]string{}
for _, t := range targets {
byInst[t.instance] = append(byInst[t.instance], t.project)
Expand Down
2 changes: 1 addition & 1 deletion internal/orchestrate/resources.go
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ func toResourceConnector(connect PgConnector) resource.PgConnector {
// daemon/endpoint-free in tests. Postgres + MinIO are live in this milestone.
func buildRegistry(d UpDeps) *resource.Registry {
return resource.NewRegistry(
resource.Postgres{Connect: toResourceConnector(d.PgConnect)},
resource.Postgres{Connect: toResourceConnector(retryingPgConnect(d.PgConnect))},
resource.MinIO{Factory: d.S3Factory},
resource.NATS{Factory: d.NatsFactory},
resource.Kafka{Factory: d.KafkaFactory},
Expand Down
2 changes: 1 addition & 1 deletion templates/localstack/golden.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ services:
healthcheck:
test:
- CMD-SHELL
- curl -sf http://localhost:4566/_localstack/health | grep -q running
- curl -sf http://localhost:4566/_localstack/health | grep -Eq 'available|running'
timeout: 5s
interval: 10s
retries: 8
Expand Down
9 changes: 7 additions & 2 deletions templates/localstack/template.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,13 @@ service:
volumes:
- "localstackdata:/var/lib/localstack"
healthcheck:
# /_localstack/health reports per-service readiness; gate on the edge being up.
test: ["CMD-SHELL", "curl -sf http://localhost:4566/_localstack/health | grep -q running"]
# /_localstack/health reports per-service readiness. On a fresh start the
# configured SERVICES are "available" (a service only flips to "running" after
# its first request), so gate on ANY service being available|running — the
# honest "edge up + providers loaded" signal. Gating on "running" alone
# deadlocks: nothing is "running" until traffic arrives, so the container
# never goes healthy and the up saga aborts (verified on localstack 3.8.1).
test: ["CMD-SHELL", "curl -sf http://localhost:4566/_localstack/health | grep -Eq 'available|running'"]
interval: 10s
timeout: 5s
retries: 8
Expand Down
Loading