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