From e7be7e02cce47154578984c9a41c825d752bc2b3 Mon Sep 17 00:00:00 2001 From: Gael Date: Mon, 2 Nov 2026 18:54:22 -0500 Subject: [PATCH] fix(run): retain hosted frontend reload and dependency installation state Publish the previously exercised owner implementation for draft review. --- cmd/run/extend.go | 3 + cmd/run/frontend_delivery_lifecycle_test.go | 246 ++++++++++++ cmd/run/frontend_reload_test.go | 423 ++++++++++++++++++++ cmd/run/launcher.go | 108 ++++- cmd/run/launcher_admission.go | 38 +- cmd/run/launcher_admission_test.go | 17 +- cmd/run/launcher_frontend_watch.go | 127 ++++++ cmd/run/launcher_lane.go | 158 ++++++++ cmd/run/native_delivery_schedule.go | 11 + cmd/run/recovery_scope_test.go | 54 +++ cmd/run/service.go | 44 +- cmd/run/solution.go | 1 + cmd/run/workspace.go | 1 + docs/cli-reference.md | 3 + docs/commands.md | 26 +- docs/module-manifests.md | 46 +++ pkg/engine/host.go | 21 +- pkg/engine/host_recovery_scope_test.go | 73 ++++ pkg/engine/host_test.go | 51 ++- pkg/gitops/native_applied.go | 73 +++- pkg/gitops/native_http_applied_test.go | 74 ++++ pkg/gitops/wire_kits_test.go | 18 +- pkg/solutionrun/frontend_local.go | 81 ++-- pkg/solutionrun/frontend_local_test.go | 255 +++++++++++- pkg/solutionrun/frontend_watch.go | 83 ++++ pkg/solutionrun/frontend_watch_test.go | 141 +++++++ 26 files changed, 2093 insertions(+), 83 deletions(-) create mode 100644 cmd/run/frontend_delivery_lifecycle_test.go create mode 100644 cmd/run/frontend_reload_test.go create mode 100644 cmd/run/launcher_frontend_watch.go create mode 100644 cmd/run/launcher_lane.go create mode 100644 cmd/run/recovery_scope_test.go create mode 100644 pkg/engine/host_recovery_scope_test.go create mode 100644 pkg/solutionrun/frontend_watch.go create mode 100644 pkg/solutionrun/frontend_watch_test.go diff --git a/cmd/run/extend.go b/cmd/run/extend.go index f690bd7d8..adb453351 100644 --- a/cmd/run/extend.go +++ b/cmd/run/extend.go @@ -216,6 +216,9 @@ func extendRunningWorkspace(ctx context.Context, flow *orchestration.Flow, basel if admitErr != nil { return admitErr } + if err := cluster.refreshFrontendWatches(ctx); err != nil { + return err + } for _, unique := range admitted { cli.Info("admitted %s: its identity is projected, and it is delivered to the host once running", unique) } diff --git a/cmd/run/frontend_delivery_lifecycle_test.go b/cmd/run/frontend_delivery_lifecycle_test.go new file mode 100644 index 000000000..1076e8f09 --- /dev/null +++ b/cmd/run/frontend_delivery_lifecycle_test.go @@ -0,0 +1,246 @@ +package run + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + compositionv1 "github.com/codefly-dev/authority/libraries/authority-client/go/gen/host/composition/v1" + "github.com/codefly-dev/cli/pkg/gitops" + "github.com/codefly-dev/core/resources" + "github.com/codefly-dev/core/solutionhost" + "github.com/stretchr/testify/require" +) + +func TestBatchStopsStaleRemainingSendAfterRuntimeCompletion(t *testing.T) { + cluster, peer := admissionLauncher(t) + service, err := cluster.composed().LoadService(t.Context(), &resources.ServiceWithModule{Module: "beta", Name: "api"}) + require.NoError(t, err) + var once sync.Once + peer.intercept = func(ctx context.Context, _ *compositionv1.ApplyManifestRequest, m *solutionhost.ModuleManifest) error { + if m.Alias == "alpha" { + once.Do(func() { + attempt, err := cluster.beginRuntimeStart(ctx, service) + require.NoError(t, err) + attempt.Complete(true) + }) + } + return nil + } + require.NoError(t, testDelivery(hostAlwaysReady, cluster.admitSelected).run(t.Context())) + for _, m := range peer.snapshots() { + if m.Alias == "beta" { + require.NotEmpty(t, m.Endpoints, "no pre-completion beta manifest may be sent after alpha's network wait") + } + } +} + +func TestEditDuringRenderIsRejectedBeforeFirstSend(t *testing.T) { + cluster, peer := admissionLauncher(t) + file := reloadPage(t, cluster) + var once sync.Once + cluster.flow.(*admissionFlow).beforeCarriers = func(module string) { + if module == "alpha" { + once.Do(func() { require.NoError(t, os.WriteFile(file, []byte("during render"), 0600)) }) + } + } + require.NoError(t, hostDelivery{schedule: cluster.schedule, stop: cluster.stop, ready: hostAlwaysReady, deliver: cluster.admitSelected, log: testDelivery(hostAlwaysReady, nil).log}.run(t.Context())) + var alpha []*solutionhost.ModuleManifest + for _, m := range peer.snapshots() { + if m.Alias == "alpha" { + alpha = append(alpha, m) + } + } + require.Len(t, alpha, 1, "stale rendered candidate must never reserve or send") + require.EqualValues(t, 1, alpha[0].Generation) +} + +func TestFrontendChangeDuringSendConvergesAndConcurrentAdmissionsSerialize(t *testing.T) { + cluster, peer := admissionLauncher(t) + file := reloadPage(t, cluster) + require.NoError(t, cluster.admitSelected(t.Context())) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + peer.mu.Lock() + peer.intercept = func(ctx context.Context, _ *compositionv1.ApplyManifestRequest, m *solutionhost.ModuleManifest) error { + if m.Alias == "alpha" { + once.Do(func() { + close(entered) + select { + case <-release: + case <-ctx.Done(): + } + }) + } + return ctx.Err() + } + peer.mu.Unlock() + require.NoError(t, os.WriteFile(file, []byte("first edit"), 0600)) + select { + case <-entered: + case <-time.After(8 * time.Second): + t.Fatal("frontend delivery did not enter host") + } + inFlight := latestAlpha(peer) + require.NoError(t, os.WriteFile(file, []byte("second edit"), 0600)) + close(release) + require.Eventually(t, func() bool { + m := latestAlpha(peer) + return m.Generation > inFlight.Generation && m.Frontend.Payload.Digest != inFlight.Frontend.Payload.Digest + }, 8*time.Second, 20*time.Millisecond) + done := make(chan error, 2) + for i := 0; i < 2; i++ { + go func() { done <- testDelivery(hostAlwaysReady, cluster.admitSelected).run(t.Context()) }() + } + require.NoError(t, <-done) + require.NoError(t, <-done) +} + +func TestPartialAndFullRemovalFenceFrontendAndShutdownDrains(t *testing.T) { + cluster, peer := admissionLauncher(t) + file := reloadPage(t, cluster) + workspace := composeService(t, cluster.composed(), "alpha", "worker") + _, err := cluster.admit(t.Context(), workspace) + require.NoError(t, err) + flow := cluster.flow.(*admissionFlow) + flow.selected = append(flow.selected, "alpha/worker") + require.NoError(t, cluster.admitSelected(t.Context())) + _, err = cluster.withdraw(t.Context(), []string{"alpha/worker"}) + require.NoError(t, err) + cluster.mu.Lock() + require.False(t, cluster.laneLocked("alpha").terminal) + cluster.mu.Unlock() + require.False(t, latestAlpha(peer).Removed) + _, err = cluster.withdraw(t.Context(), []string{"alpha/api"}) + require.NoError(t, err) + require.True(t, latestAlpha(peer).Removed) + before := len(peer.snapshots()) + require.NoError(t, os.WriteFile(file, []byte("after withdrawal"), 0600)) + time.Sleep(800 * time.Millisecond) + require.Len(t, peer.snapshots(), before) + require.ErrorIs(t, cluster.deliver(t.Context(), "alpha"), errWorkloadGone) + cluster.close() + require.ErrorIs(t, cluster.admitSelected(t.Context()), errRunStopped) + require.Len(t, peer.snapshots(), before, "no delivery after close") +} + +func TestShutdownCancelsAdmittedHostCallBeforeClosingTransport(t *testing.T) { + cluster, peer := admissionLauncher(t) + entered := make(chan struct{}) + var once sync.Once + peer.intercept = func(ctx context.Context, _ *compositionv1.ApplyManifestRequest, _ *solutionhost.ModuleManifest) error { + once.Do(func() { close(entered) }) + <-ctx.Done() + return ctx.Err() + } + done := make(chan error, 1) + go func() { done <- cluster.admitSelected(t.Context()) }() + <-entered + cluster.close() + require.Error(t, <-done) + before := len(peer.snapshots()) + require.ErrorIs(t, cluster.deliver(t.Context(), "alpha"), errRunStopped) + require.Len(t, peer.snapshots(), before) +} + +func TestLegacyHTTPUncertainDeliveryAndAppliedReplaySameReservation(t *testing.T) { + for _, failAt := range []string{"deliver", "applied", "withheld", "forbidden", "superseded", "removed"} { + t.Run(failAt, func(t *testing.T) { + cluster, peer := admissionLauncher(t) + require.NoError(t, cluster.admitSelected(t.Context())) + manifest := *latestAlpha(peer) + var mu sync.Mutex + var generations []uint64 + var gets atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + data, err := io.ReadAll(r.Body) + require.NoError(t, err) + signed, err := solutionhost.ParseSigned(data) + require.NoError(t, err) + require.NotNil(t, signed.Envelope) + m, err := solutionhost.ManifestFromVerified(signed.Document) + require.NoError(t, err) + mu.Lock() + generations = append(generations, m.Generation) + n := len(generations) + mu.Unlock() + if failAt == "deliver" && n == 1 { + w.WriteHeader(503) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"disposition": "accepted", "generation": m.Generation}) + return + } + n := gets.Add(1) + if failAt == "applied" && n == 1 { + w.WriteHeader(503) + return + } + if failAt == "forbidden" { + w.WriteHeader(403) + return + } + mu.Lock() + generation := generations[len(generations)-1] + mu.Unlock() + observed := manifest + observed.Generation = generation + digest, err := observed.Digest() + require.NoError(t, err) + answer := map[string]any{"binding": manifest.Binding, "generation": generation, "digest": digest, "release": manifest.Release.Identity(), "removed": false} + if failAt == "superseded" { + answer["generation"] = generation + 1 + } + if failAt == "removed" { + answer["removed"] = true + } + if failAt == "withheld" { + answer["pending"] = map[string]any{"generation": generation, "reason": "consent required"} + } + _ = json.NewEncoder(w).Encode(answer) + })) + defer server.Close() + d := &gitops.NativeDeliverer{Signer: cluster.signer, StateDir: t.TempDir(), DeliveryURL: server.URL, Token: func(string) (string, error) { return "writer", nil }, ObserverToken: func() (string, error) { return "observer", nil }} + err := testDelivery(hostAlwaysReady, func(ctx context.Context) error { + if _, err := d.Deliver(ctx, &manifest); err != nil { + return err + } + return d.WaitForApplied(ctx, &manifest) + }).run(t.Context()) + mu.Lock() + defer mu.Unlock() + if failAt == "withheld" || failAt == "forbidden" || failAt == "superseded" || failAt == "removed" { + require.Error(t, err) + require.Len(t, generations, 1) + } else { + require.NoError(t, err) + require.Len(t, generations, 2) + require.Equal(t, generations[0], generations[1]) + } + }) + } +} + +func TestFrontendWatchFlagsAreExplicitAndDefaultOff(t *testing.T) { + for _, cmd := range []string{"service", "solution", "workspace"} { + var value string + switch cmd { + case "service": + value = ServiceCmd.Flags().Lookup("watch-frontends").DefValue + case "solution": + value = SolutionCmd.Flags().Lookup("watch-frontends").DefValue + case "workspace": + value = WorkspaceCmd.Flags().Lookup("watch-frontends").DefValue + } + require.Equal(t, "false", value) + } + require.Nil(t, ClusterCmd.Flags().Lookup("watch-frontends")) +} diff --git a/cmd/run/frontend_reload_test.go b/cmd/run/frontend_reload_test.go new file mode 100644 index 000000000..65e1694aa --- /dev/null +++ b/cmd/run/frontend_reload_test.go @@ -0,0 +1,423 @@ +package run + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + compositionv1 "github.com/codefly-dev/authority/libraries/authority-client/go/gen/host/composition/v1" + "github.com/codefly-dev/cli/pkg/gitops" + "github.com/codefly-dev/cli/pkg/orchestration" + "github.com/codefly-dev/cli/pkg/solutionrun" + "github.com/codefly-dev/core/agents/manager" + corecomposition "github.com/codefly-dev/core/composition" + "github.com/codefly-dev/core/resources" + coreservices "github.com/codefly-dev/core/services" + "github.com/codefly-dev/core/solutionhost" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func frontendPage(t *testing.T, cluster *launcher) string { + t.Helper() + dir := filepath.Join(cluster.composed().Dir(), "modules", "alpha") + moduleFile := filepath.Join(dir, resources.ModuleConfigurationName) + body, err := os.ReadFile(moduleFile) + require.NoError(t, err) + require.NoError(t, os.WriteFile(moduleFile, append([]byte("service-entry: api\n"), body...), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "services/api", solutionrun.FrontendFileName), []byte("title: Alpha\norder: 1\nexposed_module: ./Page\nhost_contract: 1\n"), 0600)) + remote := filepath.Join(dir, "services/fe-remote") + require.NoError(t, os.MkdirAll(remote, 0755)) + failed, _ := json.Marshal(filepath.Join(cluster.stateDir, "frontend-failed")) + for name, content := range map[string]string{ + "package.json": `{"private":true,"scripts":{"build":"node build.cjs"}}`, + "Page.tsx": "before", + "build.cjs": `const fs=require('fs');const text=fs.readFileSync('Page.tsx','utf8');if(text==='broken'){fs.writeFileSync(` + string(failed) + `,'failed');process.exit(1);}fs.mkdirSync('dist',{recursive:true});fs.writeFileSync('dist/mf-manifest.json',JSON.stringify({text}));`, + } { + require.NoError(t, os.WriteFile(filepath.Join(remote, name), []byte(content), 0600)) + } + t.Setenv("npm_config_cache", t.TempDir()) + t.Setenv("npm_config_offline", "true") + t.Setenv("npm_config_audit", "false") + t.Setenv("npm_config_fund", "false") + workspace, err := resources.LoadWorkspaceFromDir(t.Context(), cluster.composed().Dir()) + require.NoError(t, err) + cluster.workspace = workspace + return filepath.Join(remote, "Page.tsx") +} + +func reloadPage(t *testing.T, cluster *launcher) string { + t.Helper() + file := frontendPage(t, cluster) + cluster.watchFrontends = true + cluster.schedule.Deadline = 15 * time.Second + require.NoError(t, cluster.refreshFrontendWatches(t.Context())) + return file +} + +func latestAlpha(peer *admissionPeer) *solutionhost.ModuleManifest { + var latest *solutionhost.ModuleManifest + for _, m := range peer.snapshots() { + if m.Alias == "alpha" { + latest = m + } + } + return latest +} + +func TestFrontendEditBuildsAndDeliversWithoutRuntimeStart(t *testing.T) { + cluster, peer := admissionLauncher(t) + file := reloadPage(t, cluster) + require.NoError(t, cluster.admitSelected(t.Context())) + first := latestAlpha(peer) + require.NotNil(t, first.Frontend.Payload) + require.NoError(t, os.WriteFile(file, []byte("after"), 0600)) + require.Eventually(t, func() bool { + m := latestAlpha(peer) + return m.Generation > first.Generation && m.Frontend.Payload.Digest != first.Frontend.Payload.Digest + }, 8*time.Second, 20*time.Millisecond) + cluster.mu.Lock() + require.Empty(t, cluster.running, "a frontend edit cannot fabricate a Start") + cluster.mu.Unlock() + // The signed payload identifies a real retained archive, not a source-only claim. + last := latestAlpha(peer) + digest := string(last.Frontend.Payload.Digest) + require.FileExists(t, filepath.Join(cluster.composed().Dir(), ".codefly/selection-artifacts", digest[len("sha256:"):])) + archive, err := os.ReadFile(filepath.Join(cluster.composed().Dir(), ".codefly/selection-artifacts", digest[len("sha256:"):])) + require.NoError(t, err) + extracted := t.TempDir() + require.NoError(t, corecomposition.ExtractArchive(t.Context(), archive, extracted)) + payload, err := os.ReadFile(filepath.Join(extracted, "mf-manifest.json")) + require.NoError(t, err) + var document struct{ Text string } + require.NoError(t, json.Unmarshal(payload, &document)) + require.Equal(t, "after", document.Text) + require.Eventually(t, func() bool { + cluster.mu.Lock() + defer cluster.mu.Unlock() + return !cluster.laneLocked("alpha").scheduled + }, time.Second, 10*time.Millisecond) + prior := len(peer.snapshots()) + require.NoError(t, os.MkdirAll(filepath.Join(filepath.Dir(file), "dist"), 0755)) + require.NoError(t, os.WriteFile(filepath.Join(filepath.Dir(file), "dist/mf-manifest.json"), []byte("output only"), 0600)) + time.Sleep(900 * time.Millisecond) + require.Len(t, peer.snapshots(), prior, "unconditional build output must cause no delivery") +} + +func TestPendingFrontendAdmissionRepairsAndTerminalDoesNotRestart(t *testing.T) { + for _, terminal := range []bool{false, true} { + t.Run(fmt.Sprint(terminal), func(t *testing.T) { + cluster, peer := admissionLauncher(t) + file := reloadPage(t, cluster) + require.NoError(t, os.WriteFile(file, []byte("broken"), 0600)) + if terminal { + cluster.schedule.Deadline = 300 * time.Millisecond + } + service, err := cluster.composed().LoadService(t.Context(), &resources.ServiceWithModule{Module: "alpha", Name: "api"}) + require.NoError(t, err) + attempt, err := cluster.beginRuntimeStart(t.Context(), service) + require.NoError(t, err) + done := make(chan error, 1) + go func() { done <- attempt.Admit(t.Context()) }() + if terminal { + require.Error(t, <-done) + attempt.Complete(false) + require.NoError(t, os.WriteFile(file, []byte("repaired"), 0600)) + time.Sleep(900 * time.Millisecond) + require.Empty(t, peer.snapshots(), "repair after terminal admission cannot mint another Start/Admit") + } else { + // Observe failure in the actual npm build before repairing its source. + require.Eventually(t, func() bool { _, err := os.Stat(filepath.Join(cluster.stateDir, "frontend-failed")); return err == nil }, 5*time.Second, 10*time.Millisecond) + require.NoError(t, os.WriteFile(file, []byte("repaired"), 0600)) + require.NoError(t, <-done) + require.NotEmpty(t, peer.snapshots()) + attempt.Complete(true) + } + }) + } +} + +func TestBatchRechecksAfterPriorSendAndLostResponseReplays(t *testing.T) { + cluster, peer := admissionLauncher(t) + var calls atomic.Int32 + peer.intercept = func(_ context.Context, req *compositionv1.ApplyManifestRequest, m *solutionhost.ModuleManifest) error { + require.NotNil(t, req.Carrier.Envelope, "real delivery signs the carrier") + if calls.Add(1) == 1 { + cluster.mu.Lock() + cluster.invalidateLocked("beta") + cluster.mu.Unlock() + return status.Error(codes.Unavailable, "response lost after host applied") + } + return nil + } + require.NoError(t, testDelivery(hostAlwaysReady, cluster.admitSelected).run(t.Context())) + sent := peer.snapshots() + require.GreaterOrEqual(t, len(sent), 3) + require.Equal(t, "alpha", sent[0].Alias) + require.Equal(t, "alpha", sent[1].Alias) + require.Equal(t, sent[0].Generation, sent[1].Generation, "lost response replays its reservation") + require.Equal(t, "beta", sent[len(sent)-1].Alias) +} + +func TestAppliedNoProgressRetriesWithinOriginalBound(t *testing.T) { + cluster, peer := admissionLauncher(t) + require.NoError(t, cluster.admitSelected(t.Context())) + manifest := latestAlpha(peer) + synctest.Test(t, func(t *testing.T) { + var observed, attempts int + d := &gitops.NativeDeliverer{Applied: func(context.Context, string) (*gitops.AppliedManifest, bool, error) { + observed++ + if attempts == 1 { + return nil, false, nil + } + digest, err := manifest.Digest() + return &gitops.AppliedManifest{Binding: manifest.Binding, Generation: manifest.Generation, Digest: digest, Release: manifest.Release.Identity()}, true, err + }} + delivery := testDelivery(hostAlwaysReady, func(ctx context.Context) error { attempts++; return d.WaitForApplied(ctx, manifest) }) + delivery.schedule.Deadline = 2 * time.Minute + require.NoError(t, delivery.run(t.Context())) + require.Equal(t, 2, attempts) + require.Greater(t, observed, 40) + }) + for _, code := range []codes.Code{codes.Unavailable, codes.PermissionDenied, codes.Canceled} { + t.Run(code.String(), func(t *testing.T) { + attempts := 0 + d := &gitops.NativeDeliverer{Applied: func(context.Context, string) (*gitops.AppliedManifest, bool, error) { + if attempts == 1 { + return nil, false, status.Error(code, "observation") + } + digest, err := manifest.Digest() + return &gitops.AppliedManifest{Binding: manifest.Binding, Generation: manifest.Generation, Digest: digest, Release: manifest.Release.Identity()}, true, err + }} + err := testDelivery(hostAlwaysReady, func(ctx context.Context) error { + attempts++ + return d.WaitForApplied(ctx, manifest) + }).run(t.Context()) + if code == codes.Unavailable { + require.NoError(t, err) + require.Equal(t, 2, attempts) + } else { + require.Error(t, err) + require.Equal(t, 1, attempts) + } + }) + } +} + +func TestAppliedObservationCannotRenewCallerDeadline(t *testing.T) { + cluster, peer := admissionLauncher(t) + require.NoError(t, cluster.admitSelected(t.Context())) + manifest := latestAlpha(peer) + synctest.Test(t, func(t *testing.T) { + attempts := 0 + d := &gitops.NativeDeliverer{Applied: func(context.Context, string) (*gitops.AppliedManifest, bool, error) { return nil, false, nil }} + delivery := testDelivery(hostAlwaysReady, func(ctx context.Context) error { + attempts++ + return d.WaitForApplied(ctx, manifest) + }) + delivery.schedule.Deadline = 10 * time.Second + started := time.Now() + err := delivery.run(t.Context()) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, 1, attempts) + require.Equal(t, 10*time.Second, time.Since(started)) + var uncertain *gitops.AppliedUnconfirmed + require.False(t, errors.As(err, &uncertain), "caller expiry is terminal, not an internal observation bound") + }) +} + +// The real flow publishes its authority participants during manager initialization. +func TestInitialRunInstallsFrontendWatcherAfterParticipantsPublish(t *testing.T) { + cluster, _ := admissionLauncher(t) + file := frontendPage(t, cluster) + workspace := cluster.composed() + module, err := workspace.LoadModuleFromName(t.Context(), "alpha") + require.NoError(t, err) + service, err := module.LoadServiceFromName(t.Context(), "api") + require.NoError(t, err) + flow, err := orchestration.NewFlow(t.Context(), workspace, module, service, cluster.env, orchestration.RunMode) + require.NoError(t, err) + flow.WithRuntimeContext(resources.RuntimeContextNative) + flow.WithStandAlone(true) + flow.WithExcludeRoot(true) + require.Empty(t, flow.LocalAuthorityServices()) + cluster.flow = flow + cluster.watchFrontends = true + launcherMu.Lock() + previous := launcherActive + launcherActive = cluster + launcherMu.Unlock() + t.Cleanup(func() { + launcherMu.Lock() + launcherActive = previous + launcherMu.Unlock() + _ = flow.Stop() + _ = flow.Shutdown() + }) + require.NoError(t, initializeRunManagers(t.Context(), flow)) + require.Contains(t, flow.LocalAuthorityServices(), "alpha/api") + cluster.mu.Lock() + lane := cluster.laneLocked("alpha") + watching, before := lane.watch != nil, lane.revision + cluster.mu.Unlock() + require.True(t, watching, "selected frontend must already be watched before Load or Start") + require.NoError(t, os.WriteFile(file, []byte("after"), 0600)) + require.Eventually(t, func() bool { cluster.mu.Lock(); defer cluster.mu.Unlock(); return lane.revision > before }, 5*time.Second, 20*time.Millisecond) +} + +// A normal native origin exercises InitManagers with a real CLI protocol peer. +func TestInitialNativeRunWatchesSelectedFrontend(t *testing.T) { + t.Setenv(resources.CodeflyHomeEnv, t.TempDir()) + t.Setenv("CODEFLY_TEST_PEER_ROOT", t.TempDir()) + t.Setenv(manager.AgentSourceEnv, "local") + binary := filepath.Join(t.TempDir(), "protocol-peer") + build := exec.CommandContext(t.Context(), "go", "build", "-o", binary, "./testdata/peer") + build.Dir = "../../pkg/internal/protocoltest" + build.Env = append(os.Environ(), "GOWORK=off") + output, err := build.CombinedOutput() + require.NoError(t, err, "%s", output) + agent := &resources.Agent{Kind: resources.ServiceAgent, Publisher: "example.test", Name: "frontend-startup", Version: "0.0.1"} + path, err := agent.Path(t.Context()) + require.NoError(t, err) + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0700)) + require.NoError(t, os.Symlink(binary, path)) + cluster, _ := admissionLauncher(t) + file := frontendPage(t, cluster) + serviceFile := filepath.Join(cluster.composed().Dir(), "modules", "alpha", "services", "api", resources.ServiceConfigurationName) + declaration := "kind: service\nname: api\nversion: 0.0.0\nmodule-identity: true\nagent:\n kind: codefly:service\n publisher: example.test\n name: frontend-startup\n version: 0.0.1\nendpoints:\n - name: http\n api: http\n visibility: internal\n" + require.NoError(t, os.WriteFile(serviceFile, []byte(declaration), 0600)) + workspace, err := resources.LoadWorkspaceFromDir(t.Context(), cluster.composed().Dir()) + require.NoError(t, err) + cluster.workspace = workspace + module, err := workspace.LoadModuleFromName(t.Context(), "alpha") + require.NoError(t, err) + service, err := module.LoadServiceFromName(t.Context(), "api") + require.NoError(t, err) + flow, err := orchestration.NewFlow(t.Context(), workspace, module, service, cluster.env, orchestration.RunMode) + require.NoError(t, err) + flow.WithRuntimeContext(resources.RuntimeContextNative) + flow.WithStandAlone(true) + flow.WithTemporaryPorts(true) + require.Empty(t, flow.LocalAuthorityServices()) + cluster.flow = flow + cluster.watchFrontends = true + launcherMu.Lock() + previous := launcherActive + launcherActive = cluster + launcherMu.Unlock() + t.Cleanup(func() { + launcherMu.Lock() + launcherActive = previous + launcherMu.Unlock() + _ = flow.Stop() + _ = flow.Shutdown() + for _, key := range flow.AgentCacheKeys() { + coreservices.ClearAgent(key) + } + }) + require.NoError(t, initializeRunManagers(t.Context(), flow)) + require.Contains(t, flow.LocalAuthorityServices(), "alpha/api") + cluster.mu.Lock() + lane := cluster.laneLocked("alpha") + watching, before := lane.watch != nil, lane.revision + cluster.mu.Unlock() + require.True(t, watching, "ordinary selected native service has a frontend watch before Load") + require.NoError(t, os.WriteFile(file, []byte("after"), 0600)) + require.Eventually(t, func() bool { cluster.mu.Lock(); defer cluster.mu.Unlock(); return lane.revision > before }, 5*time.Second, 20*time.Millisecond) +} + +func TestInitialFrontendWatchOptOutAndSetupFailure(t *testing.T) { + for _, tc := range []struct { + name string + enabled, malformed bool + }{ + {name: "disabled"}, {name: "invalid declaration", enabled: true, malformed: true}, + } { + t.Run(tc.name, func(t *testing.T) { + cluster, _ := admissionLauncher(t) + frontendPage(t, cluster) + if tc.malformed { + p := filepath.Join(cluster.composed().Dir(), "modules", "alpha", "services", "api", solutionrun.FrontendFileName) + require.NoError(t, os.WriteFile(p, []byte("title: [invalid\n"), 0600)) + } + workspace := cluster.composed() + module, err := workspace.LoadModuleFromName(t.Context(), "alpha") + require.NoError(t, err) + service, err := module.LoadServiceFromName(t.Context(), "api") + require.NoError(t, err) + flow, err := orchestration.NewFlow(t.Context(), workspace, module, service, cluster.env, orchestration.RunMode) + require.NoError(t, err) + flow.WithRuntimeContext(resources.RuntimeContextNative) + flow.WithStandAlone(true) + flow.WithExcludeRoot(true) + cluster.flow = flow + cluster.watchFrontends = tc.enabled + launcherMu.Lock() + previous := launcherActive + launcherActive = cluster + launcherMu.Unlock() + t.Cleanup(func() { + launcherMu.Lock() + launcherActive = previous + launcherMu.Unlock() + _ = flow.Stop() + _ = flow.Shutdown() + }) + err = initializeRunManagers(t.Context(), flow) + if tc.malformed { + require.Error(t, err) + } else { + require.NoError(t, err) + } + cluster.mu.Lock() + watching := cluster.laneLocked("alpha").watch != nil + cluster.mu.Unlock() + require.False(t, watching) + cluster.close() // cleanup remains safe after both disabled and failed setup + }) + } +} + +func TestInitialFrontendWatchDoesNotStartWhenManagerInitializationFails(t *testing.T) { + t.Setenv(resources.CodeflyHomeEnv, t.TempDir()) + t.Setenv(manager.AgentSourceEnv, "local") + cluster, _ := admissionLauncher(t) + frontendPage(t, cluster) + workspace := cluster.composed() + module, err := workspace.LoadModuleFromName(t.Context(), "alpha") + require.NoError(t, err) + service, err := module.LoadServiceFromName(t.Context(), "api") + require.NoError(t, err) + flow, err := orchestration.NewFlow(t.Context(), workspace, module, service, cluster.env, orchestration.RunMode) + require.NoError(t, err) + flow.WithRuntimeContext(resources.RuntimeContextNative) + flow.WithStandAlone(true) + cluster.flow = flow + cluster.watchFrontends = true + launcherMu.Lock() + previous := launcherActive + launcherActive = cluster + launcherMu.Unlock() + t.Cleanup(func() { + launcherMu.Lock() + launcherActive = previous + launcherMu.Unlock() + _ = flow.Stop() + _ = flow.Shutdown() + }) + require.Error(t, initializeRunManagers(t.Context(), flow), "the deliberately absent agent must fail initialization") + cluster.mu.Lock() + watching := cluster.laneLocked("alpha").watch != nil + cluster.mu.Unlock() + require.False(t, watching, "failed manager initialization must not install watchers") + cluster.close() +} diff --git a/cmd/run/launcher.go b/cmd/run/launcher.go index 8b349a04c..276496852 100644 --- a/cmd/run/launcher.go +++ b/cmd/run/launcher.go @@ -83,7 +83,11 @@ type launcher struct { composition sync.RWMutex hostIdentityAnchor, hostAnchorDir string - mu sync.Mutex + mu sync.Mutex + lanes map[string]*moduleLane + closed bool + active sync.WaitGroup + watchFrontends bool // running are the identity workloads that reached Running: the ones a // module's manifest lists. running map[string]bool @@ -125,6 +129,9 @@ var ( // attachLauncher wires the run as the composition's cluster when its // environment declares a host and the run is native. It is a no-op otherwise. func attachLauncher(ctx context.Context, flow *orchestration.Flow, workspace *resources.Workspace, env *environments.Environment, fixture string) error { + if watchFrontends && (runtimeContext == resources.RuntimeContextContainer || env == nil || env.Host == nil) { + return fmt.Errorf("--watch-frontends requires a native controller with a declared host") + } launcherMu.Lock() defer launcherMu.Unlock() if launcherActive != nil { @@ -135,6 +142,9 @@ func attachLauncher(ctx context.Context, flow *orchestration.Flow, workspace *re if err != nil { return err } + if cluster != nil { + cluster.watchFrontends = watchFrontends + } launcherActive = cluster return nil } @@ -211,10 +221,19 @@ func stopLauncher() { func (cluster *launcher) close() { cluster.stopOnce.Do(func() { + cluster.mu.Lock() + cluster.closed = true + for _, lane := range cluster.lanes { + if lane.watch != nil { + lane.watch.StopWatcher() + } + } if cluster.stop != nil { close(cluster.stop) } + cluster.mu.Unlock() }) + cluster.active.Wait() cluster.host.close() // A run torn down keeps no staged frontend copy: the next run stages and // installs again, and reuses every bundle still retained. @@ -361,7 +380,12 @@ func (cluster *launcher) admit(ctx context.Context, workspace *resources.Workspa cluster.identities[unique] = identities[unique] newcomers[unique] = identities[unique] } + cluster.mu.Lock() cluster.workspace = workspace + for _, identity := range identities { + cluster.invalidateLocked(identity.Module) + } + cluster.mu.Unlock() cluster.declare(newcomers) if cluster.vouch != nil && cluster.hostIdentityAnchor != "" && len(added) > 0 { // The host re-reads the anchor: a newcomer is vouched for from now on. @@ -751,6 +775,35 @@ func (cluster *launcher) withdraw(ctx context.Context, uniques []string) ([]stri // settles to the same generation, which the host replays. A module this run // never delivered has nothing to withdraw. func (cluster *launcher) withdrawModule(ctx context.Context, module string) (bool, error) { + cluster.mu.Lock() + lane := cluster.laneLocked(module) + if !lane.terminal { + lane.terminal = true + cluster.invalidateLocked(module) + if lane.watch != nil { + lane.watch.StopWatcher() + } + } + cluster.mu.Unlock() + schedule := cluster.schedule + if schedule.Deadline == 0 { + schedule = nativeDeliverySchedule + } + var done bool + err := hostDelivery{schedule: schedule, stop: cluster.stop, ready: func(context.Context) error { return nil }, log: wool.Get(ctx), deliver: func(ctx context.Context) error { + var err error + done, err = cluster.withdrawOnce(ctx, module) + return err + }}.run(ctx) + return done, err +} + +func (cluster *launcher) withdrawOnce(ctx context.Context, module string) (bool, error) { + ctx, fences, unlock, err := cluster.lockModules(ctx, []string{module}) + if err != nil { + return false, err + } + defer unlock() if err := cluster.acquireDelivery(ctx); err != nil { return false, err } @@ -772,6 +825,9 @@ func (cluster *launcher) withdrawModule(ctx context.Context, module string) (boo if err != nil { return false, err } + if err := cluster.checkFences(fences, true); err != nil { + return false, err + } receipt, err := deliverer.Deliver(ctx, withdrawal) switch { case errors.Is(err, solutionhost.ErrWithdrawn): @@ -799,14 +855,17 @@ func (cluster *launcher) withdrawModule(ctx context.Context, module string) (boo // run's authority is touched. func (cluster *launcher) forget(unique string) { cluster.composition.Lock() + cluster.mu.Lock() + if identity, ok := cluster.identities[unique]; ok { + cluster.invalidateLocked(identity.Module) + } delete(cluster.identities, unique) + delete(cluster.running, unique) + cluster.mu.Unlock() cluster.composition.Unlock() if cluster.vouch != nil { cluster.vouch.Forget(unique) } - cluster.mu.Lock() - delete(cluster.running, unique) - cluster.mu.Unlock() } // delivers reports whether this run delivers a manifest for an identity's @@ -846,6 +905,18 @@ func (cluster *launcher) deliver(ctx context.Context, module string) error { // no other module's delivery; an early one (after a failed start) builds // nothing. func (cluster *launcher) deliverReadiness(ctx context.Context, module string, early bool) error { + return cluster.reconcileReadiness(ctx, module, early, false) +} + +func (cluster *launcher) reconcileReadiness(ctx context.Context, module string, early, frontendOnly bool) error { + ctx, fences, unlock, err := cluster.lockModules(ctx, []string{module}) + if err != nil { + return err + } + defer unlock() + if err := cluster.checkFences(fences, false); err != nil { + return err + } if len(cluster.declaredIdentities(module)) == 0 { return nil } @@ -854,7 +925,18 @@ func (cluster *launcher) deliverReadiness(ctx context.Context, module string, ea return err } var bundle *solutionhost.FrontendPayload + var inputs string if !early { + inputs, err = cluster.frontendInputDigest(ctx, loaded) + if err != nil { + return err + } + cluster.mu.Lock() + unchanged := cluster.laneLocked(module).inputs == inputs + cluster.mu.Unlock() + if frontendOnly && unchanged { + return nil + } if bundle, err = cluster.frontendBundle(ctx, loaded); err != nil { return err } @@ -882,7 +964,23 @@ func (cluster *launcher) deliverReadiness(ctx context.Context, module string, ea if err != nil { return err } - return cluster.applyDelivery(ctx, deliverer, gatewayURL, module, manifest) + if !early { + if err := cluster.checkFrontendInputs(ctx, loaded, inputs); err != nil { + return err + } + } + if err := cluster.checkFences(fences, false); err != nil { + return err + } + if err := cluster.applyDelivery(ctx, deliverer, gatewayURL, module, manifest); err != nil { + return err + } + if !early { + cluster.mu.Lock() + cluster.laneLocked(module).inputs = inputs + cluster.mu.Unlock() + } + return cluster.checkFences(fences, false) } // renderDelivery renders a loaded module's manifest. An early delivery keeps diff --git a/cmd/run/launcher_admission.go b/cmd/run/launcher_admission.go index 43b35c16e..b94fd5190 100644 --- a/cmd/run/launcher_admission.go +++ b/cmd/run/launcher_admission.go @@ -2,6 +2,7 @@ package run import ( "context" + "errors" "fmt" "sort" @@ -30,6 +31,7 @@ func (cluster *launcher) beginRuntimeStart(ctx context.Context, service *resourc } cluster.mu.Lock() delete(cluster.running, unique) + cluster.invalidateLocked(identity.Module) waiter := cluster.waitFor(unique) var projectErr error if cluster.vouch != nil { @@ -41,6 +43,7 @@ func (cluster *launcher) beginRuntimeStart(ctx context.Context, service *resourc cluster.waitersMu.Lock() current := cluster.waiters[unique] == waiter if current { + cluster.invalidateLocked(identity.Module) if started { cluster.running[unique] = true } else { @@ -96,6 +99,14 @@ func (cluster *launcher) admitSelected(ctx context.Context) error { names = append(names, module) } sort.Strings(names) + ctx, fences, unlock, err := cluster.lockModules(ctx, names) + if err != nil { + return err + } + defer unlock() + if err := cluster.checkFences(fences, false); err != nil { + return err + } // A declared page is part of what a module is: the authority admits a // module's roles only beside its page, so admission names the page's // bundle as a ready delivery does. It is built here, before the delivery @@ -103,13 +114,18 @@ func (cluster *launcher) admitSelected(ctx context.Context) error { // reuses the bundle retained for unchanged inputs. loadedModules := make([]*resources.Module, len(names)) bundles := make([]*solutionhost.FrontendPayload, len(names)) + inputs := make([]string, len(names)) for n, module := range names { loaded, err := cluster.composed().LoadModuleFromName(ctx, module) if err != nil { return err } + inputs[n], err = cluster.frontendInputDigest(ctx, loaded) + if err != nil { + return cluster.waitFrontendRepair(ctx, fences[n], err) + } if bundles[n], err = cluster.frontendBundle(ctx, loaded); err != nil { - return err + return cluster.waitFrontendRepair(ctx, fences[n], err) } loadedModules[n] = loaded } @@ -136,11 +152,29 @@ func (cluster *launcher) admitSelected(ctx context.Context) error { manifests = append(manifests, manifest) } for n, manifest := range manifests { + if err := cluster.checkRoutingSnapshot(ctx); err != nil { + return err + } + for i, loaded := range loadedModules { + if err := cluster.checkFrontendInputs(ctx, loaded, inputs[i]); err != nil { + if errors.Is(err, errDeliveryChanged) { + return err + } + return cluster.waitFrontendRepair(ctx, fences[i], err) + } + } + if err := cluster.checkFences(fences, false); err != nil { + return err + } if err := cluster.applyDelivery(ctx, deliverer, gatewayURL, names[n], manifest); err != nil { return err } + cluster.mu.Lock() + cluster.laneLocked(names[n]).admitted = true + cluster.laneLocked(names[n]).inputs = inputs[n] + cluster.mu.Unlock() } - return nil + return cluster.checkFences(fences, false) } // checkBindingTargets decides, at admission, a module's bindings by where they diff --git a/cmd/run/launcher_admission_test.go b/cmd/run/launcher_admission_test.go index 91e70846d..6c16c7648 100644 --- a/cmd/run/launcher_admission_test.go +++ b/cmd/run/launcher_admission_test.go @@ -34,11 +34,15 @@ import ( type admissionFlow struct { pairedLauncherFlow - selected []string + selected []string + beforeCarriers func(string) } func (f *admissionFlow) LocalAuthorityServices() []string { return f.selected } func (f *admissionFlow) ReadyEndpointCarriers(_ context.Context, module string, _ *basev0.NetworkAccess, ready []string) (map[string]string, error) { + if f.beforeCarriers != nil { + f.beforeCarriers(module) + } carriers := map[string]string{} for _, unique := range ready { ref, _ := resources.ParseServiceWithOptionalModule(unique) @@ -52,16 +56,23 @@ type admissionPeer struct { compositionv1.UnimplementedCompositionServiceServer mu sync.Mutex manifests []*solutionhost.ModuleManifest + intercept func(context.Context, *compositionv1.ApplyManifestRequest, *solutionhost.ModuleManifest) error } -func (p *admissionPeer) ApplyManifest(_ context.Context, req *compositionv1.ApplyManifestRequest) (*compositionv1.ApplyManifestResponse, error) { +func (p *admissionPeer) ApplyManifest(ctx context.Context, req *compositionv1.ApplyManifestRequest) (*compositionv1.ApplyManifestResponse, error) { m, err := solutionhost.ManifestFromVerified(req.GetCarrier().GetDocument()) if err != nil { return nil, err } p.mu.Lock() p.manifests = append(p.manifests, m) + intercept := p.intercept p.mu.Unlock() + if intercept != nil { + if err := intercept(ctx, req, m); err != nil { + return nil, err + } + } return &compositionv1.ApplyManifestResponse{Generation: m.Generation}, nil } @@ -105,7 +116,7 @@ func admissionLauncher(t *testing.T) (*launcher, *admissionPeer) { identities, err := gitops.ResolveNativeIdentities(t.Context(), workspace, env) require.NoError(t, err) flow := &admissionFlow{pairedLauncherFlow: pairedLauncherFlow{fakeLauncherFlow: fakeLauncherFlow{ready: func(string) bool { return true }}, address: listener.Addr().String()}, selected: []string{"model/model", "alpha/api", "beta/api"}} - cluster := &launcher{workspace: workspace, env: env, signer: signer, vouch: vouch, identities: identities, stateDir: t.TempDir(), hostModule: "model", hostService: "model", hostEndpoint: "rest", flow: flow, running: map[string]bool{}, delivered: map[string]*solutionhost.ModuleManifest{}, schedule: testDeliverySchedule, stop: make(chan struct{})} + cluster := &launcher{workspace: workspace, env: env, signer: signer, vouch: vouch, identities: identities, stateDir: t.TempDir(), hostModule: "model", hostService: "model", hostEndpoint: "rest", flow: flow, running: map[string]bool{}, delivered: map[string]*solutionhost.ModuleManifest{}, pendingWithdrawals: map[string]bool{}, schedule: testDeliverySchedule, stop: make(chan struct{})} cluster.declare(identities) t.Cleanup(cluster.close) return cluster, peer diff --git a/cmd/run/launcher_frontend_watch.go b/cmd/run/launcher_frontend_watch.go new file mode 100644 index 000000000..06825c3ad --- /dev/null +++ b/cmd/run/launcher_frontend_watch.go @@ -0,0 +1,127 @@ +package run + +import ( + "context" + "fmt" + + "github.com/codefly-dev/cli/pkg/composition" + "github.com/codefly-dev/cli/pkg/solutionrun" + "github.com/codefly-dev/core/agents/helpers/code" + "github.com/codefly-dev/core/agents/services" + "github.com/codefly-dev/core/builders" + "github.com/codefly-dev/core/wool" +) + +var watchFrontends bool + +const watchFrontendsUsage = "Watch selected local frontend sources and rebuild signed host bundles (independent of agent hot-reload; refresh the browser after delivery)" + +func (cluster *launcher) refreshFrontendWatches(ctx context.Context) error { + if !cluster.watchFrontends { + return nil + } + seen := map[string]bool{} + for _, unique := range cluster.flow.LocalAuthorityServices() { + identity, ok := cluster.identity(unique) + if !ok || !cluster.delivers(identity) || seen[identity.Module] { + continue + } + seen[identity.Module] = true + name := identity.Module + cluster.mu.Lock() + lane := cluster.laneLocked(name) + already := lane.watch != nil || lane.terminal || cluster.closed + cluster.mu.Unlock() + if already { + continue + } + selection, err := composition.SelectModule(ctx, cluster.composed(), name) + if err != nil { + return err + } + if selection.Verified { + continue + } + module, err := cluster.composed().LoadModuleFromName(ctx, name) + if err != nil { + return err + } + frontend, err := solutionrun.DeclaredFrontend(ctx, module) + if err != nil { + return err + } + if frontend == nil { + continue + } + tree, files, err := solutionrun.FrontendWatchInterest(ctx, module) + if err != nil { + return err + } + dependencies := builders.NewDependencies("frontend") + for _, file := range files { + dependencies.AddDependencies(builders.NewDependency(file)) + } + base := &services.Base{Location: tree.Root, Wool: wool.Get(ctx).In("frontend.watch", wool.Field("module", name))} + if err := base.SetupWatcher(ctx, services.NewWatchConfiguration(dependencies).WithTrees(tree), func(code.Change) error { cluster.frontendChanged(ctx, name); return nil }); err != nil { + return fmt.Errorf("frontend watch %s: %w", name, err) + } + cluster.mu.Lock() + if cluster.closed || lane.terminal || lane.watch != nil { + base.StopWatcher() + } else { + lane.watch = base + } + cluster.mu.Unlock() + } + return nil +} + +// Observation never waits for a build. The existing module gate serializes all +// producers; scheduled only coalesces source notifications into one caller. +func (cluster *launcher) frontendChanged(ctx context.Context, module string) { + cluster.mu.Lock() + lane := cluster.laneLocked(module) + if cluster.closed || lane.terminal { + cluster.mu.Unlock() + return + } + cluster.invalidateLocked(module) + if !lane.admitted { + cluster.mu.Unlock() + wool.Get(ctx).Info("frontend edit recorded; it can repair a pending admission, otherwise restart the owning run", wool.Field("module", module)) + return + } + cluster.scheduleFrontendLocked(ctx, module, lane) + cluster.mu.Unlock() +} + +func (cluster *launcher) scheduleFrontendLocked(ctx context.Context, module string, lane *moduleLane) { + if lane.scheduled || cluster.closed || lane.terminal { + return + } + lane.scheduled = true + cluster.active.Add(1) + go func() { + defer cluster.active.Done() + w := wool.Get(ctx).In("frontend.reload", wool.Field("module", module)) + var attempted uint64 + err := hostDelivery{schedule: cluster.schedule, stop: cluster.stop, ready: cluster.hostReady, log: w, + deliver: func(ctx context.Context) error { + cluster.mu.Lock() + attempted = lane.revision + cluster.mu.Unlock() + return cluster.reconcileReadiness(ctx, module, false, true) + }, + }.run(context.WithoutCancel(ctx)) + cluster.mu.Lock() + lane.scheduled = false + // Only a later legitimate edit schedules another bounded caller. + if attempted != 0 && attempted != lane.revision { + cluster.scheduleFrontendLocked(ctx, module, lane) + } + cluster.mu.Unlock() + if err != nil && !deliveryAbandoned(err) { + w.Error("frontend delivery unconfirmed; a later source edit retries", wool.ErrField(err)) + } + }() +} diff --git a/cmd/run/launcher_lane.go b/cmd/run/launcher_lane.go new file mode 100644 index 000000000..74088bd66 --- /dev/null +++ b/cmd/run/launcher_lane.go @@ -0,0 +1,158 @@ +package run + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/codefly-dev/cli/pkg/solutionrun" + "github.com/codefly-dev/core/agents/services" + "github.com/codefly-dev/core/resources" + "golang.org/x/sync/semaphore" +) + +var errDeliveryChanged = errors.New("module inputs or lifecycle changed before delivery settled") + +func (cluster *launcher) frontendInputDigest(ctx context.Context, module *resources.Module) (string, error) { + frontend, err := solutionrun.DeclaredFrontend(ctx, module) + if err != nil || frontend == nil { + return "", err + } + return (solutionrun.LocalFrontend{Docker: cluster.flow.ServiceRuntimeContext(resources.ServiceUnique(module.Name, module.ServiceEntry)) == resources.RuntimeContextContainer}).InputDigest(ctx, module) +} + +func (cluster *launcher) checkFrontendInputs(ctx context.Context, module *resources.Module, expected string) error { + current, err := cluster.frontendInputDigest(ctx, module) + if err != nil { + return err + } + if current != expected { + return errDeliveryChanged + } + return nil +} + +// moduleLane orders the existing admission, readiness and withdrawal callers. +// Its revision is only a scheduling fence; NativeDeliverer owns generations. +// All fields except gate are protected by launcher.mu. +type moduleLane struct { + gate *semaphore.Weighted + revision uint64 + changed chan struct{} + terminal, admitted, scheduled bool + inputs string + watch *services.Base +} + +func (cluster *launcher) laneLocked(module string) *moduleLane { + if cluster.lanes == nil { + cluster.lanes = map[string]*moduleLane{} + } + if cluster.lanes[module] == nil { + cluster.lanes[module] = &moduleLane{gate: semaphore.NewWeighted(1), changed: make(chan struct{})} + } + return cluster.lanes[module] +} + +func (cluster *launcher) invalidateLocked(module string) { + lane := cluster.laneLocked(module) + lane.revision++ + close(lane.changed) + lane.changed = make(chan struct{}) +} + +type moduleFence struct { + module string + revision uint64 +} + +// Gates are acquired in one order, without holding state/composition locks. +// Teardown cancels and drains every caller before closing transport or staging. +func (cluster *launcher) lockModules(parent context.Context, modules []string) (context.Context, []moduleFence, func(), error) { + modules = append([]string(nil), modules...) + sort.Strings(modules) + cluster.mu.Lock() + if cluster.closed { + cluster.mu.Unlock() + return parent, nil, nil, errRunStopped + } + cluster.active.Add(1) + lanes := make([]*moduleLane, len(modules)) + for i, name := range modules { + lanes[i] = cluster.laneLocked(name) + } + cluster.mu.Unlock() + ctx, cancel := context.WithCancel(parent) + go func() { + select { + case <-cluster.stop: + cancel() + case <-ctx.Done(): + } + }() + held := 0 + release := func() { + for i := held - 1; i >= 0; i-- { + lanes[i].gate.Release(1) + } + cancel() + cluster.active.Done() + } + for _, lane := range lanes { + if err := lane.gate.Acquire(ctx, 1); err != nil { + release() + return ctx, nil, nil, err + } + held++ + } + cluster.mu.Lock() + fences := make([]moduleFence, len(modules)) + for i, name := range modules { + fences[i] = moduleFence{name, lanes[i].revision} + } + cluster.mu.Unlock() + return ctx, fences, release, nil +} + +// This read is the admission linearization point. Events after it remain dirty; +// they cannot revoke a network request already admitted to flight. +func (cluster *launcher) checkFences(fences []moduleFence, withdrawal bool) error { + cluster.mu.Lock() + defer cluster.mu.Unlock() + if cluster.closed { + return errRunStopped + } + for _, fence := range fences { + lane := cluster.laneLocked(fence.module) + if lane.terminal && !withdrawal { + return errWorkloadGone + } + if !withdrawal && lane.watch != nil && lane.watch.Watcher.Err() != nil { + return fmt.Errorf("frontend observation incomplete for %s: %w", fence.module, lane.watch.Watcher.Err()) + } + if lane.revision != fence.revision { + return errDeliveryChanged + } + } + return nil +} + +func (cluster *launcher) waitFrontendRepair(ctx context.Context, fence moduleFence, cause error) error { + cluster.mu.Lock() + lane := cluster.laneLocked(fence.module) + watching, changed, current := lane.watch != nil, lane.changed, lane.revision + cluster.mu.Unlock() + if !watching { + return cause + } + if current != fence.revision { + return errDeliveryChanged + } + select { + case <-changed: + return errDeliveryChanged + case <-ctx.Done(): + return fmt.Errorf("frontend repair did not complete during the pending admission; restart the owning run after repair: %w (build: %v)", ctx.Err(), cause) + } +} diff --git a/cmd/run/native_delivery_schedule.go b/cmd/run/native_delivery_schedule.go index dd8043066..96069e821 100644 --- a/cmd/run/native_delivery_schedule.go +++ b/cmd/run/native_delivery_schedule.go @@ -12,6 +12,7 @@ import ( "github.com/codefly-dev/cli/pkg/gitops" "github.com/codefly-dev/cli/pkg/orchestration" + "github.com/codefly-dev/cli/pkg/solutionrun" basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0" "github.com/codefly-dev/core/resources" "github.com/codefly-dev/core/wool" @@ -75,6 +76,16 @@ func deliveryAbandoned(err error) bool { // manifest the host refuses, a refusal it could not classify — is its answer, // and fails at once with its reason. func transientDeliveryRefusal(err error) bool { + if errors.Is(err, errDeliveryChanged) || errors.Is(err, solutionrun.ErrFrontendInputsChanged) { + return true + } + if errors.Is(err, context.Canceled) { + return false + } + var uncertain *gitops.AppliedUnconfirmed + if errors.As(err, &uncertain) { + return true + } var refusal *gitops.HostStatusError if errors.As(err, &refusal) { return refusal.Transient() diff --git a/cmd/run/recovery_scope_test.go b/cmd/run/recovery_scope_test.go new file mode 100644 index 000000000..bcf065656 --- /dev/null +++ b/cmd/run/recovery_scope_test.go @@ -0,0 +1,54 @@ +package run + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/codefly-dev/cli/pkg/processgroup" +) + +func TestRunCommandRecoveryWaitsForResolvedWorkspace(t *testing.T) { + fixture, _, _ := loadRecoveryFixture(t, "") + workspace := fixture.Dir() + foreign := t.TempDir() + // Invoke from the service directory: recovery must still select its workspace. + t.Chdir(filepath.Join(workspace, "modules/app/services/api")) + previousReap, previousPins, previousHeadless := reapRunProcesses, pinsAlreadyResolved, headless + previousScopeExplicit := namingScopeExplicit + t.Cleanup(func() { + reapRunProcesses, pinsAlreadyResolved, headless = previousReap, previousPins, previousHeadless + namingScopeExplicit = previousScopeExplicit + }) + t.Setenv(processgroup.LaunchWorkspaceEnv, foreign) + pinsAlreadyResolved, headless = true, true + for _, test := range []struct { + name string + args []string + want int + }{ + {"failure before resolution", []string{"missing/api"}, 0}, + {"failure after resolution", []string{"app/api", "missing/api"}, 2}, + } { + t.Run(test.name, func(t *testing.T) { + calls := 0 + reapRunProcesses = func(_ context.Context, scope processgroup.Scope) (processgroup.CleanupEvidence, error) { + calls++ + if scope.All() || !scope.Includes(workspace) || scope.Includes(foreign) { + t.Fatalf("command recovery scope escaped workspace: %+v", scope) + } + if got := os.Getenv(processgroup.LaunchWorkspaceEnv); got != workspace { + t.Fatalf("resolved launch attribution = %q; want %q", got, workspace) + } + return processgroup.CleanupEvidence{}, nil + } + if err := runServiceCommand(ServiceCmd, test.args); err == nil { + t.Fatal("expected missing service to refuse before graph startup") + } + if calls != test.want { + t.Fatalf("recovery calls = %d; want %d", calls, test.want) + } + }) + } +} diff --git a/cmd/run/service.go b/cmd/run/service.go index 44c4ab20e..d96037db0 100644 --- a/cmd/run/service.go +++ b/cmd/run/service.go @@ -53,6 +53,9 @@ Examples: RunE: runServiceCommand, } +// Keep command-boundary recovery observable in tests without starting a graph. +var reapRunProcesses = processgroup.ReapStaleProcessGroupsWithEvidence + // resolveRunPins materializes the workspace's composed pinned modules unless the // caller already did, which pinsAlreadyResolved records. // @@ -101,13 +104,16 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) { // cleanupRun is the run once it exists, for the deferred cleanup below, // which is registered before it. var cleanupRun *serviceRun + // No automatic recovery is permitted until the composing workspace resolves. + var recoveryWorkspace string cli.Init() defer func() { services.ClearAgents() + if recoveryWorkspace == "" { + return + } cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) - // Machine-wide on purpose: this recovers records whose owner has exited, - // which is garbage collection and never signals a live workspace's run. - evidence, err := processgroup.ReapStaleProcessGroupsWithEvidence(cleanupCtx, processgroup.AllWorkspaces()) + evidence, err := reapRunProcesses(cleanupCtx, processgroup.InWorkspace(recoveryWorkspace)) cleanupCancel() if err != nil { returnErr = errors.Join(returnErr, fmt.Errorf("reap managed processes after run: %w", err)) @@ -134,13 +140,6 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) { cli.StartCapture() } - // Reap any process groups orphaned by a prior ungraceful CLI exit - // (parent SIGKILL, terminal force-closed). Without this, zombie - // trees from a previous `codefly run` survive and hold ports, - // making the next run appear to fork-bomb or fail port binding. - if err := processgroup.ReapStaleProcessGroups(ctx); err != nil { - cli.Warning("stale process-group sweep failed: %v", err) - } // SDK / test runs (`codefly run --cli-server`) spawn per-run dependency // containers under a unique naming scope that are never reused. Mark // them ephemeral BEFORE provisioning so the sweep can reap them even @@ -175,6 +174,16 @@ func runServiceCommand(cmd *cobra.Command, args []string) (returnErr error) { if err != nil { return fmt.Errorf("cannot load required service: %w", err) } + if workspace == nil || workspace.Dir() == "" { + return errors.New("cannot recover run processes without a resolved workspace") + } + recoveryWorkspace = workspace.Dir() + // Imported services can execute outside the composing checkout. Record the + // resolved owner, not merely the directory where this command was invoked. + processgroup.MarkLaunchWorkspace(recoveryWorkspace) + if _, err := reapRunProcesses(ctx, processgroup.InWorkspace(recoveryWorkspace)); err != nil { + cli.Warning("stale process-group sweep failed: %v", err) + } if err = common.WithSilenceE(ctx, workspace, silent); err != nil { return err @@ -1048,7 +1057,7 @@ func initRunService(ctx context.Context, workspace *resources.Workspace, module // leaves live runners the caller must tear down via stopFresh(). Handing back // a nil flow here silently orphaned them — their process groups then survived // until the next run's reaper. - err = flow.InitManagers(ctx) + err = initializeRunManagers(ctx, flow) if err != nil { return flow, w.Wrap(err) } @@ -1059,6 +1068,18 @@ func initRunService(ctx context.Context, workspace *resources.Workspace, module return flow, nil } +func initializeRunManagers(ctx context.Context, flow *orchestration.Flow) error { + if err := flow.InitManagers(ctx); err != nil { + return err + } + // The flow publishes its local authority participants in InitManagers. + // Watching earlier silently installs no watches for the initial run. + if cluster := activeLauncher(); cluster != nil { + return cluster.refreshFrontendWatches(ctx) + } + return nil +} + // parseSetOverrides turns repeatable --set entries of the form // "service:KEY=VAL" into a serviceName -> KEY -> VAL map. It splits on the // first ':' (so the service prefix is unambiguous) and the first '=' (so the @@ -1155,6 +1176,7 @@ func stopService(ctx context.Context, flow *orchestration.Flow) error { } func init() { + ServiceCmd.Flags().BoolVar(&watchFrontends, "watch-frontends", false, watchFrontendsUsage) ServiceCmd.Flags().BoolVar(&withCLIServer, "cli-server", false, "Start CLI server") ServiceCmd.Flags().BoolVar(&openDashboard, "open", false, "Open the dashboard in the default browser (requires --cli-server)") ServiceCmd.Flags().StringVar(&runtimeContext, "runtime-context", defaultRuntimeContext(), "Runtime context for the flow (native/container/nix/free; free picks the first advertised backend)") diff --git a/cmd/run/solution.go b/cmd/run/solution.go index 5e7363116..ed8224d11 100644 --- a/cmd/run/solution.go +++ b/cmd/run/solution.go @@ -161,6 +161,7 @@ func entryUnique(mod *resources.Module) string { } func init() { + SolutionCmd.Flags().BoolVar(&watchFrontends, "watch-frontends", false, watchFrontendsUsage) // Solution-facing subset of the run flags, bound to the same package vars // runServiceCommand reads. The advanced/testing flags (cli-server, …) stay // ServiceCmd-only. diff --git a/cmd/run/workspace.go b/cmd/run/workspace.go index bb826ccb6..68d2e96a0 100644 --- a/cmd/run/workspace.go +++ b/cmd/run/workspace.go @@ -136,6 +136,7 @@ func CompositionRoots(ctx context.Context, workspace *resources.Workspace) ([]st } func init() { + WorkspaceCmd.Flags().BoolVar(&watchFrontends, "watch-frontends", false, watchFrontendsUsage) WorkspaceCmd.Flags().StringVar(&fixture, "fixture", "", "Fixture override (defaults to the selected Codefly environment)") WorkspaceCmd.Flags().StringVar(&environmentName, "env", orchestration.LocalEnvironmentName, "Workspace environment to run") WorkspaceCmd.Flags().BoolVar(&headless, "headless", false, "Run without TUI (auto-enabled when no TTY, e.g. MCP, CI, pipes)") diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 331d74313..75e2d86d7 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -4734,6 +4734,7 @@ Flags: --stand-alone Begin service as standalone, i.e. without its dependencies --start-docker Auto-start a local Docker engine (OrbStack/Docker Desktop/colima/…) if a service needs Docker and it isn't running; --start-docker=false to disable (default true) --temporary-ports show network Run this flow as a disposable invocation: OS-probed ephemeral ports plus a generated naming scope isolating its agents, containers and runtime state (advanced; not previewable via show network). The Codefly SDK sets it for test-owned dependency stacks. Passing --naming-scope wins, and passing it empty asks for no scope at all + --watch-frontends Watch selected local frontend sources and rebuild signed host bundles (independent of agent hot-reload; refresh the browser after delivery) ``` ## `codefly run solution` @@ -4758,6 +4759,7 @@ Flags: --set strings Per-service runtime env override (repeatable), e.g. --set warden:CODEFLY__FIXTURE=dogfood --silent strings Silence services in CLI output --temporary-ports show network Run this flow as a disposable invocation: OS-probed ephemeral ports plus a generated naming scope isolating its agents, containers and runtime state (advanced; not previewable via show network). The Codefly SDK sets it for test-owned dependency stacks. Passing --naming-scope wins, and passing it empty asks for no scope at all + --watch-frontends Watch selected local frontend sources and rebuild signed host bundles (independent of agent hot-reload; refresh the browser after delivery) ``` ## `codefly run workspace` @@ -4787,6 +4789,7 @@ Flags: --silent strings Silence services in CLI output --start-docker Auto-start a local Docker engine if a service needs Docker and it isn't running (default true) --temporary-ports show network Run this flow as a disposable invocation: OS-probed ephemeral ports plus a generated naming scope isolating its agents, containers and runtime state (advanced; not previewable via show network). The Codefly SDK sets it for test-owned dependency stacks. Passing --naming-scope wins, and passing it empty asks for no scope at all + --watch-frontends Watch selected local frontend sources and rebuild signed host bundles (independent of agent hot-reload; refresh the browser after delivery) ``` ## `codefly sbom` diff --git a/docs/commands.md b/docs/commands.md index 2a201bd70..7958caa34 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -298,7 +298,9 @@ A component declaring `solution.frontend.codefly.yaml` has its remote built before native delivery, only when what the build reads changed: its npm project's files, tracked or new (never an untracked `.env`), the local packages it depends on, and its lockfile. It builds a staged copy of exactly -those files with the remote's own `npm ci` (when the lockfile changed) and +those files with the remote's own `npm ci` (`npm install` without a lockfile) +when its manifest, lockfile, npm configuration or local dependency contents +changed, and `npm run build`, natively, or in the pinned Node image when the service owning the page runs in a container, and before the run-wide delivery lock. Failure names the component and includes the build's output; an old dist is never @@ -306,6 +308,10 @@ delivered instead. The complete generated bundle is retained by digest in the existing artifact cache. See [module manifests](module-manifests.md) for source and cache semantics. +A changed installation starts with an empty staged `node_modules`. Its success +marker is separate from the retained bundle: a failed or interrupted install +is retried, and reverting source can still reuse a matching retained bundle. + ### `codefly run service [name...]` Run one or more services locally with their dependency graph. @@ -321,6 +327,24 @@ resolve). `--service-path`, `--stand-alone` and `--exclude-root` each select or carve out a single service, so naming more than one root with any of them is rejected. +Automatic stale-process recovery on entry and exit is limited to the resolved +composing workspace. It retains the existing authenticated-owner checks; it does +not signal groups attributed to another workspace or groups whose workspace is +unknown. A failure before workspace resolution performs no automatic sweep. +The shared workspace host used by MCP and gateway callers applies the same bound +after validating its root; a standalone source tree without a containing workspace +skips automatic recovery. Such orphans need explicitly scoped operator recovery. +This is a workspace boundary, not isolation between sessions sharing that root. +Dead registry metadata may still be reconciled outside it. Explicit machine-wide +maintenance retains its existing behavior. + +Independently embedded hosts still rely on the process-wide launch attribution: +an explicit host root does not yet carry independent provenance into Core's +process registration. Concurrent hosts for different roots can therefore leave +orphans attributed to the wrong workspace, which scoped recovery can skip or +select under that recorded root. This change does not qualify recovery for those +embedders; they need per-launch attribution before relying on that behavior. + ```bash codefly run service api codefly run service lastlogin-go/backend wiki/backend # One graph, shared host started once diff --git a/docs/module-manifests.md b/docs/module-manifests.md index 3e5d2f96d..12e346e6c 100644 --- a/docs/module-manifests.md +++ b/docs/module-manifests.md @@ -137,6 +137,52 @@ manifest names those exact bytes. A build failure refuses delivery with the component name, build command and final 40 lines of build output; it never falls back to a previous bundle. Headless components perform no frontend build. +The local builder verifies that the staged files match the selected input digest +before invoking npm or the container builder. It reselects and hashes the source +after the build; concurrent edits refuse that candidate before retention rather +than labeling old output with a new source identity. A later reconciliation uses +the current source. This check alone does not start a watcher or publish a bundle. + + +### Hosted frontend editing + +`codefly run service`, `run solution` and `run workspace` accept +`--watch-frontends` (off by default). A native controller with a declared host +watches local-source pages in its selected authority closure. Verified immutable +module packages and the host's own page are excluded. A container-owned page +still uses its existing container builder; a fully container controller and +`run cluster` do not support this option. Backend agent `hot-reload` is independent. + +The watcher observes the enclosing repository (component directory outside Git), +pruning only build outputs, plus the exact Git index and exclude files. Invalid +package dependencies retain repair interest. Git configuration outside that tree +is not a live input; restart the run after changing that environment. Growth adds +eligible watches and full removal stops them. Browser refresh loads the delivered +same-origin artifact; this is not browser HMR. A terminally closed filesystem +watch stream requires a controlled run restart; recoverable registration errors +retain their existing retry/rescan behavior. + +Admission, readiness, frontend edits and withdrawal share per-module gates. +Source digests and lifecycle revisions are checked before each signed batch send; +a batch preflight failure sends nothing, but sequential host application is not a +transaction across modules. An in-flight request may apply before a later edit; +the edit remains pending for reconciliation. Existing persisted reservations and +bounded delivery retries handle uncertain responses; receipt and confirmed Applied +remain distinct. Applied success requires the exact binding, generation, digest, +release and live status; a superseding generation is not this delivery succeeding. +Supersession, identity disagreement, policy refusal and explicit withholding are +terminal. Exhaustion +is unconfirmed, and a later legitimate event retries rather than claiming success. + +A failed initial page build can be repaired only while its actual bounded +Start/Admit call remains pending. After terminal timeout/refusal, repair followed +by a controlled owning-run restart is required; a file edit never starts a workload +or invents admission. An already admitted module keeps its last applied page on +build failure, and a later edit retries. Shutdown cancels and drains reconciliation +before closing transport and staging. Delivery does not bypass installation or +consent: adopting a changed release in an existing installation still requires +the host's supported installation lifecycle. + ## Declaring the host The environment declares the host it delivers to. Every field is required, diff --git a/pkg/engine/host.go b/pkg/engine/host.go index 58c6083c0..f870048fd 100644 --- a/pkg/engine/host.go +++ b/pkg/engine/host.go @@ -23,7 +23,7 @@ import ( // a package variable so tests can drive its failure path; production binds the // real reaper. Reaping is best-effort (see NewWorkspaceHost), never a // construction precondition. -var reapStaleProcessGroups = processgroup.ReapStaleProcessGroups +var reapStaleProcessGroups = processgroup.ReapStaleProcessGroupsWithEvidence // Config configures a WorkspaceHost. type Config struct { @@ -55,17 +55,6 @@ func NewWorkspaceHost(cfg Config) (*WorkspaceHost, error) { if root == "" { return nil, fmt.Errorf("workspace host root is required") } - // RECOVERY BOUNDARY: WorkspaceHost is the shared process-owning entry point - // for the CLI, MCP, Gateway gRPC, and Mind's in-process Gateway. Reap PGIDs - // left by dead owners here so every front door self-heals after a crash; the - // reaper preserves groups whose owning process is still alive. - // - // Reaping is best-effort: a failure (signal permission, an unusual runs - // directory, a race with a concurrent host) must not stop every front door - // from starting. Surface it and continue rather than failing construction. - if err := reapStaleProcessGroups(reapContext(cfg.LogWriter)); err != nil { - logHostWarning(cfg.LogWriter, fmt.Sprintf("could not reap stale workspace processes: %v", err)) - } absolute, err := filepath.Abs(root) if err != nil { return nil, fmt.Errorf("resolve workspace host root: %w", err) @@ -78,6 +67,14 @@ func NewWorkspaceHost(cfg Config) (*WorkspaceHost, error) { if info != nil && !info.IsDir() { return nil, fmt.Errorf("workspace host root is not a directory: %s", absolute) } + // A host root may be a nested source tree or a not-yet-created path. Only + // recover after validation, and only when it has a known containing workspace. + // Unknown attribution is not permission to signal another workspace's group. + if workspace, ok := processgroup.CurrentWorkspace(absolute); info != nil && ok { + if _, err := reapStaleProcessGroups(reapContext(cfg.LogWriter), processgroup.InWorkspace(workspace)); err != nil { + logHostWarning(cfg.LogWriter, fmt.Sprintf("could not reap stale workspace processes: %v", err)) + } + } supervisor := NewAgentSupervisor(AgentSupervisorConfig{ Root: absolute, LogWriter: cfg.LogWriter, diff --git a/pkg/engine/host_recovery_scope_test.go b/pkg/engine/host_recovery_scope_test.go new file mode 100644 index 000000000..59778c804 --- /dev/null +++ b/pkg/engine/host_recovery_scope_test.go @@ -0,0 +1,73 @@ +package engine + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/codefly-dev/cli/pkg/processgroup" + "github.com/codefly-dev/core/resources" +) + +func recoveryTestWorkspace(t *testing.T) string { + t.Helper() + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, resources.WorkspaceConfigurationName), []byte("name: recovery-test\n"), 0o600); err != nil { + t.Fatal(err) + } + return dir +} + +func TestHostAutomaticRecoveryRequiresKnownWorkspace(t *testing.T) { + workspace := recoveryTestWorkspace(t) + nested := filepath.Join(workspace, "source") + if err := os.Mkdir(nested, 0o700); err != nil { + t.Fatal(err) + } + alias := filepath.Join(t.TempDir(), "alias") + if err := os.Symlink(nested, alias); err != nil { + t.Fatal(err) + } + file := filepath.Join(workspace, "not-directory") + if err := os.WriteFile(file, nil, 0o600); err != nil { + t.Fatal(err) + } + original := reapStaleProcessGroups + t.Cleanup(func() { reapStaleProcessGroups = original }) + for _, test := range []struct { + name, root string + wantRecovery, wantError bool + }{ + {"workspace", workspace, true, false}, + {"nested source", nested, true, false}, + {"symlink source", alias, true, false}, + {"standalone source", t.TempDir(), false, false}, + {"missing below workspace", filepath.Join(workspace, "absent"), false, false}, + {"empty", "", false, true}, + {"file below workspace", file, false, true}, + } { + t.Run(test.name, func(t *testing.T) { + calls := 0 + reapStaleProcessGroups = func(_ context.Context, scope processgroup.Scope) (processgroup.CleanupEvidence, error) { + calls++ + if scope.All() || !scope.Includes(workspace) || scope.Includes(t.TempDir()) { + t.Fatalf("recovery escaped containing workspace: %+v", scope) + } + return processgroup.CleanupEvidence{}, nil + } + host, err := NewWorkspaceHost(Config{Root: test.root}) + if (err != nil) != test.wantError { + t.Fatalf("construction error = %v", err) + } + if host != nil { + if err := host.Close(); err != nil { + t.Fatal(err) + } + } + if (calls == 1) != test.wantRecovery || calls > 1 { + t.Fatalf("recovery calls = %d; wanted recovery %v", calls, test.wantRecovery) + } + }) + } +} diff --git a/pkg/engine/host_test.go b/pkg/engine/host_test.go index b4cef9b52..a20e919e5 100644 --- a/pkg/engine/host_test.go +++ b/pkg/engine/host_test.go @@ -16,6 +16,7 @@ import ( "testing" "time" + "github.com/codefly-dev/cli/pkg/processgroup" codev0 "github.com/codefly-dev/core/generated/go/codefly/services/code/v0" "github.com/codefly-dev/core/resources" runnersbase "github.com/codefly-dev/core/runners/base" @@ -50,6 +51,8 @@ func TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners(t *testing.T) } } t.Setenv("HOME", t.TempDir()) + workspace := recoveryTestWorkspace(t) + t.Setenv(processgroup.LaunchWorkspaceEnv, workspace) owner := exec.Command(os.Args[0], "-test.run=^TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners$") owner.Env = append(os.Environ(), workspaceHostOrphanHelper+"=owner") @@ -79,8 +82,9 @@ func TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners(t *testing.T) if err := syscall.Kill(-pid, 0); err != nil { t.Fatalf("orphan process group %d was not alive before recovery: %v", pid, err) } + foreignPID := startForeignHostOrphan(t, recoveryTestWorkspace(t)) - host, err := NewWorkspaceHost(Config{Root: t.TempDir()}) + host, err := NewWorkspaceHost(Config{Root: workspace}) if err != nil { t.Fatal(err) } @@ -88,6 +92,9 @@ func TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners(t *testing.T) deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { if err := syscall.Kill(-pid, 0); errors.Is(err, syscall.ESRCH) { + if err := syscall.Kill(-foreignPID, 0); err != nil { + t.Fatalf("another workspace's orphan was signalled: %v", err) + } return } time.Sleep(25 * time.Millisecond) @@ -95,6 +102,36 @@ func TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners(t *testing.T) t.Fatalf("WorkspaceHost did not reap orphan process group %d", pid) } +func startForeignHostOrphan(t *testing.T, workspace string) int { + t.Helper() + owner := exec.Command(os.Args[0], "-test.run=^TestWorkspaceHostReapsProcessGroupsLeftByDeadInProcessOwners$") + owner.Env = append(os.Environ(), workspaceHostOrphanHelper+"=owner", processgroup.LaunchWorkspaceEnv+"="+workspace) + stdout, err := owner.StdoutPipe() + if err != nil { + t.Fatal(err) + } + owner.Stderr = os.Stderr + if err := owner.Start(); err != nil { + t.Fatal(err) + } + line, readErr := bufio.NewReader(stdout).ReadString('\n') + if err := owner.Wait(); err != nil { + t.Fatal(err) + } + if readErr != nil { + t.Fatal(readErr) + } + pid, err := strconv.Atoi(strings.TrimSpace(line)) + if err != nil || pid <= 1 { + t.Fatalf("foreign orphan pid %q: %v", line, err) + } + t.Cleanup(func() { _ = syscall.Kill(-pid, syscall.SIGKILL) }) + if err := syscall.Kill(-pid, 0); err != nil { + t.Fatal(err) + } + return pid +} + func TestWorkspaceHostConstructionSurvivesReaperFailure(t *testing.T) { // Reaping leaked process groups is a best-effort self-heal, not a // construction precondition. A reaper that errors (signal permission, an @@ -102,12 +139,12 @@ func TestWorkspaceHostConstructionSurvivesReaperFailure(t *testing.T) { // host — and therefore the whole front door — from starting. original := reapStaleProcessGroups t.Cleanup(func() { reapStaleProcessGroups = original }) - reapStaleProcessGroups = func(context.Context) error { - return errors.New("simulated reaper failure") + reapStaleProcessGroups = func(context.Context, processgroup.Scope) (processgroup.CleanupEvidence, error) { + return processgroup.CleanupEvidence{}, errors.New("simulated reaper failure") } var logs bytes.Buffer - host, err := NewWorkspaceHost(Config{Root: t.TempDir(), LogWriter: &logs}) + host, err := NewWorkspaceHost(Config{Root: recoveryTestWorkspace(t), LogWriter: &logs}) if err != nil { t.Fatalf("host construction must survive a reaper failure, got: %v", err) } @@ -126,9 +163,9 @@ func TestWorkspaceHostConstructionSurvivesReaperFailure(t *testing.T) { func TestWorkspaceHostReaperNarratesToTheLogSink(t *testing.T) { original := reapStaleProcessGroups t.Cleanup(func() { reapStaleProcessGroups = original }) - reapStaleProcessGroups = func(ctx context.Context) error { + reapStaleProcessGroups = func(ctx context.Context, _ processgroup.Scope) (processgroup.CleanupEvidence, error) { wool.Get(ctx).In("processgroup.reconcile").Info("reaped managed process group") - return nil + return processgroup.CleanupEvidence{}, nil } escaped := &recordingProcessor{} @@ -136,7 +173,7 @@ func TestWorkspaceHostReaperNarratesToTheLogSink(t *testing.T) { t.Cleanup(func() { wool.SetFallbackLogger(nil) }) var logs bytes.Buffer - host, err := NewWorkspaceHost(Config{Root: t.TempDir(), LogWriter: &logs}) + host, err := NewWorkspaceHost(Config{Root: recoveryTestWorkspace(t), LogWriter: &logs}) if err != nil { t.Fatal(err) } diff --git a/pkg/gitops/native_applied.go b/pkg/gitops/native_applied.go index 901f18208..6d6ced445 100644 --- a/pkg/gitops/native_applied.go +++ b/pkg/gitops/native_applied.go @@ -3,6 +3,7 @@ package gitops import ( "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -11,8 +12,22 @@ import ( "time" "github.com/codefly-dev/core/solutionhost" + "google.golang.org/grpc/status" ) +// AppliedUnconfirmed reports observation uncertainty, never a host refusal or +// successful application. The caller may replay within its original deadline. +type AppliedUnconfirmed struct{ Cause error } + +func (e *AppliedUnconfirmed) Error() string { return e.Cause.Error() } +func (e *AppliedUnconfirmed) Unwrap() error { return e.Cause } + +func observationResult(parent context.Context, result *error) { + if parent.Err() == nil && errors.Is(*result, context.DeadlineExceeded) { + *result = &AppliedUnconfirmed{Cause: *result} + } +} + // withdrawalAttempts and withdrawalInterval bound the wait for a delivered // withdrawal to be APPLIED: a delivery receipt only confirms the inbox. const ( @@ -23,7 +38,8 @@ const ( // WaitForWithdrawal observes the host's applied record of a withdrawn binding // through its declared Applied callback or the legacy read-only endpoint, // until it holds the exact withdrawal this run delivered. -func (d *NativeDeliverer) WaitForWithdrawal(ctx context.Context, withdrawal *solutionhost.ModuleManifest) error { +func (d *NativeDeliverer) WaitForWithdrawal(ctx context.Context, withdrawal *solutionhost.ModuleManifest) (result error) { + defer observationResult(ctx, &result) ctx, cancel := context.WithTimeout(ctx, time.Duration(withdrawalAttempts)*withdrawalInterval) defer cancel() if withdrawal == nil || !withdrawal.Removed || (d.Applied == nil && d.ObserverToken == nil) { @@ -50,7 +66,7 @@ func (d *NativeDeliverer) WaitForWithdrawal(ctx context.Context, withdrawal *sol case <-time.After(withdrawalInterval): } } - return fmt.Errorf("binding %s withdrawal generation %d is received but not applied; retained for the next composition refresh", withdrawal.Binding, withdrawal.Generation) + return &AppliedUnconfirmed{Cause: fmt.Errorf("binding %s withdrawal generation %d is received but not applied", withdrawal.Binding, withdrawal.Generation)} } // AppliedManifest is the host's applied identity of a binding, supplied by @@ -72,9 +88,14 @@ type AppliedManifest struct { // WaitForApplied holds a delivery to its outcome: the host applied this // generation, or withheld it and said why. A receipt says the carrier was // accepted into the inbox; only the applied state says the module is served. -func (d *NativeDeliverer) WaitForApplied(ctx context.Context, manifest *solutionhost.ModuleManifest) error { - if manifest == nil || (d.Applied == nil && d.ObserverToken == nil) { - return fmt.Errorf("applied-state observation requires a manifest and an observer") +func (d *NativeDeliverer) WaitForApplied(ctx context.Context, manifest *solutionhost.ModuleManifest) (result error) { + defer observationResult(ctx, &result) + if manifest == nil || manifest.Removed || (d.Applied == nil && d.ObserverToken == nil) { + return fmt.Errorf("applied-state observation requires a live manifest and an observer") + } + digest, err := manifest.Digest() + if err != nil { + return err } ctx, cancel := context.WithTimeout(ctx, time.Duration(withdrawalAttempts)*withdrawalInterval) defer cancel() @@ -83,16 +104,28 @@ func (d *NativeDeliverer) WaitForApplied(ctx context.Context, manifest *solution if err != nil { return err } + if found && applied == nil { + return fmt.Errorf("host reported applied binding %s without its record", manifest.Binding) + } + if found && applied.Binding != manifest.Binding { + return fmt.Errorf("applied record names binding %q, not %s", applied.Binding, manifest.Binding) + } if found && applied.Pending != nil && applied.Pending.Generation >= manifest.Generation { return fmt.Errorf("the host received binding %s generation %d and withheld it: %s", manifest.Binding, applied.Pending.Generation, applied.Pending.Reason) } - if found && applied.Generation >= manifest.Generation { + if found && applied.Generation > manifest.Generation { + return fmt.Errorf("binding %s generation %d was superseded by applied generation %d", manifest.Binding, manifest.Generation, applied.Generation) + } + if found && applied.Generation == manifest.Generation { + if applied.Digest != digest || applied.Release != manifest.Release.Identity() || applied.Removed { + return fmt.Errorf("applied record does not match binding %s generation %d", manifest.Binding, manifest.Generation) + } return nil } if attempt+1 == withdrawalAttempts { - return fmt.Errorf("binding %s generation %d is received and not applied after %s, and the host reports no reason", - manifest.Binding, manifest.Generation, time.Duration(withdrawalAttempts)*withdrawalInterval) + return &AppliedUnconfirmed{Cause: fmt.Errorf("binding %s generation %d is received and not applied after %s, and the host reports no reason", + manifest.Binding, manifest.Generation, time.Duration(withdrawalAttempts)*withdrawalInterval)} } select { case <-ctx.Done(): @@ -105,7 +138,11 @@ func (d *NativeDeliverer) WaitForApplied(ctx context.Context, manifest *solution // appliedState reads the host's applied record of binding; found is false on 404. func (d *NativeDeliverer) appliedState(ctx context.Context, binding string) (*AppliedManifest, bool, error) { if d.Applied != nil { - return d.Applied(ctx, binding) + applied, found, err := d.Applied(ctx, binding) + if ctx.Err() == nil && err != nil && (&HostStatusError{Code: status.Code(err)}).Transient() { + err = &AppliedUnconfirmed{Cause: err} + } + return applied, found, err } token, err := d.ObserverToken() if err != nil { @@ -129,13 +166,20 @@ func (d *NativeDeliverer) appliedState(ctx context.Context, binding string) (*Ap return nil, false, nil } if response.StatusCode != http.StatusOK { - return nil, false, fmt.Errorf("host refused applied-state observation (%s)", response.Status) + return nil, false, &HostStatusError{Status: response.StatusCode, Message: fmt.Sprintf("host refused applied-state observation (%s)", response.Status)} } - var applied AppliedManifest - if err := json.NewDecoder(io.LimitReader(response.Body, 1<<16)).Decode(&applied); err != nil { + var record struct { + AppliedManifest + Removed *bool `json:"removed"` + } + if err := json.NewDecoder(io.LimitReader(response.Body, 1<<16)).Decode(&record); err != nil { return nil, false, fmt.Errorf("read applied state: %w", err) } - return &applied, true, nil + if record.Binding == "" || record.Generation == 0 || record.Digest == "" || record.Release == "" || record.Removed == nil { + return nil, false, fmt.Errorf("read applied state: missing identity fields") + } + record.AppliedManifest.Removed = *record.Removed + return &record.AppliedManifest, true, nil } func (d *NativeDeliverer) withdrawalApplied(ctx context.Context, withdrawal *solutionhost.ModuleManifest, digest string) (bool, error) { @@ -152,6 +196,9 @@ func (d *NativeDeliverer) withdrawalApplied(ctx context.Context, withdrawal *sol if applied.Binding != withdrawal.Binding { return false, fmt.Errorf("applied record names binding %q, not %s", applied.Binding, withdrawal.Binding) } + if applied.Pending != nil && applied.Pending.Generation >= withdrawal.Generation { + return false, fmt.Errorf("the host withheld binding %s withdrawal generation %d: %s", withdrawal.Binding, applied.Pending.Generation, applied.Pending.Reason) + } if applied.Generation < withdrawal.Generation { return false, nil } diff --git a/pkg/gitops/native_http_applied_test.go b/pkg/gitops/native_http_applied_test.go index 944d9a5cf..b45a26982 100644 --- a/pkg/gitops/native_http_applied_test.go +++ b/pkg/gitops/native_http_applied_test.go @@ -6,7 +6,9 @@ import ( "errors" "net/http" "net/http/httptest" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -91,3 +93,75 @@ func TestWithdrawalCompletionPrefersTheDeclaredAppliedInterface(t *testing.T) { d.ObserverToken = nil require.NoError(t, d.WaitForWithdrawal(t.Context(), withdrawal)) } + +// The observer describes the host's current identity, not a monotonic success +// counter for this writer. A superseding release (including removal) is terminal. +func TestLegacyHTTPAppliedRequiresTheExactDeliveredIdentity(t *testing.T) { + module, services := nativeModule(t, "crm", false) + manifest := renderNative(t, module, services, "api") + manifest.Generation = 5 + digest, err := manifest.Digest() + require.NoError(t, err) + for _, tc := range []struct { + name string + change func(map[string]any) + malformed bool + previous bool + wantErr string + }{ + {name: "exact"}, + {name: "newer", change: func(v map[string]any) { v["generation"] = uint64(6) }, wantErr: "superseded"}, + {name: "newer removed", change: func(v map[string]any) { v["generation"] = uint64(6); v["removed"] = true; v["digest"] = "another" }, wantErr: "superseded"}, + {name: "wrong binding", change: func(v map[string]any) { v["binding"] = "other" }, wantErr: "binding"}, + {name: "wrong digest", change: func(v map[string]any) { v["digest"] = "another" }, wantErr: "does not match"}, + {name: "missing digest", change: func(v map[string]any) { delete(v, "digest") }, wantErr: "missing identity"}, + {name: "missing generation", change: func(v map[string]any) { delete(v, "generation") }, wantErr: "missing identity"}, + {name: "missing removed", change: func(v map[string]any) { delete(v, "removed") }, wantErr: "missing identity"}, + {name: "null removed", change: func(v map[string]any) { v["removed"] = nil }, wantErr: "missing identity"}, + {name: "wrong release", change: func(v map[string]any) { v["release"] = "another" }, wantErr: "does not match"}, + {name: "removed", change: func(v map[string]any) { v["removed"] = true }, wantErr: "does not match"}, + {name: "malformed", malformed: true, wantErr: "read applied state"}, + {name: "lower then exact", previous: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var calls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, http.MethodGet, r.Method) + require.Equal(t, "/platform/_delivery/applied/"+manifest.Binding, r.URL.Path) + require.Equal(t, "Bearer observer", r.Header.Get("Authorization")) + n := calls.Add(1) + if tc.malformed { + _, _ = w.Write([]byte(`{"generation":`)) + return + } + value := map[string]any{"binding": manifest.Binding, "generation": manifest.Generation, "digest": digest, "release": manifest.Release.Identity(), "removed": false} + if tc.change != nil { + tc.change(value) + } + if tc.previous && n == 1 { + value["generation"] = manifest.Generation - 1 + value["digest"] = "prior" + value["removed"] = true + } + _ = json.NewEncoder(w).Encode(value) + })) + defer server.Close() + d := &NativeDeliverer{DeliveryURL: server.URL, ObserverToken: func() (string, error) { return "observer", nil }} + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + err := d.WaitForApplied(ctx, manifest) + if tc.wantErr != "" { + require.ErrorContains(t, err, tc.wantErr) + var uncertain *AppliedUnconfirmed + require.False(t, errors.As(err, &uncertain), "identity disagreement is not observation uncertainty") + } else { + require.NoError(t, err) + } + wantCalls := int32(1) + if tc.previous { + wantCalls = 2 + } + require.Equal(t, wantCalls, calls.Load()) + }) + } +} diff --git a/pkg/gitops/wire_kits_test.go b/pkg/gitops/wire_kits_test.go index 07de2fe36..da6c1d908 100644 --- a/pkg/gitops/wire_kits_test.go +++ b/pkg/gitops/wire_kits_test.go @@ -46,7 +46,7 @@ func TestTheModuleContractKitRunsThroughTheRender(t *testing.T) { config, err := os.ReadFile(configuration) require.NoError(t, err) require.Contains(t, string(config), "modules:\n - name: shop\n") - require.NoError(t, os.WriteFile(configuration, []byte(strings.Replace(string(config), "modules:\n - name: shop\n", "modules:\n - name: shop\n - name: assistant\n - name: documents\n - name: model-gateway\n", 1)), 0o644)) + require.NoError(t, os.WriteFile(configuration, []byte(strings.Replace(string(config), "modules:\n - name: shop\n", "modules:\n - name: shop\n - name: assistant\n - name: documents\n - name: model-gateway\n - name: records\n", 1)), 0o644)) // The callees a binding that names its calls is resolved against // (binding-names-calls): each declares what its calls require in its own // published contract, a protobuf one and an OpenAPI one. @@ -58,7 +58,7 @@ func TestTheModuleContractKitRunsThroughTheRender(t *testing.T) { filepath.Join("modules", "assistant", resources.ModuleConfigurationName): "kind: module\nname: assistant\nservices:\n - name: api\n", filepath.Join("modules", "assistant", "services", "api", resources.ServiceConfigurationName): hostedServiceYAML("api", "assistant") + "module-identity: true\n", filepath.Join("modules", "assistant", "module.package.codefly.yaml"): "schema: codefly/module-package/v1\nid: acme/assistant\nversion: 1.0.0\n", - filepath.Join("configurations", "staging", "assistant.env"): "MODEL_AUDIENCE=model-gateway\nMODEL_RESOURCE_KIND=modelservice.profiles\nMODEL_BINDING=model\nEVIDENCE_AUDIENCE=documents\nEVIDENCE_RESOURCE_KIND=documents.passages\nEVIDENCE_BINDING=evidence\nANNOTATIONS_PREFIX=annotations\nDOCUMENTS_AUDIENCE=documents\nREPLICA_AUDIENCE=documents\nCOLLECTIONS_AUDIENCE=documents\n", + filepath.Join("configurations", "staging", "assistant.env"): "MODEL_AUDIENCE=model-gateway\nMODEL_RESOURCE_KIND=modelservice.profiles\nMODEL_BINDING=model\nEVIDENCE_AUDIENCE=documents\nEVIDENCE_RESOURCE_KIND=documents.passages\nEVIDENCE_BINDING=evidence\nANNOTATIONS_PREFIX=annotations\nDOCUMENTS_AUDIENCE=documents\nREPLICA_AUDIENCE=documents\nCOLLECTIONS_AUDIENCE=documents\nRECORDS_AUDIENCE=records\n", } { full := filepath.Join(workspace.Dir(), rel) require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) @@ -174,10 +174,10 @@ func TestTheModuleContractKitAdapterFailsOnUnrelatedErrors(t *testing.T) { require.False(t, past) } -// writeCalleePackages composes the two modules the kit's calls fixture calls: +// writeCalleePackages composes the modules the kit's calls fixtures call: // documents, whose protobuf contract marks ResolveEvidence and // ReadDocumentExcerpt, and model-gateway, whose OpenAPI contract marks its two -// routes. +// routes; and records, whose setup call declares records.types:define. func writeCalleePackages(t *testing.T, root string) { t.Helper() read := &hopv1.Policy{Requires: []*hopv1.Requirement{{ResourceKind: "documents", Actions: []string{"read"}}, {ResourceKind: "documents.collection", Actions: []string{"read"}}}} @@ -197,6 +197,16 @@ func writeCalleePackages(t *testing.T, root string) { "/v1/model/profiles":{"get":{"x-codefly-authority-policy":{"requires":[{"resource_kind":"modelservice.profiles","actions":["read"]}]}}}, "/v1/model/profiles/{name}:generate":{"post":{"x-codefly-authority-policy":{"requires":[{"resource_kind":"modelservice.profiles","actions":["invoke"]}]}}}}}`) writeCallee(t, filepath.Join(root, "modules", "model-gateway"), "model-gateway", "rest", "openapi", "ModelService", "openapi.json", openapi) + // Matches the Records owner's published DefineRecordType permission policy. + define := &descriptorpb.MethodDescriptorProto{Name: proto.String("DefineRecordType"), InputType: proto.String(".records.v1.Empty"), OutputType: proto.String(".records.v1.Empty"), Options: &descriptorpb.MethodOptions{}} + proto.SetExtension(define.Options, hopv1.E_Policy, &hopv1.Policy{Requires: []*hopv1.Requirement{{ResourceKind: "records.types", Actions: []string{"define"}}}}) + set, err = proto.Marshal(&descriptorpb.FileDescriptorSet{File: []*descriptorpb.FileDescriptorProto{{ + Name: proto.String("records/v1/api.proto"), Package: proto.String("records.v1"), Syntax: proto.String("proto3"), + MessageType: []*descriptorpb.DescriptorProto{{Name: proto.String("Empty")}}, + Service: []*descriptorpb.ServiceDescriptorProto{{Name: proto.String("RecordsService"), Method: []*descriptorpb.MethodDescriptorProto{define}}}, + }}}) + require.NoError(t, err) + writeCallee(t, filepath.Join(root, "modules", "records"), "records", "grpc", "protobuf", "records.v1", "contract.binpb", set) } // writeCallee writes one composed module publishing one contract. diff --git a/pkg/solutionrun/frontend_local.go b/pkg/solutionrun/frontend_local.go index 212d11a08..424ec476b 100644 --- a/pkg/solutionrun/frontend_local.go +++ b/pkg/solutionrun/frontend_local.go @@ -56,8 +56,8 @@ func (b LocalFrontend) bound() int64 { type localFrontendRecord struct { // Inputs is the digest of the inputs the bundle was built from. Inputs string `json:"inputs"` - // Installed is the lockfile digest the staged copy's dependencies were - // last installed from (`npm ci`), for a native build. + // Installed is the install-input digest used to build this native bundle. + // The mutable staged installation has its own marker beside source/. Installed string `json:"installed,omitempty"` // Bundle is the retained bundle's digest. Bundle string `json:"bundle"` @@ -179,25 +179,25 @@ func (b LocalFrontend) prune(ctx context.Context, keep string) error { // Bundle answers the digest of the module's frontend bundle as its sources are // now, building it only when no bundle is retained for these sources. func (b LocalFrontend) Bundle(ctx context.Context, module *resources.Module) (string, error) { - sources, err := selectFrontendSources(ctx, module) + guard, err := b.lock(ctx, module.Name) if err != nil { return "", err } - inputs, lock, err := sources.digest(module, b.Docker) + defer func() { _ = guard.Unlock() }() + sources, err := selectFrontendSources(ctx, module) if err != nil { return "", err } - recordPath := filepath.Join(b.StateDir, module.Name+".json") - guard, err := b.lock(ctx, module.Name) + inputs, lock, err := sources.digest(module, b.Docker) if err != nil { return "", err } - defer func() { _ = guard.Unlock() }() + recordPath := filepath.Join(b.StateDir, module.Name+".json") var record localFrontendRecord if data, readErr := os.ReadFile(recordPath); readErr == nil { _ = json.Unmarshal(data, &record) } - if record.Inputs == inputs && record.Bundle != "" && b.Retained(record.Bundle) { + if record.Inputs == inputs && (b.Docker || record.Installed == lock) && record.Bundle != "" && b.Retained(record.Bundle) { return record.Bundle, nil } ctx, cancel := context.WithTimeout(ctx, localFrontendBuildBound) @@ -212,18 +212,25 @@ func (b LocalFrontend) Bundle(ctx context.Context, module *resources.Module) (st root = filepath.Join(scratch, "bundle") stage := func(ctx context.Context, _ *resources.Module, scratch string) (string, string, error) { destination := filepath.Join(scratch, "source") - return destination, sources.component, sources.sync(ctx, destination) + return destination, sources.component, sources.stage(ctx, destination, module, b.Docker, inputs) } if err = buildFrontendBundle(ctx, module, root, stage); err != nil { return "", err } } else { staging := filepath.Join(b.stagingDir(module.Name), "source") - if root, err = sources.buildNatively(ctx, module, staging, lock != record.Installed); err != nil { + if root, err = sources.buildNatively(ctx, module, staging, lock, inputs); err != nil { return "", err } record.Installed = lock } + current, err := b.InputDigest(ctx, module) + if err != nil { + return "", err + } + if current != inputs { + return "", ErrFrontendInputsChanged + } if record.Bundle, err = b.Retain(ctx, root); err != nil { return "", fmt.Errorf("retain the frontend bundle of %s: %w", module.Name, err) } @@ -262,9 +269,9 @@ func (b LocalFrontend) Bundle(ctx context.Context, module *resources.Module) (st type frontendSources struct { root, component, project string files []string - // tarballs are the files local dependencies name: with the lockfile, what - // an install reads. - tarballs map[string]bool + // installFiles are local dependency inputs, including directory contents: + // npm may copy these packages into node_modules rather than link them. + installFiles map[string]bool } func selectFrontendSources(ctx context.Context, module *resources.Module) (*frontendSources, error) { @@ -315,7 +322,11 @@ func selectFrontendSources(ctx context.Context, module *resources.Module) (*fron return nil, fmt.Errorf("select the frontend sources of %s: %w", module.Name, err) } for name, tracked := range names { - add(rootRelative(root, filepath.Join(dir, name)), tracked) + relative := rootRelative(root, filepath.Join(dir, name)) + add(relative, tracked) + if dir != projectDir { + named[relative] = true + } } } for dir := projectDir; dir != root; { @@ -336,7 +347,7 @@ func selectFrontendSources(ctx context.Context, module *resources.Module) (*fron } } } - sources := &frontendSources{root: root, component: filepath.ToSlash(component), project: rootRelative(root, projectDir), tarballs: named} + sources := &frontendSources{root: root, component: filepath.ToSlash(component), project: rootRelative(root, projectDir), installFiles: named} for relative := range selected { sources.files = append(sources.files, relative) } @@ -458,9 +469,10 @@ func listFiles(ctx context.Context, dir string, repository, recursive bool) (map } // digest is the sources by path, mode and content, with how they are built, -// and apart, the digest of what an install reads: the project's lockfile and -// the tarballs its local dependencies name, so replacing a vendored tarball -// in place installs it again. A symlink or special file among +// and apart, the digest of what an install reads: the project's manifest and +// lockfile, npm configuration, and local dependency contents (directories or +// tarballs), so changing an installed input in place installs it again. +// A symlink or special file among // them is refused by name: the deployed build refuses one too. func (s *frontendSources) digest(module *resources.Module, docker bool) (string, string, error) { sum := sha256.New() @@ -487,8 +499,8 @@ func (s *frontendSources) digest(module *resources.Module, docker bool) (string, present++ hash := sha256.Sum256(content) _, _ = fmt.Fprintf(sum, "%s\x00%o\x00%x\x00", relative, info.Mode().Perm(), hash) - if s.tarballs[relative] || relative == path.Join(s.project, "package-lock.json") || relative == path.Join(s.project, "npm-shrinkwrap.json") { - _, _ = fmt.Fprintf(install, "%s\x00%x\x00", relative, hash) + if s.installFiles[relative] || path.Base(relative) == ".npmrc" || relative == path.Join(s.project, "package.json") || relative == path.Join(s.project, "package-lock.json") || relative == path.Join(s.project, "npm-shrinkwrap.json") { + _, _ = fmt.Fprintf(install, "%s\x00%o\x00%x\x00", relative, info.Mode().Perm(), hash) installs++ } } @@ -594,23 +606,39 @@ func installCommand(projectDir string) []string { // buildNatively runs the frontend's own npm build in staging, a copy of // exactly its sources — installing first (see installCommand) when the copy's -// dependencies were not installed from this lockfile — and answers the bundle +// dependencies were not installed from these inputs — and answers the bundle // it wrote. The // component's own node_modules and files are never touched. A failure names the // command and carries the end of its output. -func (s *frontendSources) buildNatively(ctx context.Context, module *resources.Module, staging string, install bool) (string, error) { - if err := s.sync(ctx, staging); err != nil { +func (s *frontendSources) buildNatively(ctx context.Context, module *resources.Module, staging, installInputs, inputs string) (string, error) { + if err := s.stage(ctx, staging, module, false, inputs); err != nil { return "", fmt.Errorf("stage the frontend sources of %s: %w", module.Name, err) } projectDir := filepath.Join(staging, filepath.FromSlash(s.project)) + // Bundle retention and installation have different failure lifetimes. A + // failed build may have installed new dependencies; the old bundle's record + // says nothing about that mutable copy. Keep its marker outside source/, + // where syncing the authored files does not remove it. + installedPath := filepath.Join(filepath.Dir(staging), "installed") + installed, err := os.ReadFile(installedPath) + install := err != nil || string(installed) != installInputs if _, err := os.Stat(filepath.Join(projectDir, "node_modules")); err != nil { install = true } steps := [][]string{{"npm", "run", "build"}} if install { + // Invalidate before mutating anything: an interrupted/failed install + // must not leave a trusted marker. npm install (the no-lock path) can + // retain old file: copies, so install into an empty dependency tree. + if err := os.Remove(installedPath); err != nil && !errors.Is(err, fs.ErrNotExist) { + return "", err + } + if err := os.RemoveAll(filepath.Join(projectDir, "node_modules")); err != nil { + return "", err + } steps = append([][]string{installCommand(projectDir)}, steps...) } - for _, step := range steps { + for i, step := range steps { //nolint:gosec // G204: fixed npm executable and arguments, no shell. command := exec.CommandContext(ctx, step[0], step[1:]...) command.Dir = projectDir @@ -620,6 +648,11 @@ func (s *frontendSources) buildNatively(ctx context.Context, module *resources.M if err := command.Run(); err != nil { return "", fmt.Errorf("build the frontend of %s (%s in %s): %w\n%s", module.Name, strings.Join(step, " "), s.project, err, tail(output.String(), 40)) } + if install && i == 0 { + if err := os.WriteFile(installedPath, []byte(installInputs), 0o600); err != nil { + return "", err + } + } } services, err := filepath.Rel(module.Dir(), module.ServicesDir()) if err != nil { diff --git a/pkg/solutionrun/frontend_local_test.go b/pkg/solutionrun/frontend_local_test.go index 02e8635dc..09dfabf41 100644 --- a/pkg/solutionrun/frontend_local_test.go +++ b/pkg/solutionrun/frontend_local_test.go @@ -2,6 +2,8 @@ package solutionrun import ( "context" + "encoding/json" + "fmt" "os" "os/exec" "path/filepath" @@ -160,7 +162,7 @@ esac require.Equal(t, "build", invocations(), "an edit to an untracked file rebuilds") require.NoError(t, os.WriteFile(filepath.Join(root, "packages/shared/index.ts"), []byte("shared v2"), 0o600)) require.Contains(t, bundle(), `"shared":"shared v2"`) - require.Equal(t, "build", invocations(), "a sibling package edit rebuilds") + require.Equal(t, "ci build", invocations(), "a sibling package edit reinstalls and rebuilds") require.NoError(t, os.Remove(filepath.Join(remote, "NewWidget.tsx"))) require.Contains(t, bundle(), `"widget":""`, "a deleted source leaves the staged copy too") require.Equal(t, "build", invocations()) @@ -272,6 +274,257 @@ func packTarball(t *testing.T, vendorDir, version, content string) string { func strconvQuote(s string) string { return `"` + strings.ReplaceAll(s, `"`, `\"`) + `"` } +// npm can copy file: dependencies rather than link them. A staged source edit +// must reach the installed copy even when its package version and lock stay put. +func directoryDependencyFrontend(t *testing.T, locked bool) *resources.Module { + t.Helper() + if _, err := exec.LookPath("npm"); err != nil { + t.Skip("npm is not installed: local directory installs are not qualified here") + } + module := frontendModule(t, map[string]string{ + "services/schemas/package.json": `{"name":"schemas","version":"1.0.0","files":["first.json","second.json"]}`, + "services/schemas/first.json": `"first"`, + "services/fe-remote/.npmrc": "install-links=true\noffline=true\naudit=false\nfund=false\n", + "services/fe-remote/package.json": `{"name":"remote","version":"1.0.0","scripts":{"build":"node build.cjs"},"dependencies":{"schemas":"file:../schemas"}}`, + "services/fe-remote/build.cjs": `const fs=require('fs');fs.mkdirSync('dist',{recursive:true});fs.writeFileSync('dist/mf-manifest.json',JSON.stringify(fs.readdirSync('node_modules/schemas').filter(n=>n.endsWith('.json')&&n!=='package.json').sort().map(n=>JSON.parse(fs.readFileSync('node_modules/schemas/'+n)))));`, + }) + remote := filepath.Join(module.ServicesDir(), "fe-remote") + if !locked { + require.NoError(t, os.WriteFile(filepath.Join(remote, ".npmrc"), []byte("install-links=true\npackage-lock=false\noffline=true\naudit=false\nfund=false\n"), 0o600)) + return module + } + command := exec.Command("npm", "install", "--package-lock-only", "--cache", t.TempDir()) + command.Dir = remote + output, err := command.CombinedOutput() + require.NoError(t, err, string(output)) + return module +} + +func TestLocalFrontendReinstallsDirectoryDependencies(t *testing.T) { + for _, locked := range []bool{true, false} { + t.Run(fmt.Sprintf("locked=%t", locked), func(t *testing.T) { + testLocalFrontendReinstallsDirectoryDependencies(t, locked) + }) + } +} + +func testLocalFrontendReinstallsDirectoryDependencies(t *testing.T, locked bool) { + t.Helper() + module := directoryDependencyFrontend(t, locked) + remote := filepath.Join(module.ServicesDir(), "fe-remote") + lock, _ := os.ReadFile(filepath.Join(remote, "package-lock.json")) + t.Setenv("npm_config_cache", t.TempDir()) + var bundles []string + build := LocalFrontend{ + StateDir: t.TempDir(), + Retain: func(_ context.Context, root string) (string, error) { + data, err := os.ReadFile(filepath.Join(root, FrontendBundleManifest)) + bundles = append(bundles, string(data)) + return "sha256:" + string(rune('a'+len(bundles))), err + }, + Retained: func(string) bool { return true }, + } + bundle := func(want string) { + t.Helper() + _, err := build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, want, bundles[len(bundles)-1]) + } + bundle(`["first"]`) + require.NoError(t, os.WriteFile(filepath.Join(module.ServicesDir(), "schemas/second.json"), []byte(`"second"`), 0o600)) + bundle(`["first","second"]`) + require.NoError(t, os.WriteFile(filepath.Join(module.ServicesDir(), "schemas/first.json"), []byte(`"changed"`), 0o600)) + bundle(`["changed","second"]`) + require.NoError(t, os.Remove(filepath.Join(module.ServicesDir(), "schemas/second.json"))) + bundle(`["changed"]`) + require.NoError(t, os.WriteFile(filepath.Join(module.ServicesDir(), "schemas/package.json"), []byte(`{"name":"schemas","version":"1.0.0","files":["second.json"]}`), 0o600)) + bundle(`[]`) + // A record made by an older install-identity algorithm cannot keep an + // otherwise matching bundle: its node_modules may contain a stale copy. + recordPath := filepath.Join(build.StateDir, module.Name+".json") + data, err := os.ReadFile(recordPath) + require.NoError(t, err) + var record localFrontendRecord + require.NoError(t, json.Unmarshal(data, &record)) + record.Installed = "obsolete-install-identity" + data, err = json.Marshal(record) + require.NoError(t, err) + require.NoError(t, os.WriteFile(recordPath, data, 0o600)) + count := len(bundles) + bundle(`[]`) + require.Len(t, bundles, count+1) + unchangedLock, err := os.ReadFile(filepath.Join(remote, "package-lock.json")) + if locked { + require.NoError(t, err) + require.Equal(t, lock, unchangedLock) + } else { + require.ErrorIs(t, err, os.ErrNotExist) + } + require.NoDirExists(t, filepath.Join(remote, "node_modules"), "the owner's tree is never installed into") +} + +// A successful bundle records A, but a failed B build may already have installed +// B. Reverting to A and editing only source must not build against that B copy. +func TestLocalFrontendRecoversInstalledDependenciesAfterFailure(t *testing.T) { + for _, failure := range []string{"install", "build", "retain", "changed inputs"} { + t.Run(failure, func(t *testing.T) { + module := directoryDependencyFrontend(t, true) + t.Setenv("npm_config_cache", t.TempDir()) + t.Setenv("npm_config_ignore_scripts", "false") + remote := filepath.Join(module.ServicesDir(), "fe-remote") + dependency := filepath.Join(module.ServicesDir(), "schemas/first.json") + buildPath := filepath.Join(remote, "build.cjs") + buildScript, err := os.ReadFile(buildPath) + require.NoError(t, err) + // npm runs the project's preinstall after copying its dependencies. + manifest, err := os.ReadFile(filepath.Join(remote, "package.json")) + require.NoError(t, err) + manifest = []byte(strings.Replace(string(manifest), `"build":"node build.cjs"`, `"build":"node build.cjs","preinstall":"node install.cjs"`, 1)) + require.NoError(t, os.WriteFile(filepath.Join(remote, "package.json"), manifest, 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(remote, "install.cjs"), []byte(`if(require('fs').existsSync('fail-install'))process.exit(1);`), 0o600)) + failRetain := false + var bundles []string + build := LocalFrontend{ + StateDir: t.TempDir(), Retained: func(string) bool { return true }, + Retain: func(_ context.Context, root string) (string, error) { + if failRetain { + return "", fmt.Errorf("retention failed") + } + data, err := os.ReadFile(filepath.Join(root, FrontendBundleManifest)) + bundles = append(bundles, string(data)) + return "sha256:" + string(rune('a'+len(bundles))), err + }, + } + original, err := build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, `["first"]`, bundles[0]) + require.NoError(t, os.WriteFile(dependency, []byte(`"second"`), 0o600)) + switch failure { + case "install": + require.NoError(t, os.WriteFile(filepath.Join(remote, "fail-install"), nil, 0o600)) + case "build": + require.NoError(t, os.WriteFile(buildPath, []byte("process.exit(1);"), 0o600)) + case "retain": + failRetain = true + case "changed inputs": + target, err := json.Marshal(dependency) + require.NoError(t, err) + script := append(slices.Clone(buildScript), []byte("\nfs.writeFileSync("+string(target)+",'\"third\"');")...) + require.NoError(t, os.WriteFile(buildPath, script, 0o600)) + } + _, err = build.Bundle(t.Context(), module) + require.Error(t, err) + if failure == "changed inputs" { + require.ErrorIs(t, err, ErrFrontendInputsChanged) + } + installed, err := os.ReadFile(filepath.Join(build.stagingDir(module.Name), "source/services/fe-remote/node_modules/schemas/first.json")) + require.NoError(t, err) + require.Equal(t, `"second"`, string(installed), "the failed attempt already changed the installed copy") + failRetain = false + if failure == "install" { + require.NoError(t, os.Remove(filepath.Join(remote, "fail-install"))) + } + require.NoError(t, os.WriteFile(dependency, []byte(`"first"`), 0o600)) + require.NoError(t, os.WriteFile(buildPath, buildScript, 0o600)) + // The old immutable bundle is still reusable even when the mutable + // staged installation no longer matches it. + digest, err := build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, original, digest) + require.Len(t, bundles, 1) + require.NoError(t, os.WriteFile(buildPath, append(slices.Clone(buildScript), []byte("\n// source-only edit\n")...), 0o600)) + _, err = build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, []string{`["first"]`, `["first"]`}, bundles) + }) + } +} + +func TestLocalFrontendReinstallsAfterCanceledInstall(t *testing.T) { + root, module := wikiRepository(t) + remote := filepath.Join(module.ServicesDir(), "fe-remote") + started := filepath.Join(t.TempDir(), "started") + bin := t.TempDir() + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + // exec replaces the shell: canceling the command kills the sole sleeping + // process, after it has left a partial installation behind. + require.NoError(t, os.WriteFile(filepath.Join(bin, "npm"), []byte(`#!/bin/sh +set -eu +case "$*" in +ci) mkdir -p node_modules + cp ../../../../packages/shared/index.ts node_modules/value + if [ -f interrupt-install ]; then + touch '`+started+`' + exec sleep 60 + fi ;; +"run build") mkdir -p dist; cp node_modules/value dist/mf-manifest.json ;; +esac +`), 0o700)) + var bundle string + build := LocalFrontend{ + StateDir: t.TempDir(), Retained: func(string) bool { return true }, + Retain: func(_ context.Context, root string) (string, error) { + data, err := os.ReadFile(filepath.Join(root, FrontendBundleManifest)) + bundle = string(data) + return "sha256:" + bundle, err + }, + } + _, err := build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, "shared v1", bundle) + require.NoError(t, os.WriteFile(filepath.Join(root, "packages/shared/index.ts"), []byte("shared v2"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(remote, "interrupt-install"), nil, 0o600)) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := build.Bundle(ctx, module); done <- err }() + require.Eventually(t, func() bool { _, err := os.Stat(started); return err == nil }, 5*time.Second, 10*time.Millisecond) + cancel() + select { + case err := <-done: + require.Error(t, err) + case <-time.After(5 * time.Second): + t.Fatal("canceled install did not exit") + } + require.NoError(t, os.Remove(filepath.Join(remote, "interrupt-install"))) + require.NoError(t, os.WriteFile(filepath.Join(root, "packages/shared/index.ts"), []byte("shared v1"), 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(remote, "NewWidget.tsx"), []byte("a new source"), 0o600)) + _, err = build.Bundle(t.Context(), module) + require.NoError(t, err) + require.Equal(t, "shared v1", bundle) +} + +func TestFrontendInstallIdentityIncludesTransitivePackagesAndConfiguration(t *testing.T) { + root, module := wikiRepository(t) + write := func(name, content string) { + t.Helper() + full := filepath.Join(root, name) + require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) + require.NoError(t, os.WriteFile(full, []byte(content), 0o600)) + } + write("packages/shared/package.json", `{"name":"shared","dependencies":{"nested":"file:../nested"}}`) + write("packages/nested/package.json", `{"name":"nested"}`) + write("packages/nested/index.js", "first") + write("solutions/wiki/services/fe-remote/.npmrc", "install-links=true") + digest := func() string { + t.Helper() + sources, err := selectFrontendSources(t.Context(), module) + require.NoError(t, err) + _, install, err := sources.digest(module, false) + require.NoError(t, err) + return install + } + for _, change := range []struct{ name, content string }{ + {"packages/nested/index.js", "second"}, + {"solutions/wiki/services/fe-remote/.npmrc", "install-links=false"}, + {"solutions/wiki/services/fe-remote/package.json", `{"dependencies":{"shared":"file:../../../../packages/shared"},"optionalDependencies":{}}`}, + } { + before := digest() + write(change.name, change.content) + require.NotEqual(t, before, digest(), change.name) + } +} + // A project depending on a vendored tarball beside another component — as // every solution depending on the wiki's vendored UI does — reads it: the // tarball is a source at the path npm resolves, so an empty npm cache installs diff --git a/pkg/solutionrun/frontend_watch.go b/pkg/solutionrun/frontend_watch.go new file mode 100644 index 000000000..654439bc6 --- /dev/null +++ b/pkg/solutionrun/frontend_watch.go @@ -0,0 +1,83 @@ +package solutionrun + +import ( + "context" + "errors" + "fmt" + "os/exec" + "path/filepath" + "strings" + + "github.com/codefly-dev/core/agents/helpers/code" + "github.com/codefly-dev/core/resources" +) + +// ErrFrontendInputsChanged rejects a bundle built while its selected inputs +// changed. The caller may reconcile again; this is not a successful delivery. +var ErrFrontendInputsChanged = errors.New("frontend inputs changed during staging or build; reconcile the current source") + +// InputDigest uses the build's canonical selector, including local packages and +// named ignored tarballs. Watch events alone do not establish a changed build. +func (b LocalFrontend) InputDigest(ctx context.Context, module *resources.Module) (string, error) { + sources, err := selectFrontendSources(ctx, module) + if err != nil { + return "", err + } + inputs, _, err := sources.digest(module, b.Docker) + return inputs, err +} + +func (s *frontendSources) stage(ctx context.Context, destination string, module *resources.Module, docker bool, expected string) error { + if err := s.sync(ctx, destination); err != nil { + return err + } + staged := *s + staged.root = destination + actual, _, err := staged.digest(module, docker) + if err != nil { + return err + } + if actual != expected { + return ErrFrontendInputsChanged + } + return nil +} + +// FrontendWatchInterest survives an invalid package graph. Its bounded tree is +// the same repository/component boundary localPackages enforces. Only outputs +// are pruned: tracked ignored files and named ignored tarballs are valid inputs. +// Git metadata uses exact files, including worktree-specific index locations. +func FrontendWatchInterest(ctx context.Context, module *resources.Module) (code.WatchTree, []string, error) { + root := frontendRepository(module.Dir()) + repository := root != "" + if !repository { + root = module.Dir() + } + root, err := filepath.EvalSymlinks(root) + if err != nil { + return code.WatchTree{}, nil, err + } + component, err := filepath.EvalSymlinks(module.Dir()) + if err != nil { + return code.WatchTree{}, nil, err + } + services, err := filepath.Rel(module.Dir(), module.ServicesDir()) + if err != nil { + return code.WatchTree{}, nil, err + } + bundle := rootRelative(root, filepath.Join(component, services, filepath.FromSlash(FrontendBundleDir))) + tree := code.WatchTree{Root: root, Exclude: func(p string) bool { + return frontendBuildOutput(rootRelative(root, p), bundle) + }} + var files []string + if repository { + for _, name := range []string{"index", "info/exclude"} { + output, err := exec.CommandContext(ctx, "git", "-C", root, "rev-parse", "--path-format=absolute", "--git-path", name).Output() + if err != nil { + return tree, nil, fmt.Errorf("resolve frontend watch Git metadata %s: %w", name, err) + } + files = append(files, strings.TrimSpace(string(output))) + } + } + return tree, files, nil +} diff --git a/pkg/solutionrun/frontend_watch_test.go b/pkg/solutionrun/frontend_watch_test.go new file mode 100644 index 000000000..2ed289673 --- /dev/null +++ b/pkg/solutionrun/frontend_watch_test.go @@ -0,0 +1,141 @@ +package solutionrun + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/codefly-dev/core/agents/helpers/code" + "github.com/codefly-dev/core/builders" + "github.com/stretchr/testify/require" +) + +func TestFrontendStagingRejectsChangedSnapshot(t *testing.T) { + _, module := wikiRepository(t) + sources, err := selectFrontendSources(t.Context(), module) + require.NoError(t, err) + inputs, _, err := sources.digest(module, false) + require.NoError(t, err) + staging := t.TempDir() + require.NoError(t, sources.stage(t.Context(), staging, module, false, inputs)) + file := filepath.Join(module.ServicesDir(), "fe-remote", "Page.tsx") + require.NoError(t, os.WriteFile(file, []byte("changed after discovery"), 0600)) + require.ErrorIs(t, sources.stage(t.Context(), staging, module, false, inputs), ErrFrontendInputsChanged) + require.NoError(t, os.Remove(file)) + require.ErrorIs(t, sources.stage(t.Context(), staging, module, false, inputs), ErrFrontendInputsChanged) +} + +func TestFrontendWatchRetainsIgnoredInputsAndBrokenGraphRepair(t *testing.T) { + root, module := wikiRepository(t) + remote := filepath.Join(module.ServicesDir(), "fe-remote") + tracked := filepath.Join(remote, "src", "nested", "Widget.tsx") + vendor := filepath.Join(root, "vendor", "shared.tgz") + require.NoError(t, os.MkdirAll(filepath.Dir(vendor), 0755)) + require.NoError(t, os.WriteFile(vendor, []byte("tarball v1"), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(root, ".gitignore"), []byte("vendor/\nsolutions/wiki/services/fe-remote/src/\n"), 0600)) + manifest := filepath.Join(remote, "package.json") + require.NoError(t, os.WriteFile(manifest, []byte(`{"dependencies":{"shared":"file:../../../../vendor/shared.tgz"}}`), 0600)) + tree, files, err := FrontendWatchInterest(t.Context(), module) + require.NoError(t, err) + require.Len(t, files, 2) + for _, dir := range []string{filepath.Dir(vendor), filepath.Dir(tracked)} { + physical, err := filepath.EvalSymlinks(dir) + require.NoError(t, err) + require.False(t, tree.Exclude(physical), "ignored inputs are not unconditional output") + } + require.True(t, tree.Exclude(filepath.Join(tree.Root, "solutions/wiki/services/fe-remote/dist"))) + dependencies := builders.NewDependencies("frontend") + for _, file := range files { + dependencies.AddDependencies(builders.NewDependency(file)) + } + events := make(chan code.Change, 100) + watcher, err := code.NewWatcher(t.Context(), events, tree.Root, dependencies, tree) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go watcher.Start(ctx) + build := LocalFrontend{} + for _, file := range []string{tracked, vendor} { + before, err := build.InputDigest(ctx, module) + require.NoError(t, err) + require.NoError(t, os.WriteFile(file, []byte("changed"), 0600)) + relative, err := filepath.Rel(root, file) + require.NoError(t, err) + awaitFrontendEvent(t, events, relative) + after, err := build.InputDigest(ctx, module) + require.NoError(t, err) + require.NotEqual(t, before, after) + } + // Invalid metadata does not change or discard the previously registered interest. + require.NoError(t, os.WriteFile(manifest, []byte(`{"dependencies":{"missing":"file:../missing"}}`), 0600)) + _, err = build.InputDigest(ctx, module) + require.Error(t, err) + require.NoError(t, os.WriteFile(manifest, []byte(`{}`), 0600)) + awaitFrontendEvent(t, events, "solutions/wiki/services/fe-remote/package.json") + _, err = build.InputDigest(ctx, module) + require.NoError(t, err) + // Index/exclude live outside the traversed tree in a Git worktree too. + require.NoError(t, os.WriteFile(files[1], []byte("other-ignored\n"), 0600)) + awaitFrontendEvent(t, events, filepath.Join(".git", "info", "exclude")) +} + +func awaitFrontendEvent(t *testing.T, events <-chan code.Change, path string) { + t.Helper() + deadline := time.After(3 * time.Second) + for { + select { + case change := <-events: + require.NoError(t, change.Err) + if change.Path == path { + return + } + case <-deadline: + t.Fatalf("selected frontend input event not observed: %s", path) + } + } +} + +func TestLocalFrontendRejectsEditDuringRealBuildAndRepairs(t *testing.T) { + if _, err := exec.LookPath("npm"); err != nil { + t.Skip("npm unavailable; real build not qualified") + } + t.Setenv("npm_config_cache", t.TempDir()) + t.Setenv("npm_config_offline", "true") + t.Setenv("npm_config_audit", "false") + t.Setenv("npm_config_fund", "false") + barriers := t.TempDir() + ready, _ := json.Marshal(filepath.Join(barriers, "ready")) + release, _ := json.Marshal(filepath.Join(barriers, "release")) + module := frontendModule(t, map[string]string{ + "package.json": `{"private":true,"scripts":{"build":"node build.cjs"}}`, + "Page.tsx": "before", + "build.cjs": `const fs=require('fs'); fs.writeFileSync(` + string(ready) + `, 'ready'); +const wait=()=>{if(!fs.existsSync(` + string(release) + `)){setTimeout(wait,10);return;} +fs.mkdirSync('services/fe-remote/dist',{recursive:true}); +fs.writeFileSync('services/fe-remote/dist/mf-manifest.json',fs.readFileSync('Page.tsx'));};wait();`, + }) + var retained atomic.Int32 + build := LocalFrontend{StateDir: t.TempDir(), Retained: func(string) bool { return true }, Retain: func(_ context.Context, root string) (string, error) { + retained.Add(1) + data, err := os.ReadFile(filepath.Join(root, FrontendBundleManifest)) + return string(data), err + }} + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { _, err := build.Bundle(ctx, module); done <- err }() + require.Eventually(t, func() bool { _, err := os.Stat(filepath.Join(barriers, "ready")); return err == nil }, 10*time.Second, 20*time.Millisecond) + require.NoError(t, os.WriteFile(filepath.Join(module.Dir(), "Page.tsx"), []byte("after"), 0600)) + require.NoError(t, os.WriteFile(filepath.Join(barriers, "release"), nil, 0600)) + require.ErrorIs(t, <-done, ErrFrontendInputsChanged) + require.Zero(t, retained.Load(), "stale build must not be retained as the current input snapshot") + bundle, err := build.Bundle(ctx, module) + require.NoError(t, err) + require.Equal(t, "after", bundle) + require.EqualValues(t, 1, retained.Load()) +}