From 798474517e09a5903ec71368a0faeadf884b4124 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 15:03:22 -0300 Subject: [PATCH] fix(cloud): shared-localstack health gate + db-create connect race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two runtime bugs surfaced while exercising the cloud/resource stack on WSL2. 1. shared-localstack "unhealthy after 1 attempt". LocalStack 3.x reports each configured SERVICE as "available" on startup — a service only flips to "running" after its first request. The healthcheck greps solely for "running", so nothing ever matches, the container is permanently unhealthy, and the up saga aborts before any traffic can flip a service to "running" (a deadlock). Gate on `available|running` — the honest "edge up + providers loaded" signal. Verified live on localstack 3.8.1: the new check exits 0, the old one exits 1. Golden regenerated through the real render path; regression test added (internal/generate). 2. `db create` → "connect to shared postgres on 127.0.0.1:45432: read: connection reset by peer". The imperative resource path (and the up provision phase) publishes the engine's host port via an up-time compose overlay, and `compose up -d` recreates the container to bind it. For a second or two the Docker userland proxy accepts the TCP connection but Postgres isn't listening yet, so it RSTs the handshake. A single immediate connect loses the race. Add retryingPgConnect: a capped-backoff retry (30s budget) around the admin connect that retries only transient "just-restarted" errors (reset/refused/ EOF/starting-up) and fails fast on real errors (bad creds/unknown db). Wired into both the imperative registry and the saga provision phase. Unit + race tested with an injected clock/connector (no live server needed). Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/generate/cloud_engines_test.go | 25 ++++ internal/orchestrate/connect_retry.go | 107 ++++++++++++++++ internal/orchestrate/connect_retry_test.go | 139 +++++++++++++++++++++ internal/orchestrate/provision.go | 4 + internal/orchestrate/resources.go | 2 +- templates/localstack/golden.yaml | 2 +- templates/localstack/template.yaml | 9 +- 7 files changed, 284 insertions(+), 4 deletions(-) create mode 100644 internal/orchestrate/connect_retry.go create mode 100644 internal/orchestrate/connect_retry_test.go diff --git a/internal/generate/cloud_engines_test.go b/internal/generate/cloud_engines_test.go index 84ec003..2d908d7 100644 --- a/internal/generate/cloud_engines_test.go +++ b/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) { diff --git a/internal/orchestrate/connect_retry.go b/internal/orchestrate/connect_retry.go new file mode 100644 index 0000000..2c9fccb --- /dev/null +++ b/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:: 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 ` 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: 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 + } + } + } +} diff --git a/internal/orchestrate/connect_retry_test.go b/internal/orchestrate/connect_retry_test.go new file mode 100644 index 0000000..0911be3 --- /dev/null +++ b/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") + } +} diff --git a/internal/orchestrate/provision.go b/internal/orchestrate/provision.go index d5da744..64c7620 100644 --- a/internal/orchestrate/provision.go +++ b/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) diff --git a/internal/orchestrate/resources.go b/internal/orchestrate/resources.go index 35bb8af..0ab71d8 100644 --- a/internal/orchestrate/resources.go +++ b/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}, diff --git a/templates/localstack/golden.yaml b/templates/localstack/golden.yaml index 65b2ef5..3e5cf23 100644 --- a/templates/localstack/golden.yaml +++ b/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 diff --git a/templates/localstack/template.yaml b/templates/localstack/template.yaml index 7c48a84..de654c9 100644 --- a/templates/localstack/template.yaml +++ b/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