MUL-6043: Keep repo checkout responsive during daemon GC (#6803)

* fix(daemon): keep repo checkout responsive during gc

Co-authored-by: multica-agent <[email protected]>

* fix(daemon): complete repo maintenance preemption

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: Eve <[email protected]>
Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
Multica Eve
2026-08-12 14:33:46 +08:00
committed by GitHub
co-authored by multica-agent Eve
parent 84dc02cd65
commit d57bfc50ee
18 changed files with 1782 additions and 169 deletions
+3
View File
@@ -220,6 +220,7 @@ Daemon behavior is configured via flags or environment variables:
| GC artifact TTL (open issues) | — | `MULTICA_GC_ARTIFACT_TTL` | `12h` (set `0` to disable) |
| GC artifact patterns | — | `MULTICA_GC_ARTIFACT_PATTERNS` | `node_modules,.next,.turbo` |
| GC repo cache TTL (`.repos`) | — | `MULTICA_GC_REPO_TTL` | `720h` (30d; set `0` to disable) |
| GC repo maintenance | — | `MULTICA_GC_REPO_MAINTENANCE_ENABLED` | `true` (set `false`/`0` to disable heavy Git maintenance only) |
| GC Hermes memory TTL (per-agent `memories/`) | — | `MULTICA_GC_HERMES_MEMORY_TTL` | `2160h` (90d; set `0` to disable) |
#### Workspace garbage collection
@@ -232,6 +233,8 @@ The daemon periodically scans `MULTICA_WORKSPACES_ROOT` and reclaims disk space
- **Repo cache eviction** — the bare git clones under `.repos/` are shared object stores: each task workdir is a `git worktree` off one of them rather than its own clone, so a task's `.git` is only a pointer file. They are evicted only when all of the following hold: the repo is no longer attached to any workspace this daemon watches, it has no worktrees left, and no task has created a worktree from it for `MULTICA_GC_REPO_TTL`. A cache created before this stamp existed is not treated as ancient — its clock starts at the first GC cycle that sees it, so upgrading does not wipe every cache. Evicting is safe by construction: the next task that needs the repo re-clones it on demand, so a wrong eviction costs a clone, not a failure.
Short worktree cleanup and eligible cache eviction continue on every GC cycle, including while agents are active. Heavy repo maintenance (`reflog expire` and `git gc`) starts only while the daemon is otherwise idle. A checkout or newly claimed task cancels it and takes priority; interrupted work remains pending for a later idle GC cycle. Operators can disable only these heavy commands with `MULTICA_GC_REPO_MAINTENANCE_ENABLED=false` without disabling worktree cleanup or cache eviction.
- **Hermes memory store reclamation** — a Hermes agent's long-term memory (`memories/`) lives at `<profile dir>/hermes-state/<agent-id>/<hermes-profile>/`, outside any task directory, so it survives across tasks and issues (see [Hermes agent memory](#hermes-agent-memory)). A store untouched for `MULTICA_GC_HERMES_MEMORY_TTL` is removed, giving a deleted agent's memory an eventual-reclamation guarantee. The default is deliberately long: these are a handful of markdown files, and reclaiming one is user-visible amnesia rather than a cache miss. A store a running task holds is never reclaimed.
Configured patterns are basename-only — entries containing `/` or `\` are silently dropped — and `.git` subtrees are never descended into. The managed Codex cache is matched by its exact relative path, so a repository's own `.sandbox-bin` is not removed unless an operator explicitly adds that basename to `MULTICA_GC_ARTIFACT_PATTERNS`. The default list (`node_modules`, `.next`, `.turbo`) is intentionally narrow; extend it per deployment if your repos consistently produce other regenerable directories (for example, `MULTICA_GC_ARTIFACT_PATTERNS=node_modules,.next,.turbo,target,__pycache__`). To disable artifact cleanup entirely, including the managed Codex cache, set `MULTICA_GC_ARTIFACT_TTL=0`.
+59 -15
View File
@@ -8,6 +8,7 @@ import (
"io"
"net/http"
"os"
"strconv"
"strings"
"time"
@@ -349,7 +350,7 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error {
return fmt.Errorf("get working directory: %w", err)
}
reqBody := map[string]string{
reqBody := map[string]any{
"url": repoURL,
"workspace_id": workspaceID,
"workdir": workDir,
@@ -357,6 +358,7 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error {
"agent_name": agentName,
"task_id": taskID,
"checkout_mode": strings.TrimSpace(os.Getenv("MULTICA_REPO_CHECKOUT_MODE")),
"retry_busy": true,
}
data, err := json.Marshal(reqBody)
@@ -364,21 +366,48 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error {
return fmt.Errorf("encode request: %w", err)
}
client := &http.Client{Timeout: 5 * time.Minute}
resp, err := client.Post(
fmt.Sprintf("http://127.0.0.1:%s/repo/checkout", daemonPort),
"application/json",
bytes.NewReader(data),
)
if err != nil {
return fmt.Errorf("connect to daemon: %w", err)
parentCtx := cmd.Context()
if parentCtx == nil {
parentCtx = context.Background()
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("checkout failed: %s", string(body))
ctx, cancel := context.WithTimeout(parentCtx, 5*time.Minute)
defer cancel()
client := &http.Client{}
checkoutURL := fmt.Sprintf("http://127.0.0.1:%s/repo/checkout", daemonPort)
var body []byte
for {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, checkoutURL, bytes.NewReader(data))
if err != nil {
return fmt.Errorf("create daemon checkout request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return fmt.Errorf("connect to daemon: %w", err)
}
body, err = io.ReadAll(resp.Body)
closeErr := resp.Body.Close()
if err != nil {
return fmt.Errorf("read daemon checkout response: %w", err)
}
if closeErr != nil {
return fmt.Errorf("close daemon checkout response: %w", closeErr)
}
if resp.StatusCode == http.StatusServiceUnavailable && resp.Header.Get("X-Multica-Retryable") == "repo-busy" {
delay := repoCheckoutRetryDelay(resp.Header.Get("Retry-After"), time.Now())
timer := time.NewTimer(delay)
select {
case <-ctx.Done():
timer.Stop()
return fmt.Errorf("connect to daemon: %w", context.Cause(ctx))
case <-timer.C:
continue
}
}
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("checkout failed: %s", string(body))
}
break
}
var result struct {
@@ -394,3 +423,18 @@ func runRepoCheckout(cmd *cobra.Command, args []string) error {
return nil
}
func repoCheckoutRetryDelay(value string, now time.Time) time.Duration {
const (
defaultDelay = time.Second
maxDelay = 30 * time.Second
)
value = strings.TrimSpace(value)
if seconds, err := strconv.Atoi(value); err == nil && seconds >= 0 {
return min(time.Duration(seconds)*time.Second, maxDelay)
}
if retryAt, err := http.ParseTime(value); err == nil {
return min(max(retryAt.Sub(now), time.Duration(0)), maxDelay)
}
return defaultDelay
}
+66 -1
View File
@@ -6,6 +6,7 @@ import (
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/spf13/cobra"
)
@@ -190,7 +191,7 @@ func TestRunRepoRemoveRejectsMissingRepoWithoutPatch(t *testing.T) {
}
func TestRunRepoCheckoutForwardsManagedCheckoutMode(t *testing.T) {
var body map[string]string
var body map[string]any
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost || r.URL.Path != "/repo/checkout" {
http.NotFound(w, r)
@@ -225,4 +226,68 @@ func TestRunRepoCheckoutForwardsManagedCheckoutMode(t *testing.T) {
if got := body["ref"]; got != "release/v2" {
t.Fatalf("ref = %q, want release/v2", got)
}
if got := body["retry_busy"]; got != true {
t.Fatalf("retry_busy = %v, want true", got)
}
}
func TestRunRepoCheckoutRetriesServiceUnavailable(t *testing.T) {
attempts := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
attempts++
if attempts == 1 {
w.Header().Set("X-Multica-Retryable", "repo-busy")
w.Header().Set("Retry-After", "0")
http.Error(w, "repository busy", http.StatusServiceUnavailable)
return
}
json.NewEncoder(w).Encode(map[string]string{
"path": "/work/repo",
"branch_name": "agent/test/task",
})
}))
defer srv.Close()
t.Setenv("MULTICA_DAEMON_PORT", strings.TrimPrefix(srv.URL, "http://127.0.0.1:"))
t.Setenv("MULTICA_WORKSPACE_ID", "ws-1")
t.Setenv("MULTICA_AGENT_NAME", "Test Agent")
t.Setenv("MULTICA_TASK_ID", "task-1")
if err := runRepoCheckout(&cobra.Command{}, []string{"https://github.com/org/repo.git"}); err != nil {
t.Fatalf("runRepoCheckout: %v", err)
}
if attempts != 2 {
t.Fatalf("checkout attempts = %d, want 2", attempts)
}
}
func TestRunRepoCheckoutDoesNotRetryUnmarkedServiceUnavailable(t *testing.T) {
attempts := 0
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
attempts++
http.Error(w, "daemon unavailable", http.StatusServiceUnavailable)
}))
defer srv.Close()
t.Setenv("MULTICA_DAEMON_PORT", strings.TrimPrefix(srv.URL, "http://127.0.0.1:"))
if err := runRepoCheckout(&cobra.Command{}, []string{"https://github.com/org/repo.git"}); err == nil {
t.Fatal("runRepoCheckout unexpectedly succeeded")
}
if attempts != 1 {
t.Fatalf("checkout attempts = %d, want 1", attempts)
}
}
func TestRepoCheckoutRetryDelay(t *testing.T) {
t.Parallel()
now := time.Date(2026, 8, 12, 0, 0, 0, 0, time.UTC)
if got := repoCheckoutRetryDelay("7", now); got != 7*time.Second {
t.Fatalf("seconds delay = %s, want 7s", got)
}
if got := repoCheckoutRetryDelay(now.Add(time.Minute).Format(http.TimeFormat), now); got != 30*time.Second {
t.Fatalf("capped date delay = %s, want 30s", got)
}
if got := repoCheckoutRetryDelay("invalid", now); got != time.Second {
t.Fatalf("default delay = %s, want 1s", got)
}
}
+3
View File
@@ -105,6 +105,7 @@ type Config struct {
GCArtifactTTL time.Duration // when a task has been completed for at least this long but its issue is still open, drop regenerable artifacts (default: 12h, set 0 to disable)
GCArtifactPatterns []string // basename patterns whose subtrees are removed during artifact cleanup (default: node_modules, .next, .turbo)
GCRepoTTL time.Duration // evict a cached bare repo under .repos once no task has created a worktree from it for this long, it has no worktrees left, and it is no longer attached to any watched workspace (default: 30d, set 0 to disable)
GCRepoMaintenanceEnabled bool // run reflog expiry and git gc after stale agent refs are removed (default: true; disable independently as an operational kill switch)
GCCodexSessionTTL time.Duration // reclaim a per-issue Codex session store (~/.codex/multica-sessions/<agent>/<issue>) untouched for at least this long, so a done/abandoned issue's conversation history does not accumulate forever (default: 14d, set 0 to disable)
GCHermesMemoryTTL time.Duration // reclaim a per-agent Hermes memory store (<profile dir>/hermes-state/<agent>/<profile>) untouched for at least this long, so a deleted agent's memory does not sit on disk forever (default: 90d, set 0 to disable)
AutoUpdateEnabled bool // periodically check for a newer CLI release and self-update when idle (default: true on Multica Cloud, false on self-host)
@@ -430,6 +431,7 @@ func LoadConfig(overrides Overrides) (Config, error) {
if err != nil {
return Config{}, err
}
gcRepoMaintenanceEnabled := boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true)
gcArtifactPatterns := patternsFromEnv("MULTICA_GC_ARTIFACT_PATTERNS", DefaultGCArtifactPatterns)
// Auto-update config: default -> env override -> CLI override.
@@ -481,6 +483,7 @@ func LoadConfig(overrides Overrides) (Config, error) {
GCArtifactTTL: gcArtifactTTL,
GCArtifactPatterns: gcArtifactPatterns,
GCRepoTTL: gcRepoTTL,
GCRepoMaintenanceEnabled: gcRepoMaintenanceEnabled,
GCCodexSessionTTL: gcCodexSessionTTL,
GCHermesMemoryTTL: gcHermesMemoryTTL,
AutoUpdateEnabled: autoUpdateEnabled,
+12
View File
@@ -35,6 +35,18 @@ func TestDefaultGCIntervalIsTwoHours(t *testing.T) {
}
}
func TestRepoMaintenanceKillSwitchDefaultsOnAndCanDisable(t *testing.T) {
t.Setenv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", "")
if !boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true) {
t.Fatal("repo maintenance kill switch should default to enabled")
}
t.Setenv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", "false")
if boolFromEnv("MULTICA_GC_REPO_MAINTENANCE_ENABLED", true) {
t.Fatal("repo maintenance kill switch should accept false")
}
}
func TestPatternsFromEnv_DropsSeparatorBearingEntries(t *testing.T) {
t.Setenv("MULTICA_GC_ARTIFACT_PATTERNS", "node_modules, .next ,foo/bar, ../etc, ,target")
got := patternsFromEnv("MULTICA_GC_ARTIFACT_PATTERNS", nil)
+60 -4
View File
@@ -272,7 +272,7 @@ type workspaceState struct {
taskRepoRefs map[string]map[string]string // taskID -> repo URL -> checkout ref
settings json.RawMessage // workspace settings (JSONB)
lastRepoSyncErr string
repoRefreshMu sync.Mutex
repoRefreshMu contextLock
// profileSetSig is a content hash of the workspace's custom runtime
// profile list (MUL-3332) as last seen from the server. An on-demand
// refresh compares the live signature with this cached value; any drift
@@ -297,6 +297,38 @@ type workspaceState struct {
builtinVersions map[string]string
}
// contextLock is a zero-value-ready mutex whose wait can be cancelled. Repo
// checkout requests use it for workspace refresh coalescing so disconnecting a
// client never leaves the handler stuck behind another cold-cache refresh.
type contextLock struct {
once sync.Once
token chan struct{}
}
func (l *contextLock) Lock(ctx context.Context) error {
if err := ctx.Err(); err != nil {
return context.Cause(ctx)
}
l.once.Do(func() {
l.token = make(chan struct{}, 1)
l.token <- struct{}{}
})
select {
case <-ctx.Done():
return context.Cause(ctx)
case <-l.token:
if err := ctx.Err(); err != nil {
l.token <- struct{}{}
return context.Cause(ctx)
}
return nil
}
}
func (l *contextLock) Unlock() {
l.token <- struct{}{}
}
type repoCacheBackend interface {
Lookup(workspaceID, url string) string
BarePath(workspaceID, url string) string
@@ -2776,10 +2808,22 @@ func (d *Daemon) waitBackgroundSyncs() {
}
func (d *Daemon) syncWorkspaceRepos(workspaceID string, repos []RepoData) {
d.syncWorkspaceReposContext(context.Background(), workspaceID, repos)
}
func (d *Daemon) syncWorkspaceReposContext(ctx context.Context, workspaceID string, repos []RepoData) {
if d.repoCache == nil {
return
}
if err := d.repoCache.Sync(workspaceID, repoDataToInfo(repos)); err != nil {
var err error
if cache, ok := d.repoCache.(interface {
SyncContext(context.Context, string, []repocache.RepoInfo) error
}); ok {
err = cache.SyncContext(ctx, workspaceID, repoDataToInfo(repos))
} else {
err = d.repoCache.Sync(workspaceID, repoDataToInfo(repos))
}
if err != nil {
d.setWorkspaceRepoSyncError(workspaceID, err.Error())
d.logger.Warn("repo cache sync failed", "workspace_id", workspaceID, "error", err)
return
@@ -3026,7 +3070,9 @@ func (d *Daemon) ensureRepoReady(ctx context.Context, workspaceID, repoURL strin
// sibling's refresh is fresh enough for our gate read.
cacheHitOnEntry := d.workspaceRepoAllowed(workspaceID, repoURL) && d.repoCache.Lookup(workspaceID, repoURL) != ""
ws.repoRefreshMu.Lock()
if err := ws.repoRefreshMu.Lock(ctx); err != nil {
return err
}
defer ws.repoRefreshMu.Unlock()
if !cacheHitOnEntry && d.workspaceRepoAllowed(workspaceID, repoURL) && d.repoCache.Lookup(workspaceID, repoURL) != "" {
@@ -3046,7 +3092,10 @@ func (d *Daemon) ensureRepoReady(ctx context.Context, workspaceID, repoURL strin
return nil
}
d.syncWorkspaceRepos(workspaceID, resp.Repos)
d.syncWorkspaceReposContext(ctx, workspaceID, resp.Repos)
if err := ctx.Err(); err != nil {
return context.Cause(ctx)
}
if d.repoCache.Lookup(workspaceID, repoURL) != "" {
return nil
@@ -4424,6 +4473,13 @@ func (d *Daemon) runBatchPoller(pollerCtx, parentCtx context.Context, sem chan i
d.logger.Info("task received", "task", shortID(t.ID), "target", taskTarget)
taskWG.Add(1)
d.activeTasks.Add(1)
if cache, ok := d.repoCache.(interface{ CancelMaintenance() }); ok {
// A task can reuse an existing worktree and never enter the
// checkout path that normally preempts repository maintenance.
// Cancel all low-priority maintenance before the agent starts so
// direct Git operations cannot overlap it.
cache.CancelMaintenance()
}
go func(t Task, slot int) {
defer taskWG.Done()
defer d.activeTasks.Add(-1)
+20
View File
@@ -3794,6 +3794,26 @@ func TestEnsureRepoReadyConcurrentMissRefreshesOnce(t *testing.T) {
}
}
func TestContextLockCancelsWaitWithoutConsumingToken(t *testing.T) {
t.Parallel()
var lock contextLock
if err := lock.Lock(context.Background()); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
if err := lock.Lock(ctx); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("Lock error = %v, want deadline exceeded", err)
}
lock.Unlock()
if err := lock.Lock(context.Background()); err != nil {
t.Fatalf("Lock after cancelled waiter: %v", err)
}
lock.Unlock()
}
func TestShellArgsFromEnv(t *testing.T) {
t.Setenv("MULTICA_CLAUDE_ARGS", `--max-turns 60 --append-system-prompt "multi word"`)
got, err := shellArgsFromEnv("MULTICA_CLAUDE_ARGS")
+222 -27
View File
@@ -11,6 +11,7 @@ import (
"time"
"github.com/multica-ai/multica/server/internal/daemon/execenv"
"github.com/multica-ai/multica/server/internal/daemon/processtree"
"github.com/multica-ai/multica/server/internal/daemon/repocache"
)
@@ -32,6 +33,7 @@ func (d *Daemon) gcLoop(ctx context.Context) {
"orphan_ttl", d.cfg.GCOrphanTTL,
"artifact_ttl", d.cfg.GCArtifactTTL,
"repo_ttl", d.cfg.GCRepoTTL,
"repo_maintenance_enabled", d.cfg.GCRepoMaintenanceEnabled,
"artifact_patterns", d.cfg.GCArtifactPatterns,
"managed_artifact_subpaths", execenv.ManagedReclaimableArtifactSubpaths(),
)
@@ -103,7 +105,7 @@ func (d *Daemon) runGC(ctx context.Context) {
// Prune stale worktree references from all bare repo caches, then evict the
// caches nothing needs anymore. These live outside any workspace directory
// and are never reclaimed by the task walk above.
d.pruneRepoWorktrees(root, stats)
d.pruneRepoWorktreesContext(ctx, root, stats)
// Reclaim per-issue Codex session stores idle past their TTL. These live
// under the shared ~/.codex home (outside WorkspacesRoot) so resume survives
@@ -749,8 +751,16 @@ func (d *Daemon) cleanTaskArtifactsMatching(taskDir string, matcher artifactMatc
// Non-fatal: errors during the walk are ignored so callers can report a
// best-effort byte count without aborting the whole GC cycle.
func dirSize(root string) int64 {
total, _ := dirSizeContext(context.Background(), root)
return total
}
func dirSizeContext(ctx context.Context, root string) (int64, error) {
var total int64
_ = filepath.WalkDir(root, func(_ string, entry os.DirEntry, err error) error {
err := filepath.WalkDir(root, func(_ string, entry os.DirEntry, err error) error {
if ctx.Err() != nil {
return context.Cause(ctx)
}
if err != nil {
return nil
}
@@ -772,17 +782,22 @@ func dirSize(root string) int64 {
}
return nil
})
return total
return total, err
}
const (
gitCmdTimeout = 30 * time.Second
gitMaintenanceTimeout = 10 * time.Minute
repoMaintenanceMarker = ".multica-maintenance-pending"
)
// pruneRepoWorktrees runs `git worktree prune` on all bare repos in the cache,
// then evicts the ones nothing needs anymore.
func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) {
d.pruneRepoWorktreesContext(context.Background(), workspacesRoot, stats)
}
func (d *Daemon) pruneRepoWorktreesContext(ctx context.Context, workspacesRoot string, stats *gcStats) {
reposRoot := filepath.Join(workspacesRoot, reposDirName)
wsEntries, err := os.ReadDir(reposRoot)
if err != nil {
@@ -790,6 +805,9 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) {
}
for _, wsEntry := range wsEntries {
if ctx.Err() != nil {
return
}
if !wsEntry.IsDir() {
continue
}
@@ -799,6 +817,9 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) {
continue
}
for _, repoEntry := range repoEntries {
if ctx.Err() != nil {
return
}
if !repoEntry.IsDir() {
continue
}
@@ -806,7 +827,7 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) {
if !isBareRepo(barePath) {
continue
}
d.maintainRepoCache(barePath, stats)
d.maintainRepoCache(ctx, barePath, stats)
}
// Drop the per-workspace directory once its last repo is gone.
if remaining, err := os.ReadDir(wsRepoDir); err == nil && len(remaining) == 0 {
@@ -815,17 +836,45 @@ func (d *Daemon) pruneRepoWorktrees(workspacesRoot string, stats *gcStats) {
}
}
func (d *Daemon) maintainRepoCache(barePath string, stats *gcStats) {
d.withRepoLock(barePath, func() {
d.pruneWorktreeLocked(barePath)
d.evictRepoCacheLocked(barePath, stats)
func (d *Daemon) maintainRepoCache(ctx context.Context, barePath string, stats *gcStats) {
d.withRepoMaintenance(ctx, barePath, func(maintenanceCtx context.Context) {
d.pruneWorktreeLocked(maintenanceCtx, barePath)
if maintenanceCtx.Err() == nil {
d.evictRepoCacheLocked(maintenanceCtx, barePath, stats)
}
})
}
// pruneWorktree runs only the maintenance half — prune stale worktrees and
// agent branches — without considering eviction.
func (d *Daemon) pruneWorktree(barePath string) {
d.withRepoLock(barePath, func() { d.pruneWorktreeLocked(barePath) })
d.withRepoMaintenance(context.Background(), barePath, func(ctx context.Context) {
d.pruneWorktreeLocked(ctx, barePath)
})
}
type repoMaintenanceBackend interface {
WithRepoMaintenance(context.Context, string, func(context.Context) error) (bool, error)
}
// withRepoMaintenance uses the cache's foreground-priority gate when
// available. The fallback preserves test/degraded backends that predate the
// optional interface without changing the repoCacheBackend contract.
func (d *Daemon) withRepoMaintenance(ctx context.Context, barePath string, fn func(context.Context)) {
if cache, ok := d.repoCache.(repoMaintenanceBackend); ok {
ran, err := cache.WithRepoMaintenance(ctx, barePath, func(maintenanceCtx context.Context) error {
fn(maintenanceCtx)
return nil
})
if err != nil && ctx.Err() == nil {
d.logger.Warn("gc: repo maintenance lock failed", "repo", barePath, "error", err)
}
if !ran {
d.logger.Debug("gc: repo maintenance skipped for foreground work", "repo", barePath)
}
return
}
d.withRepoLock(barePath, func() { fn(ctx) })
}
// withRepoLock serializes a mutation against Sync / CreateWorktree on the same
@@ -872,7 +921,7 @@ func (d *Daemon) withRepoLock(barePath string, fn func()) {
// Evicting wrongly costs time, not correctness: the next task that needs the
// repo takes the cache-miss path in ensureRepoReady, which re-syncs and
// re-clones on demand.
func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) {
func (d *Daemon) evictRepoCacheLocked(ctx context.Context, barePath string, stats *gcStats) {
if d.cfg.GCRepoTTL <= 0 {
return
}
@@ -882,8 +931,11 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) {
return
}
worktrees, err := linkedWorktreeCount(barePath)
worktrees, err := linkedWorktreeCountContext(ctx, barePath)
if err != nil {
if ctx.Err() != nil {
return
}
d.logger.Warn("gc: worktree count failed", "repo", barePath, "error", err)
return
}
@@ -908,7 +960,10 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) {
// the repo, which on a multi-GiB cache takes long enough for a workspace to
// re-attach underneath us — putting it between the check and the delete
// would reopen most of the window this check exists to close.
bytes := dirSize(barePath)
bytes, err := dirSizeContext(ctx, barePath)
if err != nil {
return
}
// Ask again immediately before deleting. The checks above run git and walk
// the filesystem, and a workspace can re-attach this repo while they do;
@@ -938,7 +993,11 @@ func (d *Daemon) evictRepoCacheLocked(barePath string, stats *gcStats) {
// worktree and marks the bare repo's own block with a `bare` line; only the
// linked blocks represent checkouts that would break if the repo went away.
func linkedWorktreeCount(barePath string) (int, error) {
out, err := runGitGCCommand(barePath, "worktree", "list", "--porcelain")
return linkedWorktreeCountContext(context.Background(), barePath)
}
func linkedWorktreeCountContext(ctx context.Context, barePath string) (int, error) {
out, err := runGitGCCommandContext(ctx, barePath, "worktree", "list", "--porcelain")
if err != nil {
return 0, err
}
@@ -969,8 +1028,11 @@ func linkedWorktreeCount(barePath string) (int, error) {
return count, nil
}
func (d *Daemon) pruneWorktreeLocked(barePath string) {
if out, err := runGitGCCommand(barePath, "worktree", "prune"); err != nil {
func (d *Daemon) pruneWorktreeLocked(ctx context.Context, barePath string) {
if out, err := runGitGCCommandContext(ctx, barePath, "worktree", "prune"); err != nil {
if ctx.Err() != nil {
return
}
d.logger.Warn("gc: worktree prune failed",
"repo", barePath,
"output", out,
@@ -978,14 +1040,20 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) {
)
}
activeBranches, err := agentWorktreeBranches(barePath)
activeBranches, err := agentWorktreeBranchesContext(ctx, barePath)
if err != nil {
if ctx.Err() != nil {
return
}
d.logger.Warn("gc: worktree branch scan failed", "repo", barePath, "error", err)
return
}
agentBranches, err := listAgentBranches(barePath)
agentBranches, err := listAgentBranchesContext(ctx, barePath)
if err != nil {
if ctx.Err() != nil {
return
}
d.logger.Warn("gc: agent branch scan failed", "repo", barePath, "error", err)
return
}
@@ -995,7 +1063,10 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) {
if _, ok := activeBranches[branch]; ok {
continue
}
if out, err := runGitGCCommand(barePath, "branch", "-D", "--", branch); err != nil {
if out, err := runGitGCCommandContext(ctx, barePath, "branch", "-D", "--", branch); err != nil {
if ctx.Err() != nil {
return
}
d.logger.Warn("gc: agent branch delete failed",
"repo", barePath,
"branch", branch,
@@ -1006,10 +1077,31 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) {
}
deleted++
}
if deleted == 0 {
markerPath := filepath.Join(barePath, repoMaintenanceMarker)
pending := deleted > 0
if pending {
if err := os.WriteFile(markerPath, []byte(time.Now().UTC().Format(time.RFC3339Nano)+"\n"), 0o600); err != nil {
d.logger.Warn("gc: record pending repo maintenance failed", "repo", barePath, "error", err)
}
d.logger.Info("gc: deleted stale agent branches", "repo", barePath, "count", deleted)
} else if _, err := os.Stat(markerPath); err == nil {
pending = true
}
if !pending {
return
}
if !d.cfg.GCRepoMaintenanceEnabled {
d.logger.Debug("gc: heavy repo maintenance disabled", "repo", barePath)
return
}
// Agent CLIs can mutate linked-worktree refs directly, outside the daemon's
// in-process repository gate. Do not start heavy maintenance while any task
// is active; a new task or checkout that arrives after this check preempts
// through the maintenance context below.
if d.activeTasks.Load() > 0 {
d.logger.Debug("gc: heavy repo maintenance deferred while tasks are active", "repo", barePath)
return
}
d.logger.Info("gc: deleted stale agent branches", "repo", barePath, "count", deleted)
// Heavier maintenance only runs when we actually removed refs, so we don't
// turn every GC tick into a full `git gc --prune` on every cached repo. The
@@ -1022,8 +1114,24 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) {
{args: []string{"reflog", "expire", "--expire=30.days", "--all"}, timeout: gitCmdTimeout},
{args: []string{"gc", "--prune=30.days"}, timeout: gitMaintenanceTimeout},
}
completed := true
for _, step := range maintenance {
if out, err := runGitCommand(barePath, step.timeout, step.args...); err != nil {
if ctx.Err() != nil || d.activeTasks.Load() > 0 {
return
}
before := snapshotRepoMaintenanceLocks(barePath)
if out, err := runGitCommandContext(ctx, barePath, step.timeout, step.args...); err != nil {
completed = false
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, repocache.ErrMaintenancePreempted) {
d.cleanupNewRepoMaintenanceLocks(barePath, before)
}
if errors.Is(context.Cause(ctx), repocache.ErrMaintenancePreempted) {
d.logger.Info("gc: git maintenance preempted for foreground work",
"repo", barePath,
"command", strings.Join(step.args, " "),
)
return
}
d.logger.Warn("gc: git maintenance failed",
"repo", barePath,
"command", strings.Join(step.args, " "),
@@ -1032,24 +1140,107 @@ func (d *Daemon) pruneWorktreeLocked(barePath string) {
)
}
}
if completed {
if err := os.Remove(markerPath); err != nil && !os.IsNotExist(err) {
d.logger.Warn("gc: clear pending repo maintenance failed", "repo", barePath, "error", err)
}
}
}
func runGitGCCommand(barePath string, args ...string) (string, error) {
return runGitCommand(barePath, gitCmdTimeout, args...)
return runGitGCCommandContext(context.Background(), barePath, args...)
}
func runGitCommand(barePath string, timeout time.Duration, args ...string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
return runGitCommandContext(context.Background(), barePath, timeout, args...)
}
func runGitGCCommandContext(ctx context.Context, barePath string, args ...string) (string, error) {
return runGitCommandContext(ctx, barePath, gitCmdTimeout, args...)
}
func runGitCommandContext(parent context.Context, barePath string, timeout time.Duration, args ...string) (string, error) {
ctx, cancel := context.WithTimeout(parent, timeout)
defer cancel()
cmdArgs := append([]string{"-C", barePath}, args...)
cmd := exec.CommandContext(ctx, "git", cmdArgs...)
out, err := cmd.CombinedOutput()
cmd := exec.Command("git", cmdArgs...)
out, err := processtree.CombinedOutput(ctx, cmd, 5*time.Second)
return strings.TrimSpace(string(out)), err
}
type repoMaintenanceLockSnapshot map[string]struct{}
// snapshotRepoMaintenanceLocks records only lock paths known to be produced by
// the maintenance commands below. Cleanup later removes a path only if it did
// not exist in this snapshot and the process tree is confirmed gone. Checkout
// waits on the same repo gate, and task dispatch waits for CancelMaintenance's
// barrier, so no agent Git work can create a competing lock before cleanup.
func snapshotRepoMaintenanceLocks(barePath string) repoMaintenanceLockSnapshot {
locks := make(repoMaintenanceLockSnapshot)
for _, path := range repoMaintenanceLockPaths(barePath) {
locks[path] = struct{}{}
}
return locks
}
func repoMaintenanceLockPaths(barePath string) []string {
var locks []string
for _, name := range []string{"gc.pid", "packed-refs.lock"} {
path := filepath.Join(barePath, name)
if info, err := os.Lstat(path); err == nil && info.Mode().IsRegular() {
locks = append(locks, path)
}
}
for _, root := range []string{
filepath.Join(barePath, "refs"),
filepath.Join(barePath, "logs", "refs"),
filepath.Join(barePath, "objects", "info"),
filepath.Join(barePath, "objects", "pack"),
} {
_ = filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error {
if err != nil {
return nil
}
if entry.Type()&linkedDirModes != 0 {
if entry.IsDir() {
return filepath.SkipDir
}
return nil
}
if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".lock") {
locks = append(locks, path)
}
return nil
})
}
return locks
}
func (d *Daemon) cleanupNewRepoMaintenanceLocks(barePath string, before repoMaintenanceLockSnapshot) {
for _, path := range repoMaintenanceLockPaths(barePath) {
if _, existed := before[path]; existed {
continue
}
rel, err := filepath.Rel(barePath, path)
if err != nil || rel == "." || strings.HasPrefix(rel, ".."+string(os.PathSeparator)) || filepath.IsAbs(rel) {
d.logger.Warn("gc: refused maintenance lock cleanup outside repo", "repo", barePath, "path", path)
continue
}
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
d.logger.Warn("gc: maintenance lock cleanup failed", "repo", barePath, "lock", rel, "error", err)
continue
}
d.logger.Info("gc: removed lock left by interrupted maintenance", "repo", barePath, "lock", rel)
}
}
func agentWorktreeBranches(barePath string) (map[string]struct{}, error) {
out, err := runGitGCCommand(barePath, "worktree", "list", "--porcelain")
return agentWorktreeBranchesContext(context.Background(), barePath)
}
func agentWorktreeBranchesContext(ctx context.Context, barePath string) (map[string]struct{}, error) {
out, err := runGitGCCommandContext(ctx, barePath, "worktree", "list", "--porcelain")
if err != nil {
return nil, err
}
@@ -1069,10 +1260,14 @@ func agentWorktreeBranches(barePath string) (map[string]struct{}, error) {
}
func listAgentBranches(barePath string) ([]string, error) {
return listAgentBranchesContext(context.Background(), barePath)
}
func listAgentBranchesContext(ctx context.Context, barePath string) ([]string, error) {
// Trailing slash narrows the pattern to the `agent/` namespace only. Without
// it, `for-each-ref` would also return a branch literally named `agent`,
// which `agentWorktreeBranches` ignores — that branch would then be deleted.
out, err := runGitGCCommand(barePath, "for-each-ref", "--format=%(refname:short)", "refs/heads/agent/")
out, err := runGitGCCommandContext(ctx, barePath, "for-each-ref", "--format=%(refname:short)", "refs/heads/agent/")
if err != nil {
return nil, err
}
@@ -78,6 +78,26 @@ func TestEvictRepoCache_RemovesIdleDetachedRepo(t *testing.T) {
}
}
func TestEvictRepoCacheRunsWhileAnotherTaskIsActive(t *testing.T) {
d := newGCTestDaemon(t, http.NewServeMux())
d.cfg.GCRepoTTL = 24 * time.Hour
wsID := "11111111-1111-1111-1111-111111111111"
barePath := newEvictTestRepo(t, d, wsID, testRepoURL)
writeLastUsed(t, barePath, time.Now().Add(-48*time.Hour))
d.activeTasks.Add(1)
defer d.activeTasks.Add(-1)
stats := runRepoGC(d)
if _, err := os.Stat(barePath); !os.IsNotExist(err) {
t.Fatalf("expected eligible detached repo to be evicted while another task is active, stat err = %v", err)
}
if stats.repoCachesReclaimed != 1 {
t.Errorf("repo_caches_reclaimed = %d, want 1", stats.repoCachesReclaimed)
}
}
// TestEvictRepoCache_KeepsRepoStillAttachedToWorkspace is the load-bearing
// guard. Sync re-clones every listed repo that is missing whenever a workspace
// registers, which happens on every daemon start — so evicting a repo the
+179 -8
View File
@@ -26,13 +26,14 @@ func newGCTestDaemon(t *testing.T, handler http.Handler) *Daemon {
root := t.TempDir()
cfg := Config{
WorkspacesRoot: root,
GCEnabled: true,
GCInterval: 1 * time.Hour,
GCTTL: 5 * 24 * time.Hour,
GCOrphanTTL: 30 * 24 * time.Hour,
GCArtifactTTL: 12 * time.Hour,
GCArtifactPatterns: []string{"node_modules", ".next", ".turbo"},
WorkspacesRoot: root,
GCEnabled: true,
GCInterval: 1 * time.Hour,
GCTTL: 5 * 24 * time.Hour,
GCOrphanTTL: 30 * 24 * time.Hour,
GCArtifactTTL: 12 * time.Hour,
GCArtifactPatterns: []string{"node_modules", ".next", ".turbo"},
GCRepoMaintenanceEnabled: true,
}
d := New(cfg, slog.Default())
d.client = NewClient(srv.URL)
@@ -1053,6 +1054,28 @@ func TestPruneWorktree_RemovesOnlyStaleAgentBranches(t *testing.T) {
}
}
func TestMaintainRepoCacheRunsLightCleanupWhileTaskActive(t *testing.T) {
t.Parallel()
d := newGCTestDaemon(t, http.NewServeMux())
sourceRepo := createGCGitRepo(t)
barePath := filepath.Join(t.TempDir(), "cache.git")
runGitForGC(t, "", "clone", "--bare", sourceRepo, barePath)
const staleBranch = "agent/stale/87654321"
runGitForGC(t, "", "-C", barePath, "branch", staleBranch, "HEAD")
d.activeTasks.Add(1)
defer d.activeTasks.Add(-1)
d.maintainRepoCache(context.Background(), barePath, &gcStats{})
if gitRefExists(t, barePath, "refs/heads/"+staleBranch) {
t.Fatalf("expected stale branch %q to be deleted while a task is active", staleBranch)
}
if _, err := os.Stat(filepath.Join(barePath, repoMaintenanceMarker)); err != nil {
t.Fatalf("heavy maintenance should remain pending while a task is active: %v", err)
}
}
// TestPruneWorktree_IgnoresLiteralAgentBranch ensures the GC pattern is scoped
// to the `agent/` namespace. A repo whose only `agent`-shaped ref is the
// literal `refs/heads/agent` (no slash) must be left untouched — the
@@ -1111,6 +1134,29 @@ func TestPruneWorktree_SkipsMaintenanceWhenNothingDeleted(t *testing.T) {
}
}
func TestPruneWorktree_RetriesPendingMaintenanceWithoutAnotherBranchDeletion(t *testing.T) {
t.Parallel()
d := newGCTestDaemon(t, http.NewServeMux())
sourceRepo := createGCGitRepo(t)
barePath := filepath.Join(t.TempDir(), "cache.git")
runGitForGC(t, "", "clone", "--bare", sourceRepo, barePath)
sentinelPath := writeOldLooseBlob(t, barePath, "pending-maintenance", 60*24*time.Hour)
markerPath := filepath.Join(barePath, repoMaintenanceMarker)
if err := os.WriteFile(markerPath, []byte("pending\n"), 0o600); err != nil {
t.Fatal(err)
}
d.pruneWorktree(barePath)
if _, err := os.Stat(sentinelPath); !os.IsNotExist(err) {
t.Fatalf("expected pending maintenance to prune sentinel, stat err=%v", err)
}
if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
t.Fatalf("expected completed maintenance to clear pending marker, stat err=%v", err)
}
}
// writeOldLooseBlob writes a dangling loose-object blob to the bare repo and
// backdates its mtime so `git gc --prune=30.days` will consider it prunable.
// Returns the absolute path to the loose object on disk.
@@ -1137,7 +1183,7 @@ func writeOldLooseBlob(t *testing.T, barePath, content string, age time.Duration
return loose
}
func TestPruneWorktree_SerializesWithCreateWorktree(t *testing.T) {
func TestPruneWorktree_LegacyCacheFallbackSerializesWithCreateWorktree(t *testing.T) {
t.Parallel()
d := newGCTestDaemon(t, http.NewServeMux())
@@ -1209,6 +1255,131 @@ func TestPruneWorktree_SerializesWithCreateWorktree(t *testing.T) {
}
}
func TestCleanupNewRepoMaintenanceLocksPreservesPreexistingFiles(t *testing.T) {
t.Parallel()
d := newGCTestDaemon(t, http.NewServeMux())
barePath := t.TempDir()
preexisting := filepath.Join(barePath, "logs", "refs", "heads", "main.lock")
if err := os.MkdirAll(filepath.Dir(preexisting), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(preexisting, []byte("existing"), 0o644); err != nil {
t.Fatal(err)
}
before := snapshotRepoMaintenanceLocks(barePath)
created := []string{
filepath.Join(barePath, "refs", "remotes", "origin", "main.lock"),
filepath.Join(barePath, "objects", "pack", "multi-pack-index.lock"),
filepath.Join(barePath, "packed-refs.lock"),
filepath.Join(barePath, "gc.pid"),
}
for _, path := range created {
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, []byte("new"), 0o644); err != nil {
t.Fatal(err)
}
}
outsideWhitelist := filepath.Join(barePath, "hooks", "user.lock")
if err := os.MkdirAll(filepath.Dir(outsideWhitelist), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(outsideWhitelist, []byte("keep"), 0o644); err != nil {
t.Fatal(err)
}
d.cleanupNewRepoMaintenanceLocks(barePath, before)
for _, path := range created {
if _, err := os.Stat(path); !os.IsNotExist(err) {
t.Errorf("new maintenance lock %s survived, stat error = %v", path, err)
}
}
for _, path := range []string{preexisting, outsideWhitelist} {
if _, err := os.Stat(path); err != nil {
t.Errorf("unowned lock %s was removed: %v", path, err)
}
}
}
func TestPruneWorktreePreemptionCleansLocksBeforeTaskStarts(t *testing.T) {
if _, err := os.Stat("/bin/sh"); err != nil {
t.Skipf("requires a POSIX shell: %v", err)
}
realGit, err := exec.LookPath("git")
if err != nil {
t.Skipf("git not found: %v", err)
}
d := newGCTestDaemon(t, http.NewServeMux())
sourceRepo := createGCGitRepo(t)
cache := repocache.New(filepath.Join(d.cfg.WorkspacesRoot, ".repos"), slog.Default())
if err := cache.Sync("ws1", []repocache.RepoInfo{{URL: sourceRepo}}); err != nil {
t.Fatalf("cache sync failed: %v", err)
}
d.repoCache = cache
barePath := cache.Lookup("ws1", sourceRepo)
runGitForGC(t, "", "-C", barePath, "branch", "agent/stale/87654321", "HEAD")
lockPath := filepath.Join(barePath, "refs", "remotes", "origin", "main.lock")
if err := os.MkdirAll(filepath.Dir(lockPath), 0o755); err != nil {
t.Fatal(err)
}
startedPath := filepath.Join(t.TempDir(), "maintenance-started")
binDir := t.TempDir()
fakeGit := filepath.Join(binDir, "git")
script := `#!/bin/sh
if [ "$3" = "reflog" ] && [ "$4" = "expire" ]; then
: > "$2/refs/remotes/origin/main.lock"
: > "$MULTICA_TEST_MAINTENANCE_STARTED"
trap 'exit 143' TERM INT
while :; do sleep 1; done
fi
exec "$MULTICA_TEST_REAL_GIT" "$@"
`
if err := os.WriteFile(fakeGit, []byte(script), 0o755); err != nil {
t.Fatal(err)
}
t.Setenv("MULTICA_TEST_REAL_GIT", realGit)
t.Setenv("MULTICA_TEST_MAINTENANCE_STARTED", startedPath)
t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
pruneDone := make(chan struct{})
go func() {
d.pruneWorktree(barePath)
close(pruneDone)
}()
deadline := time.Now().Add(5 * time.Second)
for {
if _, err := os.Stat(startedPath); err == nil {
break
}
if time.Now().After(deadline) {
t.Fatal("timed out waiting for heavy maintenance to start")
}
time.Sleep(10 * time.Millisecond)
}
// Match production task dispatch order: mark the task active, then cancel
// maintenance and wait for its cleanup barrier before agent code starts.
d.activeTasks.Add(1)
defer d.activeTasks.Add(-1)
cache.CancelMaintenance()
select {
case <-pruneDone:
case <-time.After(5 * time.Second):
t.Fatal("pruneWorktree did not finish after preemption")
}
if _, err := os.Stat(lockPath); !os.IsNotExist(err) {
t.Fatalf("maintenance lock survived real task preemption, stat error = %v", err)
}
if err := cache.Fetch(barePath); err != nil {
t.Fatalf("fetch did not recover after preempted maintenance cleanup: %v", err)
}
}
type blockingRepoCache struct {
inner *repocache.Cache
entered chan struct{}
+51 -5
View File
@@ -51,9 +51,14 @@ type HealthResponse struct {
// ActiveTaskCount remains the compatibility/safety count of every claimed
// handleTask lifecycle. The additive counters split actual provider
// execution from local-directory parking for throughput and diagnostics.
ActiveTaskCount int64 `json:"active_task_count"`
RunningTaskCount int64 `json:"running_task_count"`
ResourceWaitTaskCount int64 `json:"resource_wait_task_count"`
ActiveTaskCount int64 `json:"active_task_count"`
RunningTaskCount int64 `json:"running_task_count"`
ResourceWaitTaskCount int64 `json:"resource_wait_task_count"`
// Repo maintenance stays a liveness-safe background activity, so health
// remains HTTP 200/running. These additive counters explain degraded repo
// checkout capacity to operators without exposing local cache paths.
RepoMaintenanceActive int `json:"repo_maintenance_active,omitempty"`
RepoCheckoutWaiters int `json:"repo_checkout_waiters,omitempty"`
Agents []string `json:"agents"`
// SkippedAgents maps a provider that WAS discovered on this machine to the
// reason the last registration round dropped it (version undetectable,
@@ -97,8 +102,18 @@ type repoCheckoutRequest struct {
AgentName string `json:"agent_name"`
TaskID string `json:"task_id"`
CheckoutMode string `json:"checkout_mode,omitempty"`
// RetryBusy is sent by clients that understand 503 + Retry-After. Older
// clients omit it and retain their historical unbounded lock-wait behavior.
RetryBusy bool `json:"retry_busy,omitempty"`
}
const (
repoCheckoutLockWaitTimeout = 10 * time.Second
repoCheckoutRetryAfter = 2 * time.Second
repoCheckoutRetryHeader = "X-Multica-Retryable"
repoCheckoutRetryValueBusy = "repo-busy"
)
// healthHandler returns the /health HTTP handler. Extracted from serveHealth
// so tests can exercise it without spinning up a listener.
func (d *Daemon) healthHandler(startedAt time.Time) http.HandlerFunc {
@@ -149,6 +164,11 @@ func (d *Daemon) healthHandler(startedAt time.Time) http.HandlerFunc {
ReloadPendingReason: d.reloadPending(),
Workspaces: wsList,
}
if reporter, ok := d.repoCache.(interface{ Activity() repocache.Activity }); ok {
activity := reporter.Activity()
resp.RepoMaintenanceActive = activity.MaintenanceActive
resp.RepoCheckoutWaiters = activity.ForegroundWaiters
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
@@ -234,6 +254,10 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc {
}
if err := d.ensureRepoReady(r.Context(), req.WorkspaceID, req.URL); err != nil {
if r.Context().Err() != nil {
d.logger.Debug("repo checkout readiness cancelled", "url", req.URL, "error", err)
return
}
statusCode := http.StatusInternalServerError
if errors.Is(err, ErrRepoNotConfigured) {
statusCode = http.StatusBadRequest
@@ -248,7 +272,7 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc {
checkoutRef = d.taskRepoDefaultRef(req.WorkspaceID, req.TaskID, req.URL)
}
result, err := d.repoCache.CreateWorktree(repocache.WorktreeParams{
params := repocache.WorktreeParams{
WorkspaceID: req.WorkspaceID,
RepoURL: req.URL,
WorkDir: req.WorkDir,
@@ -257,8 +281,30 @@ func (d *Daemon) repoCheckoutHandler() http.HandlerFunc {
TaskID: req.TaskID,
CoAuthoredByEnabled: d.workspaceCoAuthoredByEnabled(req.WorkspaceID),
IsolatedGitMetadata: req.CheckoutMode == repoCheckoutModeIsolated,
})
}
if req.RetryBusy {
params.LockWaitTimeout = repoCheckoutLockWaitTimeout
}
var result *repocache.WorktreeResult
var err error
if cache, ok := d.repoCache.(interface {
CreateWorktreeContext(context.Context, repocache.WorktreeParams) (*repocache.WorktreeResult, error)
}); ok {
result, err = cache.CreateWorktreeContext(r.Context(), params)
} else {
result, err = d.repoCache.CreateWorktree(params)
}
if err != nil {
if errors.Is(err, repocache.ErrRepoBusy) && req.RetryBusy {
w.Header().Set(repoCheckoutRetryHeader, repoCheckoutRetryValueBusy)
w.Header().Set("Retry-After", fmt.Sprintf("%.0f", repoCheckoutRetryAfter.Seconds()))
http.Error(w, "repository is busy with another operation; retry later", http.StatusServiceUnavailable)
return
}
if r.Context().Err() != nil {
d.logger.Debug("repo checkout cancelled", "url", req.URL, "error", err)
return
}
d.logger.Error("repo checkout failed", "url", req.URL, "error", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
+69 -1
View File
@@ -196,6 +196,30 @@ func TestHealthHandlerActiveTaskCountTracksCounter(t *testing.T) {
assertActiveTaskCount(t, handler, 0)
}
func TestHealthHandlerReportsRepoCoordinationActivity(t *testing.T) {
t.Parallel()
cache := &activityRepoCache{
activity: repocache.Activity{MaintenanceActive: 1, ForegroundWaiters: 3},
}
d := &Daemon{
cfg: Config{CLIVersion: "v1.0.0"},
repoCache: cache,
workspaces: map[string]*workspaceState{},
logger: slog.Default(),
}
rec := httptest.NewRecorder()
d.healthHandler(time.Now()).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/health", nil))
var resp HealthResponse
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp.RepoMaintenanceActive != 1 || resp.RepoCheckoutWaiters != 3 {
t.Fatalf("repo activity = maintenance:%d waiters:%d, want 1/3", resp.RepoMaintenanceActive, resp.RepoCheckoutWaiters)
}
}
func TestShutdownHandlerPostCancelsDaemonContext(t *testing.T) {
t.Parallel()
@@ -371,7 +395,33 @@ func TestRepoCheckoutRejectsUnknownMode(t *testing.T) {
}
}
func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache *recordingRepoCache) *Daemon {
func TestRepoCheckoutReturnsRetryableBusyToCapableClient(t *testing.T) {
t.Parallel()
const workspaceID = "ws-checkout"
const repoURL = "https://github.com/org/repo.git"
cache := &busyRepoCache{recordingRepoCache: recordingRepoCache{lookupPath: "/cache/org/repo.git"}}
d := newRepoCheckoutTestDaemon(t, workspaceID, repoURL, cache)
rec := httptest.NewRecorder()
body := strings.NewReader(`{"url":"` + repoURL + `","workspace_id":"` + workspaceID + `","workdir":"/tmp/work","task_id":"task-1","retry_busy":true}`)
d.repoCheckoutHandler().ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/repo/checkout", body))
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("expected 503, got %d: %s", rec.Code, rec.Body.String())
}
if got := rec.Header().Get("Retry-After"); got != "2" {
t.Fatalf("Retry-After = %q, want 2", got)
}
if got := rec.Header().Get(repoCheckoutRetryHeader); got != repoCheckoutRetryValueBusy {
t.Fatalf("%s = %q, want %q", repoCheckoutRetryHeader, got, repoCheckoutRetryValueBusy)
}
if got := cache.lastCreateParams().LockWaitTimeout; got != repoCheckoutLockWaitTimeout {
t.Fatalf("lock wait timeout = %s, want %s", got, repoCheckoutLockWaitTimeout)
}
}
func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache repoCacheBackend) *Daemon {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet || r.URL.Path != "/api/daemon/workspaces/"+workspaceID+"/repos" {
@@ -396,6 +446,24 @@ func newRepoCheckoutTestDaemon(t *testing.T, workspaceID, repoURL string, cache
}
}
type busyRepoCache struct {
recordingRepoCache
}
type activityRepoCache struct {
recordingRepoCache
activity repocache.Activity
}
func (c *activityRepoCache) Activity() repocache.Activity { return c.activity }
func (c *busyRepoCache) CreateWorktreeContext(_ context.Context, params repocache.WorktreeParams) (*repocache.WorktreeResult, error) {
c.mu.Lock()
c.params = append(c.params, params)
c.mu.Unlock()
return nil, repocache.ErrRepoBusy
}
type blockingLookupRepoCache struct {
path string
lookupSeen chan struct{}
@@ -0,0 +1,78 @@
//go:build !windows
package processtree
import (
"errors"
"fmt"
"os"
"os/exec"
"syscall"
"time"
)
const processTreeFinishTimeout = 5 * time.Second
type controller struct{}
func newController(cmd *exec.Cmd) (*controller, error) {
if cmd.SysProcAttr == nil {
cmd.SysProcAttr = &syscall.SysProcAttr{}
}
cmd.SysProcAttr.Setpgid = true
return &controller{}, nil
}
func (*controller) attach(_ *exec.Cmd) error { return nil }
func (*controller) interrupt(cmd *exec.Cmd) error {
if cmd.Process == nil {
return os.ErrProcessDone
}
if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM); err != nil {
if errors.Is(err, syscall.ESRCH) {
return nil
}
return err
}
return nil
}
func (*controller) stop(cmd *exec.Cmd) error {
if cmd.Process == nil {
return os.ErrProcessDone
}
if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil {
if errors.Is(err, syscall.ESRCH) {
return nil
}
return cmd.Process.Kill()
}
return nil
}
func (c *controller) finish(cmd *exec.Cmd) error {
if cmd.Process == nil {
return nil
}
pid := cmd.Process.Pid
if err := syscall.Kill(-pid, 0); errors.Is(err, syscall.ESRCH) {
return nil
}
// A normally-exited leader can still leave a descendant holding inherited
// pipes or repository locks. Kill the remaining group before returning
// ownership of the repository to another operation.
if err := syscall.Kill(-pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
return err
}
deadline := time.Now().Add(processTreeFinishTimeout)
for time.Now().Before(deadline) {
if err := syscall.Kill(-pid, 0); errors.Is(err, syscall.ESRCH) {
return nil
}
time.Sleep(10 * time.Millisecond)
}
return fmt.Errorf("process group %d still active after %s", pid, processTreeFinishTimeout)
}
func (*controller) close() {}
@@ -0,0 +1,130 @@
//go:build windows
package processtree
import (
"fmt"
"os/exec"
"syscall"
"time"
"unsafe"
"golang.org/x/sys/windows"
)
const processTreeFinishTimeout = 5 * time.Second
type controller struct {
job windows.Handle
}
type jobObjectBasicAccountingInformation struct {
TotalUserTime int64
TotalKernelTime int64
ThisPeriodTotalUserTime int64
ThisPeriodTotalKernelTime int64
TotalPageFaultCount uint32
TotalProcesses uint32
ActiveProcesses uint32
TotalTerminatedProcesses uint32
}
func newController(cmd *exec.Cmd) (*controller, error) {
job, err := windows.CreateJobObject(nil, nil)
if err != nil {
return nil, err
}
info := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{}
info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
if _, err := windows.SetInformationJobObject(
job,
windows.JobObjectExtendedLimitInformation,
uintptr(unsafe.Pointer(&info)),
uint32(unsafe.Sizeof(info)),
); err != nil {
windows.CloseHandle(job)
return nil, fmt.Errorf("set KILL_ON_JOB_CLOSE: %w", err)
}
if cmd.SysProcAttr == nil {
cmd.SysProcAttr = &syscall.SysProcAttr{}
}
cmd.SysProcAttr.HideWindow = true
return &controller{job: job}, nil
}
func (c *controller) attach(cmd *exec.Cmd) error {
process, err := windows.OpenProcess(
windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE,
false,
uint32(cmd.Process.Pid),
)
if err != nil {
return fmt.Errorf("open process: %w", err)
}
defer windows.CloseHandle(process)
if err := windows.AssignProcessToJobObject(c.job, process); err != nil {
return fmt.Errorf("assign process to job object: %w", err)
}
return nil
}
func (c *controller) stop(_ *exec.Cmd) error {
if err := windows.TerminateJobObject(c.job, 1); err != nil {
active, queryErr := c.activeProcesses()
if queryErr == nil && active == 0 {
return nil
}
return err
}
return nil
}
// Windows Job Objects have no SIGTERM equivalent; terminating the job is the
// graceful-attempt and hard-stop operation alike.
func (c *controller) interrupt(cmd *exec.Cmd) error { return c.stop(cmd) }
func (c *controller) finish(_ *exec.Cmd) error {
active, err := c.activeProcesses()
if err != nil {
return err
}
if active > 0 {
if err := windows.TerminateJobObject(c.job, 1); err != nil {
return err
}
}
deadline := time.Now().Add(processTreeFinishTimeout)
for time.Now().Before(deadline) {
active, err = c.activeProcesses()
if err != nil {
return err
}
if active == 0 {
return nil
}
time.Sleep(10 * time.Millisecond)
}
return fmt.Errorf("process job still active after %s", processTreeFinishTimeout)
}
func (c *controller) activeProcesses() (uint32, error) {
var info jobObjectBasicAccountingInformation
if err := windows.QueryInformationJobObject(
c.job,
windows.JobObjectBasicAccountingInformation,
uintptr(unsafe.Pointer(&info)),
uint32(unsafe.Sizeof(info)),
nil,
); err != nil {
return 0, fmt.Errorf("query job object members: %w", err)
}
return info.ActiveProcesses, nil
}
func (c *controller) close() {
if c.job == 0 {
return
}
_ = windows.CloseHandle(c.job)
c.job = 0
}
+92
View File
@@ -0,0 +1,92 @@
// Package processtree runs bounded helper commands whose descendants must not
// survive cancellation. It uses a Unix process group or a Windows Job Object.
package processtree
import (
"bytes"
"context"
"errors"
"fmt"
"os/exec"
"time"
)
const gracefulStopTimeout = time.Second
// CombinedOutput runs an unstarted command and returns its combined output.
// Cancellation terminates the entire process tree, waits for it to disappear,
// and returns the context cause rather than a platform-specific exit status.
func CombinedOutput(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) ([]byte, error) {
var output bytes.Buffer
cmd.Stdout = &output
cmd.Stderr = &output
err := run(ctx, cmd, waitDelay)
return output.Bytes(), err
}
// Output is the process-tree-safe equivalent of exec.Cmd.Output.
func Output(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) ([]byte, error) {
var stdout, stderr bytes.Buffer
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := run(ctx, cmd, waitDelay)
if exitErr := (*exec.ExitError)(nil); errors.As(err, &exitErr) {
exitErr.Stderr = append([]byte(nil), stderr.Bytes()...)
}
return stdout.Bytes(), err
}
// Run executes an unstarted command while owning its complete process tree.
func Run(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) error {
return run(ctx, cmd, waitDelay)
}
func run(ctx context.Context, cmd *exec.Cmd, waitDelay time.Duration) error {
if err := ctx.Err(); err != nil {
return context.Cause(ctx)
}
controller, err := newController(cmd)
if err != nil {
return fmt.Errorf("create process-tree controller: %w", err)
}
defer controller.close()
cmd.WaitDelay = waitDelay
if err := cmd.Start(); err != nil {
return err
}
if err := controller.attach(cmd); err != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = controller.finish(cmd)
return fmt.Errorf("attach process tree: %w", err)
}
waitDone := make(chan error, 1)
go func() { waitDone <- cmd.Wait() }()
var stopErr error
cancelled := false
select {
case err = <-waitDone:
case <-ctx.Done():
cancelled = true
stopErr = controller.interrupt(cmd)
timer := time.NewTimer(gracefulStopTimeout)
select {
case err = <-waitDone:
timer.Stop()
case <-timer.C:
stopErr = errors.Join(stopErr, controller.stop(cmd))
err = <-waitDone
}
}
finishErr := controller.finish(cmd)
if lifecycleErr := errors.Join(stopErr, finishErr); lifecycleErr != nil {
return fmt.Errorf("stop process tree: %w", lifecycleErr)
}
if cancelled {
return context.Cause(ctx)
}
return err
}
@@ -0,0 +1,41 @@
//go:build !windows
package processtree
import (
"context"
"errors"
"os"
"os/exec"
"path/filepath"
"testing"
"time"
)
func TestCombinedOutputKillsDescendantsHoldingOutput(t *testing.T) {
t.Parallel()
dir := t.TempDir()
marker := filepath.Join(dir, "descendant-survived")
script := filepath.Join(dir, "helper.sh")
body := "#!/bin/sh\n(sleep 0.5; echo leaked > \"$1\") &\nwait\n"
if err := os.WriteFile(script, []byte(body), 0o755); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
started := time.Now()
_, err := CombinedOutput(ctx, exec.Command(script, marker), 250*time.Millisecond)
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("CombinedOutput error = %v, want deadline exceeded", err)
}
if elapsed := time.Since(started); elapsed > time.Second {
t.Fatalf("CombinedOutput took %s, want a hard bounded return", elapsed)
}
time.Sleep(600 * time.Millisecond)
if _, err := os.Stat(marker); !os.IsNotExist(err) {
t.Fatalf("descendant survived process-tree cancellation, stat error = %v", err)
}
}
File diff suppressed because it is too large Load Diff
@@ -119,6 +119,203 @@ func TestRunGitOutputTimesOut(t *testing.T) {
}
}
func TestRepoMaintenanceYieldsToForeground(t *testing.T) {
t.Parallel()
cache := New(t.TempDir(), testLogger())
const barePath = "/cache/repo.git"
entered := make(chan struct{})
maintenanceDone := make(chan error, 1)
go func() {
ran, err := cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error {
close(entered)
<-ctx.Done()
return context.Cause(ctx)
})
if !ran && err == nil {
err = errors.New("maintenance did not run")
}
maintenanceDone <- err
}()
<-entered
foregroundDone := make(chan error, 1)
go func() {
foregroundDone <- cache.WithRepoLockContext(context.Background(), barePath, func() error { return nil })
}()
select {
case err := <-foregroundDone:
if err != nil {
t.Fatalf("foreground lock: %v", err)
}
case <-time.After(time.Second):
t.Fatal("foreground work did not preempt maintenance")
}
if err := <-maintenanceDone; !errors.Is(err, ErrMaintenancePreempted) {
t.Fatalf("maintenance error = %v, want preemption", err)
}
activity := cache.Activity()
if activity.MaintenanceActive != 0 || activity.ForegroundWaiters != 0 {
t.Fatalf("activity after release = %+v, want zero", activity)
}
}
func TestCancelMaintenanceStopsMaintenanceWithoutARepoCheckout(t *testing.T) {
t.Parallel()
cache := New(t.TempDir(), testLogger())
const barePath = "/cache/reused-worktree.git"
entered := make(chan struct{})
done := make(chan error, 1)
go func() {
ran, err := cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error {
close(entered)
<-ctx.Done()
return context.Cause(ctx)
})
if !ran && err == nil {
err = errors.New("maintenance did not run")
}
done <- err
}()
<-entered
cache.CancelMaintenance()
select {
case err := <-done:
if !errors.Is(err, ErrMaintenancePreempted) {
t.Fatalf("maintenance error = %v, want preemption", err)
}
case <-time.After(time.Second):
t.Fatal("maintenance did not stop after task-level cancellation")
}
if got := cache.Activity().MaintenanceActive; got != 0 {
t.Fatalf("maintenance active after cancellation = %d, want 0", got)
}
}
func TestCancelMaintenanceWaitsForCleanupBarrier(t *testing.T) {
t.Parallel()
cache := New(t.TempDir(), testLogger())
const barePath = "/cache/cleanup-barrier.git"
entered := make(chan struct{})
cancelled := make(chan struct{})
releaseCleanup := make(chan struct{})
maintenanceDone := make(chan struct{})
go func() {
_, _ = cache.WithRepoMaintenance(context.Background(), barePath, func(ctx context.Context) error {
close(entered)
<-ctx.Done()
close(cancelled)
<-releaseCleanup
return context.Cause(ctx)
})
close(maintenanceDone)
}()
<-entered
cancelDone := make(chan struct{})
go func() {
cache.CancelMaintenance()
close(cancelDone)
}()
<-cancelled
select {
case <-cancelDone:
t.Fatal("CancelMaintenance returned before maintenance cleanup released the repo")
case <-time.After(20 * time.Millisecond):
}
close(releaseCleanup)
select {
case <-cancelDone:
case <-time.After(time.Second):
t.Fatal("CancelMaintenance did not return after maintenance cleanup completed")
}
<-maintenanceDone
}
func TestCreateWorktreeContextReturnsBusyAfterBoundedLockWait(t *testing.T) {
t.Parallel()
root := t.TempDir()
cache := New(root, testLogger())
const (
workspaceID = "ws-1"
repoURL = "https://github.com/org/repo.git"
)
barePath := cache.BarePath(workspaceID, repoURL)
if err := os.MkdirAll(barePath, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(barePath, "HEAD"), []byte("ref: refs/heads/main\n"), 0o644); err != nil {
t.Fatal(err)
}
lock := cache.lockForRepo(barePath)
lock.Lock()
defer lock.Unlock()
_, err := cache.CreateWorktreeContext(context.Background(), WorktreeParams{
WorkspaceID: workspaceID,
RepoURL: repoURL,
WorkDir: t.TempDir(),
LockWaitTimeout: 20 * time.Millisecond,
})
if !errors.Is(err, ErrRepoBusy) {
t.Fatalf("CreateWorktreeContext error = %v, want ErrRepoBusy", err)
}
}
func TestCreateWorktreeContextCancelsRunningGitProcessTree(t *testing.T) {
binDir := t.TempDir()
marker := filepath.Join(t.TempDir(), "git-descendant-survived")
script := filepath.Join(binDir, "git")
body := "#!/bin/sh\n(sleep 0.5; echo leaked > \"$MULTICA_TEST_GIT_MARKER\") &\nwait\n"
if err := os.WriteFile(script, []byte(body), 0o755); err != nil {
t.Fatal(err)
}
t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
t.Setenv("MULTICA_TEST_GIT_MARKER", marker)
root := t.TempDir()
cache := New(root, testLogger())
const (
workspaceID = "ws-1"
repoURL = "https://github.com/org/repo.git"
)
barePath := cache.BarePath(workspaceID, repoURL)
if err := os.MkdirAll(barePath, 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(barePath, "HEAD"), []byte("ref: refs/heads/main\n"), 0o644); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
started := time.Now()
_, err := cache.CreateWorktreeContext(ctx, WorktreeParams{
WorkspaceID: workspaceID,
RepoURL: repoURL,
WorkDir: t.TempDir(),
})
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("CreateWorktreeContext error = %v, want deadline exceeded", err)
}
if elapsed := time.Since(started); elapsed > time.Second {
t.Fatalf("CreateWorktreeContext took %s after cancellation", elapsed)
}
time.Sleep(600 * time.Millisecond)
if _, err := os.Stat(marker); !os.IsNotExist(err) {
t.Fatalf("Git descendant survived checkout cancellation, stat error = %v", err)
}
}
func TestBareDirName(t *testing.T) {
t.Parallel()
tests := []struct {