diff options
| -rw-r--r-- | internal/cli/serve.go | 6 | ||||
| -rw-r--r-- | internal/executor/executor.go | 85 | ||||
| -rw-r--r-- | internal/executor/executor_test.go | 70 |
3 files changed, 161 insertions, 0 deletions
diff --git a/internal/cli/serve.go b/internal/cli/serve.go index 393b6bc..4054a3b 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -283,6 +283,12 @@ func serve(addr, basePath string) error { } go storyOrch.Run(ctx, scheduler.DefaultStoryPollInterval) + // Sweep stale dispatch workspace directories (internal/executor.Pool's + // RunWorkspaceCleanup) — ContainerRunner preserves a failed execution's + // workspace indefinitely for debugging with no expiry of its own, which + // otherwise accumulates forever. See that function's doc comment. + go pool.RunWorkspaceCleanup(ctx, executor.DefaultWorkspaceCleanupInterval, executor.DefaultWorkspaceCleanupMaxAge) + httpSrv := &http.Server{ Addr: addr, Handler: srv.Handler(), diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 659bcda..1025e2b 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "os" "path/filepath" "strings" "sync" @@ -1148,6 +1149,90 @@ func (p *Pool) RecoverStaleBlocked() { } } +// DefaultWorkspaceCleanupInterval is how often RunWorkspaceCleanup sweeps for +// stale dispatch workspaces when no interval is given. +const DefaultWorkspaceCleanupInterval = 1 * time.Hour + +// DefaultWorkspaceCleanupMaxAge is how long a preserved (failed) workspace is +// kept around for debugging before RunWorkspaceCleanup removes it. +const DefaultWorkspaceCleanupMaxAge = 24 * time.Hour + +// RunWorkspaceCleanup periodically removes stale dispatch workspace +// directories -- ContainerRunner.Run's os.MkdirTemp("", "claudomator-workspace-*") +// dirs -- left behind by failed executions. ContainerRunner.Run's own doc +// comment says "workspace is only removed on success. On failure, it's +// preserved for debugging" -- that preservation has no expiry of its own, so +// without this sweep those directories accumulate forever (confirmed in +// production: 161 stale directories, ~17.6GB, took the host to 100% disk +// full on 2026-07-10). Call this once, in a goroutine, at server startup -- +// mirrors StoryOrchestrator.Run's ticker shape exactly, including running +// one pass immediately before the first tick so a server that's been down +// doesn't wait a full interval before its first sweep. +func (p *Pool) RunWorkspaceCleanup(ctx context.Context, interval, maxAge time.Duration) { + if interval <= 0 { + interval = DefaultWorkspaceCleanupInterval + } + p.cleanupStaleWorkspaces(maxAge) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.cleanupStaleWorkspaces(maxAge) + } + } +} + +// cleanupStaleWorkspaces removes claudomator-workspace-* directories under +// os.TempDir() whose mtime is older than maxAge, except any directory still +// referenced as the sandbox_dir of a currently-BLOCKED task's latest +// execution -- that one is preserved intentionally for resumption (see +// ContainerRunner.Run) and must never be swept purely by age. A directory +// this sweep doesn't recognize (glob mismatch) or can't stat is left alone, +// not treated as an error worth aborting the whole pass over. +func (p *Pool) cleanupStaleWorkspaces(maxAge time.Duration) { + keep := map[string]bool{} + blocked, err := p.store.ListTasks(storage.TaskFilter{State: task.StateBlocked}) + if err != nil { + p.logger.Error("cleanupStaleWorkspaces: list blocked tasks", "error", err) + return + } + for _, t := range blocked { + execs, err := p.store.ListExecutions(t.ID) + if err != nil { + continue + } + for _, e := range execs { + if e.SandboxDir != "" { + keep[e.SandboxDir] = true + } + } + } + + matches, err := filepath.Glob(filepath.Join(os.TempDir(), "claudomator-workspace-*")) + if err != nil { + p.logger.Error("cleanupStaleWorkspaces: glob", "error", err) + return + } + cutoff := time.Now().Add(-maxAge) + for _, dir := range matches { + if keep[dir] { + continue + } + info, err := os.Stat(dir) + if err != nil || info.ModTime().After(cutoff) { + continue + } + if err := os.RemoveAll(dir); err != nil { + p.logger.Error("cleanupStaleWorkspaces: remove", "dir", dir, "error", err) + continue + } + p.logger.Info("cleanupStaleWorkspaces: removed stale workspace", "dir", dir, "age", time.Since(info.ModTime())) + } +} + // terminalFailureStates are dependency states that cause the waiting task to fail immediately. var terminalFailureStates = map[task.State]bool{ task.StateFailed: true, diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index a2228c3..c40f274 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -2170,3 +2170,73 @@ func TestPool_Shutdown_TimesOut(t *testing.T) { } close(unblock) // cleanup } + +// makeStaleWorkspaceDir creates a real directory matching the +// "claudomator-workspace-*" glob cleanupStaleWorkspaces looks for, backdates +// its mtime by age, and registers it for cleanup at test end regardless of +// whether the function under test already removed it. +func makeStaleWorkspaceDir(t *testing.T, age time.Duration) string { + t.Helper() + dir, err := os.MkdirTemp("", "claudomator-workspace-*") + if err != nil { + t.Fatalf("MkdirTemp: %v", err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + old := time.Now().Add(-age) + if err := os.Chtimes(dir, old, old); err != nil { + t.Fatalf("Chtimes: %v", err) + } + return dir +} + +func TestPool_CleanupStaleWorkspaces_RemovesOldUnreferencedDir(t *testing.T) { + store := testStore(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, map[string]Runner{"claude": &mockRunner{}}, store, logger) + + dir := makeStaleWorkspaceDir(t, 48*time.Hour) + + pool.cleanupStaleWorkspaces(24 * time.Hour) + + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Errorf("expected stale unreferenced workspace dir to be removed, stat err = %v", err) + } +} + +func TestPool_CleanupStaleWorkspaces_KeepsRecentDir(t *testing.T) { + store := testStore(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, map[string]Runner{"claude": &mockRunner{}}, store, logger) + + dir := makeStaleWorkspaceDir(t, 1*time.Hour) + + pool.cleanupStaleWorkspaces(24 * time.Hour) + + if _, err := os.Stat(dir); err != nil { + t.Errorf("expected recent workspace dir to be kept, got stat err = %v", err) + } +} + +func TestPool_CleanupStaleWorkspaces_KeepsBlockedTaskSandboxDirRegardlessOfAge(t *testing.T) { + store := testStore(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, map[string]Runner{"claude": &mockRunner{}}, store, logger) + + dir := makeStaleWorkspaceDir(t, 30*24*time.Hour) // deliberately far older than maxAge + + tk := makeTask("blocked-with-sandbox") + tk.State = task.StateBlocked + store.CreateTask(tk) + store.CreateExecution(&storage.Execution{ + ID: "exec-blocked-1", TaskID: tk.ID, + StartTime: time.Now().Add(-30 * 24 * time.Hour), + Status: "BLOCKED", + SandboxDir: dir, + }) + + pool.cleanupStaleWorkspaces(24 * time.Hour) + + if _, err := os.Stat(dir); err != nil { + t.Errorf("expected a BLOCKED task's sandbox_dir to be preserved regardless of age, got stat err = %v", err) + } +} |