Skip to content

Commit 9480240

Browse files
fix: make Destroy self-contained for execution cleanup (#20)
Destroy stopped the runner but left the file watcher running and never shut down the runner environment, and it reported SUCCESS after merely logging a failed process stop or cache wipe. Factor the teardown Stop and Destroy share into endExecution: stop the watcher first, then the process generation, then the owned environment, attempting every step even when an earlier one fails and aggregating the errors so the caller sees them. The environment is released only once it is actually down, so a retry can still reach it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent 43e5eae commit 9480240

2 files changed

Lines changed: 275 additions & 24 deletions

File tree

‎destroy_test.go‎

Lines changed: 210 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,210 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"errors"
6+
"os"
7+
"path/filepath"
8+
"testing"
9+
"time"
10+
11+
"github.com/codefly-dev/core/agents/services"
12+
runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0"
13+
"github.com/codefly-dev/core/resources"
14+
runners "github.com/codefly-dev/core/runners/base"
15+
"github.com/stretchr/testify/require"
16+
)
17+
18+
// destroyRuntime is a Runtime holding a real, running native process — the
19+
// shape `codefly` shutdown reaches when it calls Destroy with no preceding
20+
// Stop. The process touches a file in cacheLocation every 50ms, so a cache
21+
// wiped while it was still alive gets repopulated and the ordering is
22+
// observable rather than assumed.
23+
func destroyRuntime(t *testing.T) (*Runtime, runners.Proc, string) {
24+
t.Helper()
25+
ctx := context.Background()
26+
27+
runtime := NewRuntime(NewService())
28+
runtime.Location = t.TempDir()
29+
runtime.Logger = runtime.Wool
30+
runtime.Base.Runtime.RuntimeContext = resources.NewRuntimeContextNative()
31+
32+
cache := t.TempDir()
33+
require.NoError(t, os.WriteFile(filepath.Join(cache, "uv.json"), []byte("{}"), 0o600))
34+
runtime.cacheLocation = cache
35+
36+
env, err := runners.NewNativeEnvironment(ctx, t.TempDir())
37+
require.NoError(t, err)
38+
proc, err := env.NewProcess("sh", "-c",
39+
"while true; do touch "+filepath.Join(cache, "heartbeat")+"; sleep 0.05; done")
40+
require.NoError(t, err)
41+
require.NoError(t, proc.Start(ctx))
42+
t.Cleanup(func() { _ = proc.Stop(context.Background()) })
43+
44+
runtime.runner = &runnerHandle{proc: proc, cancel: func() {}}
45+
runtime.runnerEnvironment = env
46+
return runtime, proc, cache
47+
}
48+
49+
func TestDestroyEndsExecutionBeforeClearingCache(t *testing.T) {
50+
ctx := context.Background()
51+
runtime, proc, cache := destroyRuntime(t)
52+
53+
resp, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
54+
require.NoError(t, err)
55+
require.Equal(t, runtimev0.DestroyStatus_SUCCESS, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
56+
57+
alive, err := proc.IsRunning(ctx)
58+
require.NoError(t, err)
59+
require.False(t, alive, "Destroy must stop the service process without a preceding Stop")
60+
61+
require.Nil(t, currentRunner(runtime))
62+
require.Nil(t, runtime.runnerEnvironment, "the runner environment must be shut down and released")
63+
64+
entries, err := os.ReadDir(cache)
65+
require.NoError(t, err)
66+
require.Empty(t, entries, "the owned cache must be cleared")
67+
68+
// Several heartbeat periods: a process that outlived the wipe repopulates.
69+
time.Sleep(500 * time.Millisecond)
70+
entries, err = os.ReadDir(cache)
71+
require.NoError(t, err)
72+
require.Empty(t, entries, "the cache must be cleared after execution ended, not before")
73+
}
74+
75+
func TestDestroyStopsWatcher(t *testing.T) {
76+
ctx := context.Background()
77+
runtime, _, _ := destroyRuntime(t)
78+
require.NoError(t, os.MkdirAll(filepath.Join(runtime.Location, "code/src"), 0o700))
79+
80+
require.NoError(t, runtime.SetupWatcher(ctx, services.NewWatchConfiguration(requirements), runtime.EventHandler))
81+
events := runtime.Events
82+
83+
_, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
84+
require.NoError(t, err)
85+
86+
select {
87+
case _, open := <-events:
88+
require.False(t, open, "the watcher must be stopped, not still delivering events")
89+
case <-time.After(5 * time.Second):
90+
t.Fatal("watcher still running after Destroy")
91+
}
92+
}
93+
94+
func TestDestroyReportsCacheFailureAfterEndingExecution(t *testing.T) {
95+
if os.Geteuid() == 0 {
96+
t.Skip("root ignores directory permissions, so the wipe cannot be made to fail")
97+
}
98+
ctx := context.Background()
99+
runtime, proc, cache := destroyRuntime(t)
100+
101+
// Removing an entry needs write permission on its parent, so the wipe fails
102+
// while every other resource stays reachable.
103+
require.NoError(t, os.Chmod(cache, 0o500))
104+
t.Cleanup(func() { _ = os.Chmod(cache, 0o700) })
105+
106+
resp, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
107+
require.NoError(t, err)
108+
require.Equal(t, runtimev0.DestroyStatus_ERROR, resp.GetStatus().GetState(),
109+
"a failed cache wipe must be observable, not merely logged")
110+
require.Contains(t, resp.GetStatus().GetMessage(), "cannot remove cache")
111+
112+
alive, err := proc.IsRunning(ctx)
113+
require.NoError(t, err)
114+
require.False(t, alive, "execution cleanup must complete even when the cache wipe fails")
115+
require.Nil(t, currentRunner(runtime))
116+
require.Nil(t, runtime.runnerEnvironment)
117+
}
118+
119+
// stubbornProc fails every Stop, standing in for a process the agent cannot
120+
// reap. Only Stop is ever reached on this path.
121+
type stubbornProc struct{ runners.Proc }
122+
123+
func (stubbornProc) Stop(context.Context) error { return errors.New("process will not die") }
124+
125+
func TestDestroyReportsRunnerFailureAndStillClearsCache(t *testing.T) {
126+
ctx := context.Background()
127+
runtime, _, cache := destroyRuntime(t)
128+
runtime.runner = &runnerHandle{proc: stubbornProc{}, cancel: func() {}}
129+
130+
resp, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
131+
require.NoError(t, err)
132+
require.Equal(t, runtimev0.DestroyStatus_ERROR, resp.GetStatus().GetState(),
133+
"Destroy must not report success after only logging a failed process stop")
134+
require.Contains(t, resp.GetStatus().GetMessage(), "cannot stop the fastapi app")
135+
136+
require.Nil(t, runtime.runnerEnvironment, "a failed stop must not strand the remaining cleanup")
137+
entries, err := os.ReadDir(cache)
138+
require.NoError(t, err)
139+
require.Empty(t, entries)
140+
}
141+
142+
func TestDestroyIsIdempotent(t *testing.T) {
143+
ctx := context.Background()
144+
145+
// Before Init: no runner, no environment, no cache location claimed.
146+
fresh := NewRuntime(NewService())
147+
fresh.Location = t.TempDir()
148+
resp, err := fresh.Destroy(ctx, &runtimev0.DestroyRequest{})
149+
require.NoError(t, err)
150+
require.Equal(t, runtimev0.DestroyStatus_SUCCESS, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
151+
152+
runtime, _, _ := destroyRuntime(t)
153+
for range 3 {
154+
resp, err = runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
155+
require.NoError(t, err)
156+
require.Equal(t, runtimev0.DestroyStatus_SUCCESS, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
157+
}
158+
159+
// After Stop, Destroy still succeeds and has nothing left to release.
160+
stopped, _, _ := destroyRuntime(t)
161+
stop, err := stopped.Stop(ctx, &runtimev0.StopRequest{})
162+
require.NoError(t, err)
163+
require.Equal(t, runtimev0.StopStatus_SUCCESS, stop.GetStatus().GetState(), stop.GetStatus().GetMessage())
164+
resp, err = stopped.Destroy(ctx, &runtimev0.DestroyRequest{})
165+
require.NoError(t, err)
166+
require.Equal(t, runtimev0.DestroyStatus_SUCCESS, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
167+
}
168+
169+
// TestStartAfterDestroyCannotLeaveAReplacement covers the race `codefly`
170+
// shutdown actually runs: a Start already in flight lands after Destroy has
171+
// released the mutex. Destroy releases the execution environment, so that Start
172+
// has nothing to launch into and fails loudly instead of leaving a uvicorn
173+
// running against a wiped cache.
174+
func TestStartAfterDestroyCannotLeaveAReplacement(t *testing.T) {
175+
ctx := context.Background()
176+
runtime, _, _ := destroyRuntime(t)
177+
178+
_, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
179+
require.NoError(t, err)
180+
181+
start, err := runtime.Start(ctx, &runtimev0.StartRequest{})
182+
require.NoError(t, err)
183+
require.Equal(t, runtimev0.StartStatus_ERROR, start.GetStatus().GetState())
184+
require.Nil(t, currentRunner(runtime), "no process may survive Destroy")
185+
}
186+
187+
// TestDestroyStopsRealUvicornWithoutStop is the acceptance case against the
188+
// actual server: a listener that answers requests is gone before Destroy
189+
// reports success, with no Stop anywhere in the path.
190+
func TestDestroyStopsRealUvicornWithoutStop(t *testing.T) {
191+
ctx := context.Background()
192+
source := uvProject(t, servingApp)
193+
runtime := uvicornRuntime(t, source, false)
194+
runtime.cacheLocation = t.TempDir()
195+
196+
_, err := runtime.Start(ctx, &runtimev0.StartRequest{})
197+
require.NoError(t, err)
198+
require.Eventually(t, func() bool { return serves(runtime.port) }, 30*time.Second, 200*time.Millisecond,
199+
"the fixture must be serving before Destroy means anything")
200+
201+
handle := currentRunner(runtime)
202+
resp, err := runtime.Destroy(ctx, &runtimev0.DestroyRequest{})
203+
require.NoError(t, err)
204+
require.Equal(t, runtimev0.DestroyStatus_SUCCESS, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
205+
206+
running, err := handle.proc.IsRunning(ctx)
207+
require.NoError(t, err)
208+
require.False(t, running, "uvicorn must be gone before Destroy reports success")
209+
require.False(t, serves(runtime.port), "the listener must be gone before Destroy reports success")
210+
}

‎runtime.go‎

Lines changed: 65 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package main
22

33
import (
44
"context"
5+
"errors"
56
"fmt"
67
"io"
78
"maps"
@@ -758,6 +759,42 @@ func withOverrides(envs []*resources.EnvironmentVariable, overrides map[string]s
758759
// Lint is INHERITED from *pythonruntime.Runtime (uv run ruff check).
759760
// Build is INHERITED (no-op for Python).
760761

762+
// endExecution releases the execution resources the runtime holds: the file
763+
// watcher, the current process generation, and the runner environment. It must
764+
// be called with runnerMu held.
765+
//
766+
// The watcher goes first because it is the only thing that can ask for a
767+
// replacement process, and a teardown racing it can be handed back the very
768+
// thing it is tearing down. The remaining steps then all run even when an
769+
// earlier one fails, so one stubborn resource cannot strand the rest, and the
770+
// environment is forgotten only once it is actually down — a caller retrying
771+
// after a failure must still be able to reach it.
772+
func (s *Runtime) endExecution(ctx context.Context) error {
773+
// Cancel the watcher and let its Start goroutine's deferred close of Events
774+
// run exactly once — never close Events here, or it races that goroutine
775+
// into a "close of closed channel" panic.
776+
s.Base.StopWatcher()
777+
778+
var errs []error
779+
if err := s.stopRunner(ctx); err != nil {
780+
errs = append(errs, s.Wool.Wrapf(err, "cannot stop the fastapi app"))
781+
}
782+
if s.runnerEnvironment != nil {
783+
if err := s.runnerEnvironment.Shutdown(ctx); err != nil {
784+
errs = append(errs, s.Wool.Wrapf(err, "cannot shut down the runner environment"))
785+
} else {
786+
// A shut-down environment cannot serve another process — a docker
787+
// one has closed its client — so drop it and let a later Init build
788+
// a fresh one rather than hand Code and the REPL a dead handle.
789+
s.runnerEnvironment = nil
790+
s.FastAPI.Service.ActiveEnv = nil
791+
}
792+
}
793+
return errors.Join(errs...)
794+
}
795+
796+
// Stop releases the execution resources and keeps everything on disk, so a
797+
// later Init and Start bring the same service back.
761798
func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtimev0.StopResponse, error) {
762799
defer s.Wool.Catch()
763800
ctx = s.Wool.Inject(ctx)
@@ -766,53 +803,57 @@ func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtim
766803
defer s.runnerMu.Unlock()
767804

768805
s.Wool.Debug("stopping service")
769-
if err := s.stopRunner(ctx); err != nil {
806+
if err := s.endExecution(ctx); err != nil {
770807
return s.Base.Runtime.StopError(err)
771808
}
772-
if s.runnerEnvironment != nil {
773-
if err := s.runnerEnvironment.Shutdown(ctx); err != nil {
774-
s.Wool.Warn("error shutting down runner environment", wool.ErrField(err))
775-
}
776-
}
777-
// Cancel the watcher and let its Start goroutine's deferred close of Events
778-
// run exactly once — Stop must not close Events itself, or it races that
779-
// goroutine into a "close of closed channel" panic.
780-
s.Base.StopWatcher()
781809
return s.Base.Runtime.StopResponse()
782810
}
783811

812+
// Destroy ends execution and then removes what this service owns on the host.
813+
// It assumes nothing about what ran before it: `codefly` shutdown calls Destroy
814+
// directly, with no Stop in between, so ending execution is Destroy's own job
815+
// rather than a precondition on its callers. Safe before Init, after a failed
816+
// Init, after Stop, and repeated.
784817
func (s *Runtime) Destroy(ctx context.Context, req *runtimev0.DestroyRequest) (*runtimev0.DestroyResponse, error) {
785818
defer s.Wool.Catch()
786819
ctx = s.Wool.Inject(ctx)
787820

788821
s.Wool.Debug("Destroying service")
789822

790-
// A Destroy that skips this wipes the cache and tears down the container
791-
// while the service process is still running against them.
823+
var errs []error
824+
792825
s.runnerMu.Lock()
793-
if err := s.stopRunner(ctx); err != nil {
794-
s.Wool.Warn("cannot stop the fastapi app", wool.ErrField(err))
826+
if err := s.endExecution(ctx); err != nil {
827+
errs = append(errs, err)
795828
}
796829
s.runnerMu.Unlock()
797830

798-
s.Wool.Debug("removing cache")
799-
if err := shared.EmptyDir(ctx, s.cacheLocation); err != nil {
800-
// Best-effort: a failed cache wipe must NOT short-circuit Destroy and
801-
// skip the container teardown below — that would leak the running
802-
// container (the far more expensive resource).
803-
s.Wool.Warn("cannot remove cache", wool.ErrField(err))
831+
// Only once execution has ended: a process still running against the cache
832+
// repopulates what this removes, and is reading the venv being pulled out
833+
// from under it. An empty location means Init never claimed one — there is
834+
// nothing owned to clear, and EmptyDir rejects the empty path as an error.
835+
if s.cacheLocation != "" {
836+
s.Wool.Debug("removing cache")
837+
if err := shared.EmptyDir(ctx, s.cacheLocation); err != nil {
838+
errs = append(errs, s.Wool.Wrapf(err, "cannot remove cache"))
839+
}
804840
}
805841

806842
if s.Base.Runtime.IsContainerRuntime() {
843+
// A container outlives the agent that started it, so a Destroy whose
844+
// runtime never held one still has to reach it by name.
807845
s.Wool.Debug("running in container")
808846
dockerEnv, err := dockerrun.NewDockerEnvironment(ctx, runtimeImage, s.Service.SourceLocation, s.Base.Runtime.UniqueWithWorkspace())
809847
if err != nil {
810-
return s.Base.Runtime.DestroyError(err)
811-
}
812-
if err := dockerEnv.Shutdown(ctx); err != nil {
813-
return s.Base.Runtime.DestroyError(err)
848+
errs = append(errs, s.Wool.Wrapf(err, "cannot reach the container environment"))
849+
} else if err := dockerEnv.Shutdown(ctx); err != nil {
850+
errs = append(errs, s.Wool.Wrapf(err, "cannot shut down the container environment"))
814851
}
815852
}
853+
854+
if err := errors.Join(errs...); err != nil {
855+
return s.Base.Runtime.DestroyError(err)
856+
}
816857
return s.Base.Runtime.DestroyResponse()
817858
}
818859

0 commit comments

Comments
 (0)