diff --git a/internal/orchestrate/orchestrate.go b/internal/orchestrate/orchestrate.go index 0b8ab3f..1dd97ac 100644 --- a/internal/orchestrate/orchestrate.go +++ b/internal/orchestrate/orchestrate.go @@ -102,7 +102,7 @@ func (s *Saga) Run(ctx context.Context, phases []Phase) ([]Record, error) { s.emit(rec) if err != nil { - s.compensate(ctx, done) + s.compensate(ctx, &p, done) return records, fmt.Errorf("phase %q failed: %w", p.Name, err) } if rec.Status == StatusOK && p.Mutating && p.Compensate != nil { @@ -179,25 +179,36 @@ func (s *Saga) runBody(ctx context.Context, p Phase) (detail any, err error) { return p.Run(ctx) } -// compensate unwinds the succeeded mutating phases in reverse, clearing each -// phase row so a re-run redoes it. A compensation error is logged, not fatal — +// compensate unwinds mutating work after a failure. The FAILED phase's own +// compensation runs first (it may have partially applied — e.g. compose-up +// created some containers) but its row is KEPT as failed so `status` can surface +// it. Then the succeeded mutating phases unwind in reverse, each with its row +// cleared so a re-run redoes it. A compensation error is logged, not fatal — // best-effort cleanup must not mask the original failure. -func (s *Saga) compensate(ctx context.Context, done []Phase) { +func (s *Saga) compensate(ctx context.Context, failed *Phase, done []Phase) { + if failed != nil && failed.Mutating && failed.Compensate != nil { + s.runCompensation(ctx, *failed) + } for i := len(done) - 1; i >= 0; i-- { p := done[i] - if p.Compensate != nil { - if err := p.Compensate(ctx); err != nil { - s.DB.LogEvent("saga", s.qualified(p), "compensation failed: "+err.Error()) - } else { - s.DB.LogEvent("saga", s.qualified(p), "compensated") - } - } + s.runCompensation(ctx, p) _ = s.withLock(ctx, func() error { return s.DB.ClearPhase(s.Workspace, p.Scope, p.Name) }) } } +func (s *Saga) runCompensation(ctx context.Context, p Phase) { + if p.Compensate == nil { + return + } + if err := p.Compensate(ctx); err != nil { + s.DB.LogEvent("saga", s.qualified(p), "compensation failed: "+err.Error()) + } else { + s.DB.LogEvent("saga", s.qualified(p), "compensated") + } +} + func (s *Saga) fingerprint(ctx context.Context, p Phase) (string, error) { if p.AlwaysRun || p.Fingerprint == nil { return "", nil diff --git a/internal/orchestrate/up.go b/internal/orchestrate/up.go new file mode 100644 index 0000000..948fbab --- /dev/null +++ b/internal/orchestrate/up.go @@ -0,0 +1,377 @@ +package orchestrate + +import ( + "context" + "fmt" + "os" + "path/filepath" + "sort" + "time" + + "github.com/open-source-cloud/devstack/internal/config" + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" + "github.com/open-source-cloud/devstack/internal/health" + "github.com/open-source-cloud/devstack/internal/hooks" + "github.com/open-source-cloud/devstack/internal/lock" + "github.com/open-source-cloud/devstack/internal/state" + "github.com/open-source-cloud/devstack/internal/template" + "github.com/open-source-cloud/devstack/internal/workspace" +) + +// This file wires the concrete `up` phases over the engine (C5b). It assembles +// the phase list that BuildUp returns; the Saga (orchestrate.go) drives it. +// +// Wired here: preflight → network → generate → shared(health-gated) → +// compose-up(per project) → hooks(postUp, per project), with compensation for +// the shared ref rows and each project's compose-up. +// +// Deliberately deferred (flagged): clone (gitx), provision (needs the +// shared-Postgres host-port coupling — a flagged design item), secrets (M4/S6), +// trust (N5), and firstRun hooks (need the provisioned-volume scope_key). Those +// slot in as additional phases without changing the engine. + +// UpDeps is everything the up-saga phases need. Daemon I/O flows through the +// injected docker.Client + docker.Runner so the wiring is unit-testable with a +// mock client + a fake runner. +type UpDeps struct { + Model *config.Model + DB *state.DB + Docker docker.Client + Manager *workspace.Manager + Source template.TemplateSource + LockPath string + + Runner docker.Runner // compose CLI runner (nil → docker.ExecRunner) + Env map[string]string // generate env (nil → process env, via generate default) + Profile string + Projects []string // explicit subset; empty → every project in the workspace + + Build bool // compose up --build + NoHooks bool // skip the hooks phase + HealthTimeout time.Duration // per-shared-service gate cap (0 → health.Compile default) +} + +// BuildUp assembles the ordered up-saga phases for the requested projects. +func BuildUp(d UpDeps) ([]Phase, error) { + if d.Runner == nil { + d.Runner = docker.ExecRunner{} + } + projects := d.Projects + if len(projects) == 0 { + projects = sortedProjects(d.Model) + } + for _, p := range projects { + if _, ok := d.Model.Projects[p]; !ok { + return nil, fmt.Errorf("project %q is not in this workspace", p) + } + } + + gen, err := generate.New(d.Model, d.Source, generate.WithEnv(d.Env), generate.WithProfile(d.Profile)) + if err != nil { + return nil, err + } + + phases := []Phase{ + preflightPhase(d), + networkPhase(d), + generatePhase(d, gen), + sharedPhase(d, projects), + } + for _, p := range projects { + phases = append(phases, composeUpPhase(d, p)) + } + if !d.NoHooks { + for _, p := range projects { + phases = append(phases, hooksPhase(d, p)) + } + } + return phases, nil +} + +// preflight — daemon reachable (critical). The full doctor matrix is X6. +func preflightPhase(d UpDeps) Phase { + return Phase{ + Name: "preflight", + AlwaysRun: true, + Run: func(ctx context.Context) (any, error) { + if err := d.Docker.Ping(ctx); err != nil { + return nil, fmt.Errorf("docker daemon not reachable: %w", err) + } + return map[string]any{"context": d.Docker.ContextName()}, nil + }, + } +} + +// network — idempotent ensure of the pinned external bridge (must precede any +// compose up). Mutating but never auto-removed (shared by other workspaces). +func networkPhase(d UpDeps) Phase { + return Phase{ + Name: "network", + Mutating: true, + Fingerprint: func(context.Context) (string, error) { + return Fingerprint(generate.SharedNetwork), nil + }, + Run: func(ctx context.Context) (any, error) { + err := lock.WithLock(ctx, d.LockPath, func() error { + return d.Docker.EnsureNetwork(ctx, generate.SharedNetwork, map[string]string{ + generate.LabelManaged: "true", generate.LabelWorkspace: d.Model.Workspace.Name, + }) + }) + if err != nil { + return nil, err + } + return map[string]any{"network": generate.SharedNetwork}, nil + }, + } +} + +// generate — the deterministic pipeline, writeIfChanged. Runs before the compose +// phases (the compose files must exist to `up`); re-armed by a config edit. +func generatePhase(d UpDeps, gen *generate.Generator) Phase { + return Phase{ + Name: "generate", + Fingerprint: func(context.Context) (string, error) { + return configFingerprint(d.Model) + }, + Run: func(context.Context) (any, error) { + stacks, err := gen.GenerateAll() + if err != nil { + return nil, err + } + var written []string + for _, st := range stacks { + res, err := st.Write() + if err != nil { + return nil, fmt.Errorf("write %s: %w", st.Name, err) + } + if res.ComposeChanged { + written = append(written, st.Name) + } + } + return map[string]any{"changed": written}, nil + }, + } +} + +// shared — register ref rows, bring up only the shared services the requested +// projects use, then health-gate them. Compensation drops the ref rows. +func sharedPhase(d UpDeps, projects []string) Phase { + names := sharedNamesUsedBy(d.Model, projects) + return Phase{ + Name: "shared", + Mutating: true, + Fingerprint: func(context.Context) (string, error) { + return Fingerprint(append([]string{"shared"}, names...)...), nil + }, + Run: func(ctx context.Context) (any, error) { + if len(names) == 0 { + return map[string]any{"services": []any{}}, nil + } + for _, p := range projects { + if err := d.Manager.RegisterUp(ctx, p); err != nil { + return nil, fmt.Errorf("register refs for %s: %w", p, err) + } + } + outDir := filepath.Join(d.Model.Root, generate.GenDir, "shared") + cp := docker.Compose{ + Project: generate.SharedStackName, + File: filepath.Join(outDir, generate.ComposeFile), + Dir: outDir, Runner: d.Runner, + } + if err := cp.Up(ctx, names...); err != nil { + return nil, fmt.Errorf("compose up shared: %w", err) + } + gated, err := gateShared(ctx, d, names) + if err != nil { + return nil, err + } + return map[string]any{"services": gated}, nil + }, + Compensate: func(ctx context.Context) error { + for _, p := range projects { + if _, err := d.Manager.RegisterDown(ctx, p); err != nil { + return err + } + } + return nil + }, + } +} + +// gateShared resolves each shared service's container and polls it ready. A +// service with a healthcheck is gated on Healthy; one without, on Started. +func gateShared(ctx context.Context, d UpDeps, names []string) ([]map[string]any, error) { + out := make([]map[string]any, 0, len(names)) + for _, name := range names { + cs, err := d.Docker.ListManaged(ctx, map[string]string{ + generate.LabelManaged: "true", generate.LabelShared: name, + }) + if err != nil { + return nil, fmt.Errorf("locate shared %s: %w", name, err) + } + id := firstRunningID(cs) + if id == "" { + return nil, fmt.Errorf("shared service %s has no running container after compose up", name) + } + cond := health.Started + if det, err := d.Docker.ContainerInspect(ctx, id); err == nil && det.HasHealthcheck() { + cond = health.Healthy + } + tm := health.Compile(nil) + pollCtx := ctx + if d.HealthTimeout > 0 { + var cancel context.CancelFunc + pollCtx, cancel = context.WithTimeout(ctx, d.HealthTimeout) + defer cancel() + } + rec, err := health.Poll(pollCtx, d.Docker, health.Target{ + ContainerID: id, Service: generate.SharedAlias(name), + Project: generate.SharedStackName, Condition: cond, + }, tm) + if err != nil { + return nil, err + } + out = append(out, map[string]any{"name": generate.SharedAlias(name), "health": rec.Status}) + } + return out, nil +} + +// composeUpPhase brings one project stack up. Compensation tears it back down +// (idempotent) — refs are owned by the shared phase, not unwound here. +func composeUpPhase(d UpDeps, project string) Phase { + outDir := filepath.Join(d.Model.ProjectDir(project), generate.GenDir) + cp := func() docker.Compose { + return docker.Compose{ + Project: "devstack-" + project, + File: filepath.Join(outDir, generate.ComposeFile), + Dir: outDir, Runner: d.Runner, + } + } + return Phase{ + Name: "compose-up", + Scope: project, + Mutating: true, + Fingerprint: func(context.Context) (string, error) { + return projectFingerprint(d.Model, project) + }, + Run: func(ctx context.Context) (any, error) { + c := cp() + if d.Build { + if err := c.Build(ctx, false); err != nil { + return nil, err + } + } + if err := c.Up(ctx); err != nil { + return nil, fmt.Errorf("compose up %s: %w", project, err) + } + return map[string]any{"project": project}, nil + }, + Compensate: func(ctx context.Context) error { + c := cp() + return c.Down(ctx, false) // never -v: a failed up must not drop volumes + }, + } +} + +// hooksPhase runs a project's postUp hooks (unconditional). firstRun/postPull +// (idempotent, ledger-keyed) arrive with provision/git wiring. +func hooksPhase(d UpDeps, project string) Phase { + return Phase{ + Name: "hooks", + Scope: project, + AlwaysRun: true, + Run: func(ctx context.Context) (any, error) { + p := d.Model.Projects[project] + if len(p.Hooks.PostUp) == 0 { + return map[string]any{"ran": 0}, nil + } + outDir := filepath.Join(d.Model.ProjectDir(project), generate.GenDir) + runner := &hooks.Runner{ + Execer: hooks.OSExecer{ + BaseDir: d.Model.ProjectDir(project), + Project: "devstack-" + project, + File: filepath.Join(outDir, generate.ComposeFile), + }, + Ledger: d.DB, + Lock: func(ctx context.Context, fn func() error) error { return lock.WithLock(ctx, d.LockPath, fn) }, + } + results, err := runner.RunPhase(ctx, p.Hooks.PostUp, hooks.PhaseOpts{ + Project: project, Phase: "postUp", DefaultOnFailure: hooks.OnAbort, + }) + if err != nil { + return map[string]any{"results": results}, err + } + return map[string]any{"ran": len(results)}, nil + }, + } +} + +// --- helpers --------------------------------------------------------------- + +func firstRunningID(cs []docker.Container) string { + for _, c := range cs { + if c.Running() { + return c.ID + } + } + if len(cs) > 0 { + return cs[0].ID + } + return "" +} + +// sharedNamesUsedBy returns the sorted, de-duplicated shared service names the +// given projects consume via `uses`. +func sharedNamesUsedBy(m *config.Model, projects []string) []string { + seen := map[string]bool{} + var out []string + for _, project := range projects { + p, ok := m.Projects[project] + if !ok { + continue + } + for _, s := range p.Services { + for _, u := range s.Uses { + if ref, ok := config.ParseRef(u); ok && ref.Kind == config.RefShared && !seen[ref.Name] { + seen[ref.Name] = true + out = append(out, ref.Name) + } + } + } + } + sort.Strings(out) + return out +} + +func sortedProjects(m *config.Model) []string { + out := make([]string, 0, len(m.Projects)) + for k := range m.Projects { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// configFingerprint hashes the workspace.yaml + every project's devstack.yaml so +// any config edit re-arms generate. +func configFingerprint(m *config.Model) (string, error) { + parts := []string{readFileOrEmpty(filepath.Join(m.Root, "workspace.yaml"))} + for _, p := range sortedProjects(m) { + parts = append(parts, p, readFileOrEmpty(filepath.Join(m.ProjectDir(p), "devstack.yaml"))) + } + return Fingerprint(parts...), nil +} + +// projectFingerprint re-arms a project's compose-up on an edit to its devstack.yaml. +func projectFingerprint(m *config.Model, project string) (string, error) { + return Fingerprint(project, readFileOrEmpty(filepath.Join(m.ProjectDir(project), "devstack.yaml"))), nil +} + +func readFileOrEmpty(path string) string { + b, err := os.ReadFile(path) + if err != nil { + return "" + } + return string(b) +} diff --git a/internal/orchestrate/up_test.go b/internal/orchestrate/up_test.go new file mode 100644 index 0000000..68ed2db --- /dev/null +++ b/internal/orchestrate/up_test.go @@ -0,0 +1,227 @@ +package orchestrate + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/open-source-cloud/devstack/internal/config" + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" + "github.com/open-source-cloud/devstack/internal/state" + "github.com/open-source-cloud/devstack/internal/template" + "github.com/open-source-cloud/devstack/internal/workspace" + "github.com/open-source-cloud/devstack/templates" +) + +// fakeRunner records compose invocations and can fail selectively. +type fakeRunner struct { + cmds [][]string + fail func(args []string) bool +} + +func (f *fakeRunner) record(name string, args []string) error { + f.cmds = append(f.cmds, append([]string{name}, args...)) + if f.fail != nil && f.fail(args) { + return errors.New("compose failed") + } + return nil +} +func (f *fakeRunner) Run(_ context.Context, _ []string, _, name string, args ...string) error { + return f.record(name, args) +} +func (f *fakeRunner) Output(_ context.Context, _ []string, _, name string, args ...string) ([]byte, error) { + return nil, f.record(name, args) +} +func (f *fakeRunner) sawUp(project string) bool { + for _, c := range f.cmds { + joined := strings.Join(c, " ") + if strings.Contains(joined, "-p "+project) && strings.Contains(joined, " up ") { + return true + } + } + return false +} +func (f *fakeRunner) sawDown(project string) bool { + for _, c := range f.cmds { + joined := strings.Join(c, " ") + if strings.Contains(joined, "-p "+project) && strings.Contains(joined, " down") { + return true + } + } + return false +} + +func upFixture(t *testing.T) (UpDeps, *fakeRunner, *state.DB) { + t.Helper() + root := t.TempDir() + write := func(rel, body string) { + p := filepath.Join(root, rel) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } + write("workspace.yaml", "apiVersion: devstack/v1\nkind: Workspace\nname: demo\nshared:\n postgres: { template: postgres, params: { version: \"16\" } }\nprojects:\n - { name: app, path: app }\n") + write("app/devstack.yaml", `apiVersion: devstack/v1 +kind: Project +name: app +services: + web: + template: node.vite + uses: [workspace.shared.postgres] +hooks: + postUp: + - { name: warm, run: host, command: ["true"], onFailure: warn } +`) + + m, err := config.LoadAt(root) + if err != nil { + t.Fatalf("load: %v", err) + } + db, err := state.Open(context.Background(), filepath.Join(root, "state"), "ctx") + if err != nil { + t.Fatalf("state: %v", err) + } + t.Cleanup(func() { db.Close() }) + + mc := &docker.MockClient{ + Containers: []docker.Container{{ + ID: "pg1", Name: "devstack-shared-postgres-1", State: "running", + Labels: map[string]string{generate.LabelManaged: "true", generate.LabelShared: "postgres"}, + }}, + Details: map[string]docker.ContainerDetails{ + "pg1": {ID: "pg1", State: "running", Running: true, Health: docker.HealthHealthy}, + }, + } + src := template.NewFSSource(templates.FS) + lockPath := filepath.Join(root, "lock") + mgr := &workspace.Manager{Model: m, DB: db, Docker: mc, Source: src, LockPath: lockPath} + fr := &fakeRunner{} + d := UpDeps{ + Model: m, DB: db, Docker: mc, Manager: mgr, Source: src, + LockPath: lockPath, Runner: fr, Env: map[string]string{}, + } + return d, fr, db +} + +func TestBuildUpHappyPath(t *testing.T) { + d, fr, db := upFixture(t) + phases, err := BuildUp(d) + if err != nil { + t.Fatalf("BuildUp: %v", err) + } + saga := &Saga{Workspace: d.Model.Workspace.Name, DB: db, LockPath: d.LockPath} + recs, err := saga.Run(context.Background(), phases) + if err != nil { + t.Fatalf("saga: %v\n%+v", err, recs) + } + if AnyFailed(recs) { + t.Fatalf("a phase failed: %+v", recs) + } + + // Phase coverage: preflight, network, generate, shared, compose-up(app), hooks(app). + got := map[string]string{} + for _, r := range recs { + got[r.Phase+scopeSuffix(r.Scope)] = r.Status + } + for _, want := range []string{"preflight", "network", "generate", "shared", "compose-up@app", "hooks@app"} { + if got[want] != StatusOK { + t.Errorf("phase %q = %q, want ok (all: %+v)", want, got[want], got) + } + } + + // Network ensured. + if ok, _ := d.Docker.(*docker.MockClient).NetworkExists(context.Background(), generate.SharedNetwork); !ok { + t.Error("shared network was not ensured") + } + // Ref row added for the shared instance. + if n, _ := db.RefCount("shared-postgres"); n != 1 { + t.Errorf("ref count for shared-postgres = %d, want 1", n) + } + // Compose up ran for both stacks. + if !fr.sawUp(generate.SharedStackName) { + t.Error("did not compose up the shared stack") + } + if !fr.sawUp("devstack-app") { + t.Error("did not compose up the project stack") + } + // Generated compose files exist on disk. + if _, err := os.Stat(filepath.Join(d.Model.Root, generate.GenDir, "shared", generate.ComposeFile)); err != nil { + t.Errorf("shared compose not written: %v", err) + } + + // Re-run is all skips (except AlwaysRun preflight/hooks). + recs2, err := saga.Run(context.Background(), phases) + if err != nil { + t.Fatal(err) + } + for _, r := range recs2 { + switch r.Phase { + case "preflight", "hooks": + if r.Status != StatusOK { + t.Errorf("%s should re-run ok, got %q", r.Phase, r.Status) + } + default: + if r.Status != StatusSkipped { + t.Errorf("%s should skip on re-run, got %q", r.Phase, r.Status) + } + } + } +} + +func TestBuildUpCompensatesOnProjectFailure(t *testing.T) { + d, fr, db := upFixture(t) + // Fail the PROJECT compose up (not the shared one). + fr.fail = func(args []string) bool { + joined := strings.Join(args, " ") + return strings.Contains(joined, "-p devstack-app") && strings.Contains(joined, "up") + } + phases, err := BuildUp(d) + if err != nil { + t.Fatal(err) + } + saga := &Saga{Workspace: d.Model.Workspace.Name, DB: db, LockPath: d.LockPath} + recs, err := saga.Run(context.Background(), phases) + if err == nil { + t.Fatal("expected the saga to fail at compose-up") + } + // compose-up failed. + var failed string + for _, r := range recs { + if r.Status == StatusFailed { + failed = r.Phase + } + } + if failed != "compose-up" { + t.Errorf("failed phase = %q, want compose-up", failed) + } + // The shared phase compensated → ref rows dropped back to zero. + if n, _ := db.RefCount("shared-postgres"); n != 0 { + t.Errorf("ref count after compensation = %d, want 0", n) + } + // The failed project's stack was torn down (compose-up compensation). + if !fr.sawDown("devstack-app") { + t.Error("failed project stack was not composed down on compensation") + } +} + +func TestBuildUpUnknownProject(t *testing.T) { + d, _, _ := upFixture(t) + d.Projects = []string{"ghost"} + if _, err := BuildUp(d); err == nil { + t.Fatal("BuildUp should reject an unknown project") + } +} + +func scopeSuffix(scope string) string { + if scope == "" { + return "" + } + return "@" + scope +}