From 798474517e09a5903ec71368a0faeadf884b4124 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 15:03:22 -0300 Subject: [PATCH 1/3] 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 From ef295d1b6b13fc883a0c8c3a0778b3f117b7a883 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 15:16:27 -0300 Subject: [PATCH 2/3] feat(shared): expose shared services on stable localhost ports for GUI clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `devstack shared expose [services...]` publishes the shared engines on stable 127.0.0.1 host ports so a developer's GUI clients — DataGrip/TablePlus, a Redis or S3 browser, the RabbitMQ management UI — can connect, without a duplicate stack per repo. `devstack shared ports` is the read-only projection (ports + connection strings); `shared expose --off` returns the stack to DNS-only. Design (spec 03 host-reachability): - Opt-in and loopback-only; the default is still "no host ports" (DNS over devstack_shared). Exposure is an UP-TIME compose overlay (compose.expose.yaml), so the deterministic, golden-asserted generated compose is untouched — same posture as the provisioning overlay. - Ports are ledger-allocated (FreeHostPort), stable across runs, and sit in a distinct 5xxxx range so the expose overlay and the 4xxxx provisioning overlay never publish the same host port (a duplicate binding). Kafka is the one deliberate exception: host clients must reach the fixed advertised 127.0.0.1:49092. - Persistent: the up saga re-applies the expose overlay so host ports survive up/down. Refused on a remote backend (a remote bridge is not host-routable). - Per engine it publishes the primary protocol port plus the useful secondary UI ports (MinIO console, RabbitMQ management, NATS monitor) and prints a client-ready connection string for each (the postgres admin DSN + a reminder that per-project DBs use the documented per-project dev creds). Tests: overlay write/read round-trip, instance resolution (all/named/rejects non-engines), per-engine connection URLs, the expose-vs-provision no-collision invariant, CLI table/JSON/quiet rendering, and command registration. Verified live on WSL2: `shared ports` projection, the overlay merges cleanly with the running shared compose (`docker compose config`), and a host psql client reaches shared-postgres on the published loopback port. Determinism/golden unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/cli/expose.go | 116 +++++++++++ internal/cli/expose_test.go | 76 +++++++ internal/cli/shared.go | 2 + internal/orchestrate/expose.go | 308 ++++++++++++++++++++++++++++ internal/orchestrate/expose_test.go | 157 ++++++++++++++ internal/orchestrate/up.go | 12 +- 6 files changed, 670 insertions(+), 1 deletion(-) create mode 100644 internal/cli/expose.go create mode 100644 internal/cli/expose_test.go create mode 100644 internal/orchestrate/expose.go create mode 100644 internal/orchestrate/expose_test.go diff --git a/internal/cli/expose.go b/internal/cli/expose.go new file mode 100644 index 0000000..6a10d15 --- /dev/null +++ b/internal/cli/expose.go @@ -0,0 +1,116 @@ +package cli + +import ( + "fmt" + "text/tabwriter" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/orchestrate" +) + +// newSharedExposeCmd wires `shared expose [services...]` — publish the shared +// engines on stable 127.0.0.1 host ports so GUI clients (DataGrip, a Redis/S3 +// browser, the RabbitMQ UI) can connect. Opt-in and loopback-only; it never +// touches the deterministic generated compose (an up-time overlay). `--off` +// removes the publish and returns the stack to DNS-only. +func newSharedExposeCmd(g *GlobalOpts) *cobra.Command { + var off bool + cmd := &cobra.Command{ + Use: "expose [services...]", + Short: "Publish shared services on stable 127.0.0.1 ports for local GUI clients", + Long: "Publish the shared engines on stable 127.0.0.1 host ports so host tools and GUI\n" + + "clients (DataGrip, TablePlus, a Redis/S3 browser, the RabbitMQ management UI)\n" + + "can reach them. Ports are ledger-allocated (stable across runs) and loopback-only.\n" + + "With no arguments, every exposable shared service is published; name services to\n" + + "scope it. `--off` removes the publish. The persist survives up/down.", + RunE: func(cmd *cobra.Command, args []string) error { + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + if off { + if err := orchestrate.UnexposeShared(cmd.Context(), d); err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"exposed": []any{}}) + } + if !g.Quiet { + fmt.Fprintln(cmd.OutOrStdout(), "shared services are DNS-only again (host ports removed)") + } + return nil + } + ports, err := orchestrate.ExposeShared(cmd.Context(), d, args) + if err != nil { + return err + } + return renderExposed(cmd, g, ports) + }, + } + cmd.Flags().BoolVar(&off, "off", false, "remove the host-port publish (back to DNS-only)") + return cmd +} + +// newSharedPortsCmd wires `shared ports` — the read-only projection of the +// currently-published host ports + connection strings (lock-free snapshot). +func newSharedPortsCmd(g *GlobalOpts) *cobra.Command { + return &cobra.Command{ + Use: "ports", + Short: "Show the published 127.0.0.1 host ports for shared services (and connection strings)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + ports, err := orchestrate.ExposedStatus(cmd.Context(), d) + if err != nil { + return err + } + if len(ports) == 0 && !g.JSON { + fmt.Fprintln(cmd.OutOrStdout(), "no shared services exposed — run `devstack shared expose`") + return nil + } + return renderExposed(cmd, g, ports) + }, + } +} + +// renderExposed prints the exposed-port projection as JSON or an aligned table. +func renderExposed(cmd *cobra.Command, g *GlobalOpts, ports []orchestrate.ExposedPort) error { + if g.JSON { + return writeJSON(cmd, map[string]any{"exposed": ports}) + } + if g.Quiet { + for _, p := range ports { + if p.URL != "" { + fmt.Fprintln(cmd.OutOrStdout(), p.URL) + } + } + return nil + } + tw := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "SERVICE\tPORT\tADDRESS\tCONNECT") + for _, p := range ports { + label := p.Alias + if !p.Primary { + label = p.Alias + " (" + p.Label + ")" + } + fmt.Fprintf(tw, "%s\t%s\t127.0.0.1:%d\t%s\n", label, p.Label, p.Port, p.URL) + } + if err := tw.Flush(); err != nil { + return err + } + // A one-line reminder that per-project Postgres DBs use their own dev creds. + for _, p := range ports { + if p.Engine == "postgres" && p.Primary { + fmt.Fprintf(cmd.OutOrStdout(), + "\nper-project database: postgres://:@127.0.0.1:%d/?sslmode=disable\n", p.Port) + break + } + } + return nil +} diff --git a/internal/cli/expose_test.go b/internal/cli/expose_test.go new file mode 100644 index 0000000..dc312a9 --- /dev/null +++ b/internal/cli/expose_test.go @@ -0,0 +1,76 @@ +package cli + +import ( + "bytes" + "strings" + "testing" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/orchestrate" +) + +func exposeFixture() []orchestrate.ExposedPort { + return []orchestrate.ExposedPort{ + {Instance: "postgres", Engine: "postgres", Alias: "shared-postgres", Label: "postgres", Host: "127.0.0.1", Port: 55432, Container: 5432, Primary: true, URL: "postgres://devstack:devstack@127.0.0.1:55432/postgres?sslmode=disable"}, + {Instance: "minio", Engine: "minio", Alias: "shared-minio", Label: "console", Host: "127.0.0.1", Port: 59001, Container: 9001, Primary: false, URL: "http://127.0.0.1:59001"}, + } +} + +func TestRenderExposed_Table(t *testing.T) { + var buf bytes.Buffer + cmd := &cobra.Command{} + cmd.SetOut(&buf) + if err := renderExposed(cmd, &GlobalOpts{}, exposeFixture()); err != nil { + t.Fatal(err) + } + out := buf.String() + for _, want := range []string{"shared-postgres", "55432", "shared-minio (console)", "59001", "per-project database:"} { + if !strings.Contains(out, want) { + t.Errorf("table missing %q:\n%s", want, out) + } + } +} + +func TestRenderExposed_JSON(t *testing.T) { + var buf bytes.Buffer + cmd := &cobra.Command{} + cmd.SetOut(&buf) + if err := renderExposed(cmd, &GlobalOpts{JSON: true}, exposeFixture()); err != nil { + t.Fatal(err) + } + out := buf.String() + if !strings.Contains(out, "\"exposed\"") || !strings.Contains(out, "\"port\": 55432") { + t.Errorf("json missing fields:\n%s", out) + } +} + +func TestRenderExposed_Quiet(t *testing.T) { + var buf bytes.Buffer + cmd := &cobra.Command{} + cmd.SetOut(&buf) + if err := renderExposed(cmd, &GlobalOpts{Quiet: true}, exposeFixture()); err != nil { + t.Fatal(err) + } + out := strings.TrimSpace(buf.String()) + // Quiet emits only the connection URLs, one per line. + lines := strings.Split(out, "\n") + if len(lines) != 2 || !strings.HasPrefix(lines[0], "postgres://") { + t.Errorf("quiet should print only URLs, got:\n%s", out) + } +} + +// TestSharedExposeCommandsRegistered guards that `shared expose` and +// `shared ports` are wired into the shared command tree. +func TestSharedExposeCommandsRegistered(t *testing.T) { + sh := newSharedCmd(&GlobalOpts{}) + have := map[string]bool{} + for _, c := range sh.Commands() { + have[c.Name()] = true + } + for _, want := range []string{"expose", "ports", "status", "gc", "doctor"} { + if !have[want] { + t.Errorf("shared subcommand %q not registered", want) + } + } +} diff --git a/internal/cli/shared.go b/internal/cli/shared.go index 41a4928..2dbef72 100644 --- a/internal/cli/shared.go +++ b/internal/cli/shared.go @@ -27,6 +27,8 @@ func newSharedCmd(g *GlobalOpts) *cobra.Command { newSharedStatusCmd(g), newSharedGcCmd(g), newSharedDoctorCmd(g), + newSharedExposeCmd(g), + newSharedPortsCmd(g), ) return cmd } diff --git a/internal/orchestrate/expose.go b/internal/orchestrate/expose.go new file mode 100644 index 0000000..a4508fe --- /dev/null +++ b/internal/orchestrate/expose.go @@ -0,0 +1,308 @@ +package orchestrate + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" +) + +// This file implements `shared expose` / `shared ports`: publishing the shared +// engines on stable 127.0.0.1 host ports so a developer's GUI clients (DataGrip, +// a Redis or S3 browser, the RabbitMQ management UI) can reach them. The default +// posture is still "no host ports" (DNS over devstack_shared, spec 03); exposure +// is an explicit opt-in that, like provisioning, is an UP-TIME compose overlay — +// it never touches the deterministic, golden-asserted generated compose. +// +// Exposure uses its OWN host-port range (55xxx/58xxx…), distinct from the +// provisioning range (45xxx), so the expose overlay and the provision overlay +// never publish the same host port and can both be applied without a duplicate +// binding. Ports are ledger-allocated (FreeHostPort), so the same engine keeps +// the same host port across runs and two terminals never collide. + +const exposeFile = "compose.expose.yaml" + +// exposePort is one host-published port for a shared engine. +type exposePort struct { + container int // the in-container port to publish + label string // human label (postgres / console / management / …) + purpose string // ledger port_alloc purpose (stable per engine) + base int // host-port search base + primary bool // the port a client uses for the engine's main protocol +} + +// exposeEngines maps a shared engine (template name) to the ports `shared expose` +// publishes on 127.0.0.1. Bases sit in the 5xxxx range so they never collide with +// the 4xxxx provisioning overlay. Kafka is the exception: host clients MUST reach +// the broker on 127.0.0.1:49092 (the fixed advertised external listener from the +// template), so it reuses the kafka provision port rather than a 5xxxx one. +var exposeEngines = map[string][]exposePort{ + "postgres": {{5432, "postgres", "pg-expose", 55432, true}}, + "redis": {{6379, "redis", "redis-expose", 56379, true}}, + "minio": {{9000, "s3", "minio-expose", 59000, true}, {9001, "console", "minio-console-expose", 59001, false}}, + "localstack": {{4566, "aws", "localstack-expose", 54566, true}}, + "ministack": {{4566, "aws", "ministack-expose", 54567, true}}, + "nats": {{4222, "nats", "nats-expose", 54222, true}, {8222, "monitor", "nats-monitor-expose", 58222, false}}, + "kafka": {{19092, "kafka", "kafka-provision", 49092, true}}, + "rabbitmq": {{5672, "amqp", "rmq-expose", 55672, true}, {15672, "management", "rmq-mgmt-expose", 55673, false}}, +} + +// ExposableEngine reports whether an engine has a defined host-expose port set. +func ExposableEngine(engine string) bool { + _, ok := exposeEngines[engine] + return ok +} + +// ExposedPort is one host-published shared-service port with a client-ready +// connection hint (the `--json` schema + the plain-table source). +type ExposedPort struct { + Instance string `json:"instance"` + Engine string `json:"engine"` + Alias string `json:"alias"` + Label string `json:"label"` + Host string `json:"host"` + Port int `json:"port"` + Container int `json:"container"` + Primary bool `json:"primary"` + URL string `json:"url,omitempty"` +} + +// publishedPort is one host:container mapping for the overlay writer. +type publishedPort struct { + host int + container int +} + +// resolveExposeInstances returns the shared instances to expose: the requested +// subset (validated), or — when none are named — every shared instance whose +// engine supports exposure. Order is stable (sorted) for deterministic output. +func resolveExposeInstances(d UpDeps, requested []string) ([]string, error) { + shared := d.Model.Workspace.Shared + if len(requested) == 0 { + var all []string + for name, s := range shared { + if ExposableEngine(s.Template) { + all = append(all, name) + } + } + sort.Strings(all) + if len(all) == 0 { + return nil, fmt.Errorf("no exposable shared services in this workspace (declare one under workspace.shared and run `devstack up`)") + } + return all, nil + } + var out []string + for _, name := range requested { + s, ok := shared[name] + if !ok { + return nil, fmt.Errorf("no shared service %q in this workspace", name) + } + if !ExposableEngine(s.Template) { + return nil, fmt.Errorf("shared service %q (engine %q) has no host-expose ports defined", name, s.Template) + } + out = append(out, name) + } + sort.Strings(out) + return out, nil +} + +// allocateExposePorts resolves the stable host ports for each instance's expose +// port set (idempotent via the ledger) and builds the ExposedPort projection. +func allocateExposePorts(ctx context.Context, d UpDeps, insts []string) ([]ExposedPort, map[string][]publishedPort, error) { + pub := map[string][]publishedPort{} + var out []ExposedPort + for _, inst := range insts { + engine := d.Model.Workspace.Shared[inst].Template + params := d.Model.Workspace.Shared[inst].Params + for _, ep := range exposeEngines[engine] { + port, err := d.Manager.FreeHostPort(ctx, generate.SharedAlias(inst), ep.purpose, ep.base) + if err != nil { + return nil, nil, fmt.Errorf("allocate %s host port for %s: %w", ep.label, inst, err) + } + pub[inst] = append(pub[inst], publishedPort{host: port, container: ep.container}) + out = append(out, ExposedPort{ + Instance: inst, Engine: engine, Alias: generate.SharedAlias(inst), + Label: ep.label, Host: "127.0.0.1", Port: port, Container: ep.container, + Primary: ep.primary, URL: connectionURL(engine, ep, params, port), + }) + } + } + return out, pub, nil +} + +// writeExposeOverlay writes the persistent up-time overlay that publishes each +// instance's expose ports on 127.0.0.1. Loopback-only (never 0.0.0.0) so nothing +// leaves the host. Instances/ports are sorted so the file is byte-stable. +func writeExposeOverlay(root string, pub map[string][]publishedPort) (string, error) { + insts := make([]string, 0, len(pub)) + for inst := range pub { + insts = append(insts, inst) + } + sort.Strings(insts) + var b strings.Builder + b.WriteString("services:\n") + for _, inst := range insts { + ports := pub[inst] + sort.Slice(ports, func(i, j int) bool { return ports[i].container < ports[j].container }) + fmt.Fprintf(&b, " %s:\n ports:\n", inst) + for _, p := range ports { + fmt.Fprintf(&b, " - \"127.0.0.1:%d:%d\"\n", p.host, p.container) + } + } + dir := filepath.Join(root, generate.GenDir, "shared") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + path := filepath.Join(dir, exposeFile) + if err := os.WriteFile(path, []byte(b.String()), 0o644); err != nil { + return "", err + } + return path, nil +} + +// exposeOverlayPath returns the overlay path (whether or not it exists yet). +func exposeOverlayPath(root string) string { + return filepath.Join(root, generate.GenDir, "shared", exposeFile) +} + +// fileExists reports whether path is an existing regular file. +func fileExists(path string) bool { + info, err := os.Stat(path) + return err == nil && !info.IsDir() +} + +// ExposeShared publishes the requested shared instances (or all exposable ones) +// on stable 127.0.0.1 host ports and applies the overlay by recreating those +// services. It returns the connection projection. Refuses a remote (ViaProxy) +// backend, whose bridge network is not host-routable (spec 21). +func ExposeShared(ctx context.Context, d UpDeps, requested []string) ([]ExposedPort, error) { + if d.Backend.Reachability() == docker.ViaProxy { + return nil, fmt.Errorf("cannot publish host ports on a %s: a remote bridge network is not host-routable (spec 21); reach it through a tunnel instead", d.Backend.String()) + } + insts, err := resolveExposeInstances(d, requested) + if err != nil { + return nil, err + } + outDir := filepath.Join(d.Model.Root, generate.GenDir, "shared") + if _, err := os.Stat(filepath.Join(outDir, generate.ComposeFile)); err != nil { + return nil, fmt.Errorf("shared stack not generated yet — run `devstack up` first") + } + out, pub, err := allocateExposePorts(ctx, d, insts) + if err != nil { + return nil, err + } + overlay, err := writeExposeOverlay(d.Model.Root, pub) + if err != nil { + return nil, err + } + if err := composeUpShared(ctx, d, outDir, []string{overlay}, insts); err != nil { + return nil, fmt.Errorf("apply expose overlay: %w", err) + } + return out, nil +} + +// UnexposeShared removes the expose overlay and recreates the shared services +// without their host ports (DNS-only again). Ledger port rows are left in place +// (idempotent — a later `expose` reuses the same ports). +func UnexposeShared(ctx context.Context, d UpDeps) error { + path := exposeOverlayPath(d.Model.Root) + if _, err := os.Stat(path); err != nil { + return nil // nothing exposed + } + insts := exposedInstances(d.Model.Root) + if err := os.Remove(path); err != nil { + return err + } + outDir := filepath.Join(d.Model.Root, generate.GenDir, "shared") + if _, err := os.Stat(filepath.Join(outDir, generate.ComposeFile)); err != nil { + return nil // stack not up; overlay removal is enough + } + return composeUpShared(ctx, d, outDir, nil, insts) +} + +// ExposedStatus is the read-only projection for `shared ports`: it re-derives the +// currently-exposed ports from the persisted overlay + the ledger, without +// mutating anything (lock-free snapshot). +func ExposedStatus(ctx context.Context, d UpDeps) ([]ExposedPort, error) { + insts := exposedInstances(d.Model.Root) + if len(insts) == 0 { + return nil, nil + } + out, _, err := allocateExposePorts(ctx, d, insts) + return out, err +} + +// exposedInstances reads which instances currently have an expose overlay by +// parsing the overlay's top-level service keys. Returns nil if not exposed. +func exposedInstances(root string) []string { + data, err := os.ReadFile(exposeOverlayPath(root)) + if err != nil { + return nil + } + var insts []string + for line := range strings.SplitSeq(string(data), "\n") { + // Top-level service keys are indented exactly two spaces: " :". + if strings.HasPrefix(line, " ") && !strings.HasPrefix(line, " ") && strings.HasSuffix(strings.TrimSpace(line), ":") { + insts = append(insts, strings.TrimSuffix(strings.TrimSpace(line), ":")) + } + } + sort.Strings(insts) + return insts +} + +// composeUpShared runs `compose up -d ` for the shared stack with the given +// override files (nil = none), pinned to the active backend. +func composeUpShared(ctx context.Context, d UpDeps, outDir string, overrides, insts []string) error { + runner := d.Runner + if runner == nil { + runner = docker.ExecRunner{} + } + cp := docker.Compose{ + Project: generate.SharedStackName, + File: filepath.Join(outDir, generate.ComposeFile), + Dir: outDir, + Runner: runner, + Overrides: overrides, + ContextEnv: d.Backend.ComposeEnv(), + } + return cp.Up(ctx, insts...) +} + +// connectionURL builds a client-ready connection hint for one exposed port. +// Credentials shown are the shared engine's dev admin creds (loopback-only, +// container-isolation-is-a-non-goal threat model); per-project DB creds follow +// the documented postgres://:@… DSN. +func connectionURL(engine string, ep exposePort, params map[string]any, port int) string { + host := fmt.Sprintf("127.0.0.1:%d", port) + switch engine { + case "postgres": + user := paramString(params, "rootUser", "devstack") + pass := paramString(params, "rootPassword", "devstack") + return fmt.Sprintf("postgres://%s:%s@%s/postgres?sslmode=disable", user, pass, host) + case "redis": + return "redis://" + host + case "minio": + return "http://" + host // S3 endpoint / console URL + case "localstack", "ministack": + return "http://" + host // AWS endpoint-url + case "nats": + if ep.label == "monitor" { + return "http://" + host + } + return "nats://" + host + case "kafka": + return host // bootstrap server + case "rabbitmq": + if ep.label == "management" { + return "http://" + host + } + user := paramString(params, "user", "devstack") + return fmt.Sprintf("amqp://%s@%s", user, host) + } + return host +} diff --git a/internal/orchestrate/expose_test.go b/internal/orchestrate/expose_test.go new file mode 100644 index 0000000..8364179 --- /dev/null +++ b/internal/orchestrate/expose_test.go @@ -0,0 +1,157 @@ +package orchestrate + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/open-source-cloud/devstack/internal/config" +) + +func exposeModel() *config.Model { + return &config.Model{ + Workspace: config.Workspace{ + Name: "w", + Shared: map[string]config.SharedSvc{ + "postgres": {Template: "postgres"}, + "minio": {Template: "minio"}, + "localstack": {Template: "localstack"}, + "web": {Template: "node.vite"}, // not an engine → not exposable + }, + }, + } +} + +func TestResolveExposeInstances_AllExposable(t *testing.T) { + d := UpDeps{Model: exposeModel()} + got, err := resolveExposeInstances(d, nil) + if err != nil { + t.Fatal(err) + } + want := []string{"localstack", "minio", "postgres"} // sorted; "web" excluded + if strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("all = %v, want %v", got, want) + } +} + +func TestResolveExposeInstances_NamedAndErrors(t *testing.T) { + d := UpDeps{Model: exposeModel()} + got, err := resolveExposeInstances(d, []string{"minio", "postgres"}) + if err != nil || strings.Join(got, ",") != "minio,postgres" { + t.Fatalf("named = %v err=%v", got, err) + } + if _, err := resolveExposeInstances(d, []string{"nope"}); err == nil { + t.Error("unknown instance should error") + } + if _, err := resolveExposeInstances(d, []string{"web"}); err == nil { + t.Error("non-engine shared service should be rejected as non-exposable") + } +} + +func TestWriteAndReadExposeOverlay(t *testing.T) { + root := t.TempDir() + pub := map[string][]publishedPort{ + "minio": {{host: 59001, container: 9001}, {host: 59000, container: 9000}}, + "postgres": {{host: 55432, container: 5432}}, + } + path, err := writeExposeOverlay(root, pub) + if err != nil { + t.Fatal(err) + } + data, _ := os.ReadFile(path) + got := string(data) + // Instances sorted; minio's ports sorted by container (9000 before 9001). + want := "services:\n" + + " minio:\n ports:\n" + + " - \"127.0.0.1:59000:9000\"\n" + + " - \"127.0.0.1:59001:9001\"\n" + + " postgres:\n ports:\n" + + " - \"127.0.0.1:55432:5432\"\n" + if got != want { + t.Errorf("overlay mismatch:\n--- got ---\n%s\n--- want ---\n%s", got, want) + } + // exposedInstances must round-trip the service keys (and ignore the port lines). + insts := exposedInstances(root) + if strings.Join(insts, ",") != "minio,postgres" { + t.Errorf("exposedInstances = %v, want [minio postgres]", insts) + } +} + +func TestExposedInstances_NoneWhenAbsent(t *testing.T) { + if got := exposedInstances(t.TempDir()); got != nil { + t.Errorf("no overlay → nil, got %v", got) + } +} + +func TestConnectionURL(t *testing.T) { + pgParams := map[string]any{"rootUser": "admin", "rootPassword": "s3cret"} + cases := []struct { + engine string + ep exposePort + params map[string]any + port int + want string + }{ + {"postgres", exposePort{5432, "postgres", "", 0, true}, pgParams, 55432, "postgres://admin:s3cret@127.0.0.1:55432/postgres?sslmode=disable"}, + {"postgres", exposePort{5432, "postgres", "", 0, true}, nil, 55432, "postgres://devstack:devstack@127.0.0.1:55432/postgres?sslmode=disable"}, + {"redis", exposePort{6379, "redis", "", 0, true}, nil, 56379, "redis://127.0.0.1:56379"}, + {"minio", exposePort{9000, "s3", "", 0, true}, nil, 59000, "http://127.0.0.1:59000"}, + {"localstack", exposePort{4566, "aws", "", 0, true}, nil, 54566, "http://127.0.0.1:54566"}, + {"nats", exposePort{8222, "monitor", "", 0, false}, nil, 58222, "http://127.0.0.1:58222"}, + {"nats", exposePort{4222, "nats", "", 0, true}, nil, 54222, "nats://127.0.0.1:54222"}, + {"kafka", exposePort{19092, "kafka", "", 0, true}, nil, 49092, "127.0.0.1:49092"}, + {"rabbitmq", exposePort{15672, "management", "", 0, false}, nil, 55673, "http://127.0.0.1:55673"}, + {"rabbitmq", exposePort{5672, "amqp", "", 0, true}, nil, 55672, "amqp://devstack@127.0.0.1:55672"}, + } + for _, tc := range cases { + if got := connectionURL(tc.engine, tc.ep, tc.params, tc.port); got != tc.want { + t.Errorf("%s/%s = %q, want %q", tc.engine, tc.ep.label, got, tc.want) + } + } +} + +// TestExposePortsNeverCollideWithProvision is the load-bearing invariant: the +// expose overlay and the provision overlay must never publish the SAME host port +// (base) for the SAME engine, or applying both recreates the container with a +// duplicate binding. Kafka is the deliberate exception — its host clients MUST +// use the fixed advertised 49092, so it reuses the provision port. +func TestExposePortsNeverCollideWithProvision(t *testing.T) { + provBase := map[string]int{} + for engine, ov := range engineOverlays { + provBase[engine] = ov.portBase + } + for engine, ports := range exposeEngines { + for _, ep := range ports { + if pb, ok := provBase[engine]; ok && ep.base == pb && engine != "kafka" { + t.Errorf("engine %q expose base %d collides with provision base %d", engine, ep.base, pb) + } + } + } + // Every expose base must be unique across all engines/ports (no two services + // fight for the same host port at allocation time either). + seen := map[int]string{} + for _, ports := range exposeEngines { + for _, ep := range ports { + if prev, ok := seen[ep.base]; ok && prev != ep.purpose { + t.Errorf("expose base %d reused across purposes %q and %q", ep.base, prev, ep.purpose) + } + seen[ep.base] = ep.purpose + } + } +} + +func TestFileExists(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "x") + if fileExists(f) { + t.Error("missing file → false") + } + _ = os.WriteFile(f, []byte("y"), 0o644) + if !fileExists(f) { + t.Error("present file → true") + } + if fileExists(dir) { + t.Error("directory → false") + } +} diff --git a/internal/orchestrate/up.go b/internal/orchestrate/up.go index 4b66b0f..3c4350c 100644 --- a/internal/orchestrate/up.go +++ b/internal/orchestrate/up.go @@ -427,6 +427,7 @@ func sharedPhase(d UpDeps, projects, names, provInstances []string) Phase { // Publish each provisioned Postgres on 127.0.0.1: via an // up-time overlay so host-side pgx (the provision phase) can reach it, // without touching the deterministic generated compose. + var overrides []string if len(prov) > 0 { ports := map[string]int{} for _, inst := range prov { @@ -440,8 +441,17 @@ func sharedPhase(d UpDeps, projects, names, provInstances []string) Phase { if err != nil { return nil, err } - cp.Overrides = []string{overlay} + overrides = append(overrides, overlay) } + // Re-apply a prior `shared expose` so GUI-client host ports persist + // across up/down (its 5xxxx range never collides with provisioning's + // 4xxxx). Skipped on a remote backend (bridge is not host-routable). + if d.Backend.Reachability() != docker.ViaProxy { + if p := exposeOverlayPath(d.Model.Root); fileExists(p) { + overrides = append(overrides, p) + } + } + cp.Overrides = overrides if err := cp.Up(ctx, names...); err != nil { return nil, fmt.Errorf("compose up shared: %w", err) } From bba9a2d785827ab42040c88c3443dc4fdb2190c6 Mon Sep 17 00:00:00 2001 From: Gustavo Bertoi Date: Wed, 1 Jul 2026 15:43:26 -0300 Subject: [PATCH 3/3] test(e2e): command-surface release gate (db/s3/expose/localstack) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner directive: every command we built must actually work against a live stack with solid tests before a release — the coverage that would have caught the shared-localstack health deadlock and the `db create` connect race before they reached main. Neither was caught by unit/golden tests because nothing drove these commands end to end against a real daemon. Adds tests/e2e/commands_test.go, two tiers: - TestE2E_Commands_CoreResources (postgres + minio + redis): up (all three health-gated) → shared status → db create/list/drop → s3 mb/ls/rb → resource list → messaging degrades cleanly with no engine → shared expose + a real TCP dial of the published port (what DataGrip needs) → down. Runs under DEVSTACK_E2E=1 (per-PR CI) — the daily-driver commands. - TestE2E_Commands_LocalStackHealthAndAws (adds localstack): the health-gate regression (up must reach [ok], not the [failed] "unhealthy after 1 attempt") + the `aws` shim. Heavy image → gated on DEVSTACK_E2E_CLOUD=1 and run as a nightly release gate (new nightly `e2e-cloud` job). Stacks on #109 (health/connect fixes) and #110 (shared expose) — both are needed for this gate to pass, so they are merged into this branch. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/nightly.yml | 23 ++++ tests/e2e/commands_test.go | 212 ++++++++++++++++++++++++++++++++++ 2 files changed, 235 insertions(+) create mode 100644 tests/e2e/commands_test.go diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 9061288..c587861 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -48,6 +48,29 @@ jobs: - name: nightly gate (fmt-check + vet + cross-build + test-race + determinism) run: make nightly + # Heavy cloud-engine command e2e (localstack + the `aws` shim, plus the full + # db/s3/expose command surface). It pulls large images (localstack), so it runs + # here as a nightly RELEASE GATE rather than on every PR — the per-PR CI already + # runs the lighter postgres+minio+redis command e2e (DEVSTACK_E2E=1). This is the + # "every command actually works against a live stack" gate. + e2e-cloud: + if: github.repository == 'open-source-cloud/devstack' + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version: ${{ env.GO_VERSION }} + check-latest: true + cache: true + - name: docker available + run: docker version + - name: cloud command e2e (localstack health gate + aws shim + db/s3/expose) + env: + DEVSTACK_E2E: "1" + DEVSTACK_E2E_CLOUD: "1" + run: go test -tags=e2e ./tests/e2e/... -run 'Commands|LocalStack' -count=1 -v + # Optional rolling pre-release. Default OFF: enable by setting the repo variable # gh variable set NIGHTLY_PRERELEASE --body true # Produces goreleaser SNAPSHOT artifacts (no real version, no git tag) and uploads diff --git a/tests/e2e/commands_test.go b/tests/e2e/commands_test.go new file mode 100644 index 0000000..0caf48c --- /dev/null +++ b/tests/e2e/commands_test.go @@ -0,0 +1,212 @@ +//go:build e2e + +package e2e + +import ( + "encoding/json" + "net" + "os" + "os/exec" + "strconv" + "strings" + "testing" + "time" +) + +// This file is the command-surface release gate (owner directive): it drives the +// real devstack binary against a live daemon and asserts every data-plane command +// group WORKS end to end — the coverage that would have caught the shared-localstack +// health deadlock and the `db create` connect race before they reached main. +// +// Two tiers: +// - CoreResources (postgres + minio + redis): db / s3 / shared expose+ports — +// the commands a developer hits daily. Runs under DEVSTACK_E2E=1 (per-PR CI). +// - CloudEngines (adds localstack): the localstack health-gate regression + the +// `aws` shim. Heavy image → additionally gated on DEVSTACK_E2E_CLOUD=1 (nightly). + +const wsCoreCloud = `apiVersion: devstack/v1 +kind: Workspace +name: e2ecmd +shared: + postgres: { template: postgres, params: { version: "18" } } + minio: { template: minio } + redis: { template: redis } +projects: + - { name: app, path: app } +` + +const projUsesAll = `apiVersion: devstack/v1 +kind: Project +name: app +services: + web: + template: node.vite + uses: [workspace.shared.postgres, workspace.shared.minio, workspace.shared.redis] +` + +func coreCloudWorkspace() map[string]string { + return map[string]string{"workspace.yaml": wsCoreCloud, "app/devstack.yaml": projUsesAll} +} + +// requireCloudE2E additionally gates the heavy localstack tier. +func requireCloudE2E(t *testing.T) { + t.Helper() + requireDaemon(t) + if os.Getenv("DEVSTACK_E2E_CLOUD") != "1" { + t.Skip("cloud-engine e2e pulls heavy images (localstack); set DEVSTACK_E2E_CLOUD=1 to run") + } +} + +func cleanupSharedStack(t *testing.T, s *sandbox) { + t.Helper() + t.Cleanup(func() { + _, _ = s.tryRun("shared", "expose", "--off") + _, _ = s.tryRun("down") + dockerComposeDown("devstack-shared") + dockerComposeDown("devstack-app") + _ = exec.Command("docker", "network", "rm", "devstack_shared").Run() + }) +} + +// TestE2E_Commands_CoreResources is the daily-driver command gate: bring up a +// postgres+minio+redis stack, then exercise db / s3 / shared expose+ports and +// assert each works (the db path is exactly the one that used to fail with the +// connect-reset race). +func TestE2E_Commands_CoreResources(t *testing.T) { + requireDaemon(t) + s := newSandbox(t, coreCloudWorkspace()) + cleanupSharedStack(t, s) + + // up: three shared engines health-gated (minio via `mc ready`, postgres via + // pg_isready, redis via redis-cli ping) + app's postgres role/db provisioned. + up := s.run(t, "up") + if !strings.Contains(up, "[ok]") || strings.Contains(up, "[failed]") { + t.Fatalf("up did not complete cleanly:\n%s", up) + } + for _, alias := range []string{"shared-postgres", "shared-minio", "shared-redis"} { + if st := s.run(t, "shared", "status"); !strings.Contains(st, alias) { + t.Errorf("shared status missing %s:\n%s", alias, st) + } + } + + // --- db (postgres, in-process pgx) -------------------------------------- + if out := s.run(t, "db", "create", "orders"); !strings.Contains(out, "app_orders") { + t.Errorf("db create did not report app_orders:\n%s", out) + } + // Idempotent second create. + s.run(t, "db", "create", "orders") + if out := s.run(t, "db", "list", "--json"); !strings.Contains(out, "app_orders") { + t.Errorf("db list --json missing app_orders:\n%s", out) + } + s.run(t, "db", "drop", "orders", "--yes") + + // --- s3 (minio, in-process aws-sdk-go-v2) ------------------------------- + if out := s.run(t, "s3", "mb", "uploads"); !strings.Contains(out, "app-uploads") { + t.Errorf("s3 mb did not report app-uploads:\n%s", out) + } + if out := s.run(t, "s3", "ls"); !strings.Contains(out, "app-uploads") { + t.Errorf("s3 ls missing app-uploads:\n%s", out) + } + s.run(t, "s3", "rb", "uploads", "--yes", "--force") + + // --- resource list (the generic surface) -------------------------------- + if out := s.run(t, "resource", "list", "--json"); !strings.Contains(out, "resources") && !strings.Contains(out, "[]") { + t.Errorf("resource list --json not valid:\n%s", out) + } + + // --- messaging degrades cleanly when the engine isn't in the workspace --- + if out, err := s.tryRun("queue", "create", "jobs", "--engine", "nats"); err == nil { + t.Errorf("queue create with no nats engine should fail cleanly, got:\n%s", out) + } + + // --- shared expose + ports (GUI-client host access) --------------------- + if out := s.run(t, "shared", "expose", "postgres"); !strings.Contains(out, "127.0.0.1") { + t.Errorf("shared expose did not print a host address:\n%s", out) + } + assertExposedPortReachable(t, s, "postgres") + s.run(t, "shared", "expose", "--off") + + s.run(t, "down") +} + +// assertExposedPortReachable parses `shared ports --json`, finds the engine's +// primary port, and dials it to prove the publish is live (what DataGrip needs). +func assertExposedPortReachable(t *testing.T, s *sandbox, engine string) { + t.Helper() + out := s.run(t, "shared", "ports", "--json") + var payload struct { + Exposed []struct { + Engine string `json:"engine"` + Port int `json:"port"` + Primary bool `json:"primary"` + } `json:"exposed"` + } + if err := json.Unmarshal([]byte(out), &payload); err != nil { + t.Fatalf("shared ports --json invalid: %v\n%s", err, out) + } + var port int + for _, p := range payload.Exposed { + if p.Engine == engine && p.Primary { + port = p.Port + } + } + if port == 0 { + t.Fatalf("no exposed primary port for %s:\n%s", engine, out) + } + addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) + deadline := time.Now().Add(15 * time.Second) + for { + conn, err := net.DialTimeout("tcp", addr, 2*time.Second) + if err == nil { + _ = conn.Close() + return + } + if time.Now().After(deadline) { + t.Fatalf("exposed %s port %s not reachable: %v", engine, addr, err) + } + time.Sleep(300 * time.Millisecond) + } +} + +// TestE2E_Commands_LocalStackHealthAndAws is the localstack health-gate +// regression: a workspace whose only shared engine is localstack must reach +// [ok] on `up` (the deadlock made this [failed] "unhealthy after 1 attempt"), +// and the `aws` shim must resolve the endpoint and list buckets. +func TestE2E_Commands_LocalStackHealthAndAws(t *testing.T) { + requireCloudE2E(t) + if _, err := exec.LookPath("aws"); err != nil { + t.Skip("aws CLI not installed; the shim test needs it") + } + ws := map[string]string{ + "workspace.yaml": `apiVersion: devstack/v1 +kind: Workspace +name: e2eaws +shared: + localstack: { template: localstack } +projects: + - { name: app, path: app } +`, + "app/devstack.yaml": `apiVersion: devstack/v1 +kind: Project +name: app +services: + web: + template: node.vite + uses: [workspace.shared.localstack] +`, + } + s := newSandbox(t, ws) + cleanupSharedStack(t, s) + + up := s.run(t, "up") + if !strings.Contains(up, "[ok]") || strings.Contains(up, "[failed]") { + t.Fatalf("localstack up did not go healthy (the health-gate regression):\n%s", up) + } + // The aws shim prepends --endpoint-url + dev creds; `s3 ls` on a fresh + // localstack succeeds with empty output. + if _, err := s.tryRun("aws", "--", "s3", "ls"); err != nil { + out, _ := s.tryRun("aws", "--", "s3", "ls") + t.Errorf("aws shim `s3 ls` failed:\n%s", out) + } + s.run(t, "down") +}