Merge pull request #109 from open-source-cloud/fix/cloud-stack-health-readiness

fix(cloud): shared-localstack health gate + db-create connect race

Author: gustavobertoi

internal/generate/cloud_engines_test.go

@@ -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) {

internal/orchestrate/connect_retry.go

@@ -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
+			}
+		}
+	}
+}

internal/orchestrate/connect_retry_test.go

@@ -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")
+	}
+}

internal/orchestrate/provision.go

@@ -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)

internal/orchestrate/resources.go

@@ -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},

templates/localstack/golden.yaml

@@ -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

templates/localstack/template.yaml

@@ -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