diff --git a/cmd/codeaf/chatv3_local_test.go b/cmd/codeaf/chatv3_local_test.go index 6180636bdc..f0f968042b 100644 --- a/cmd/codeaf/chatv3_local_test.go +++ b/cmd/codeaf/chatv3_local_test.go @@ -151,6 +151,11 @@ func TestAPlainLaunchKeepsThisMachinesDoorsWhileAHostLaunchDoesNot(t *testing.T) t.Setenv("CODEAF_PROFILE_DIR", surfaceProfile) t.Setenv("HOME", t.TempDir()) t.Setenv(config.APIKeyEnv, "not-a-real-key") + // hostOptions wires the pool's start-up errands on the terminal's profile; + // the doors exit the process behind them, this test does not, so it closes + // them through the seam the process's own close uses (poolindex.go) before + // the profile is removed. + t.Cleanup(func() { stopPoolErrands(surfaceProfile) }) if err := config.WriteSources(engineProfile, []config.PersistedSource{{ ID: "custom", Written: "engine-service", Address: "https://engine.example/v1", Key: "engine-key", Order: 1, }}); err != nil { diff --git a/cmd/codeaf/chatv3_process.go b/cmd/codeaf/chatv3_process.go index 531515d47d..b0e02e052e 100644 --- a/cmd/codeaf/chatv3_process.go +++ b/cmd/codeaf/chatv3_process.go @@ -386,6 +386,14 @@ func (p *v3Process) closeAll() { return } + // The start-up errands this process seated on its profile — the pool index + // refresh and the outbox push — were started fire-and-forget. + // [stopPoolErrands] cancels them and waits, so nothing this process started + // is still writing under its profile once it closes (poolindex.go states + // the seam). It runs first, before the conversations and the stores, because + // it is the process's own errand and not a conversation's. + stopPoolErrands(p.ProfileDir) + var waiting sync.WaitGroup for _, agent := range agents { waiting.Add(1) diff --git a/cmd/codeaf/poolindex.go b/cmd/codeaf/poolindex.go index 15f433cd51..25a05f2def 100644 --- a/cmd/codeaf/poolindex.go +++ b/cmd/codeaf/poolindex.go @@ -22,6 +22,7 @@ import ( "log" "os" "path/filepath" + "sync" "time" "github.com/Agent-Field/codeaf/internal/config" @@ -72,6 +73,94 @@ func poolIndexFor(profileDir string, cfg poolcfg.Config, now func() time.Time) f // goroutine is started when the build carries no key. var poolRefreshGo = func(scope string, fn func()) { guard.Go(scope, fn) } +// poolErrands is the lifetime of the start-up errands one profile's wiring +// started: the context they are cancelled by and the WaitGroup that answers +// when the last of them has returned. +// +// IT EXISTS BECAUSE A FIRE-AND-FORGET ERRAND OUTLIVES THE PROCESS THAT STARTED +// IT. Both errands write under the profile's pool directory — the refresh its +// doc.json and its signature, the push the outbox and the install nonce — and +// [guard.Go] joins nothing at shutdown, so a process that closed left them +// running against a profile nobody was waiting for. In a test whose profile is +// a temporary directory, that is the directory removed out from under a live +// writer; on a door that reopens on another profile it is a write into a +// directory the process no longer owns. Neither errand is on a run's path, so +// joining them at close costs the person nothing. +type poolErrands struct { + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup +} + +// poolErrandSet holds the errands of each wired profile, so [stopPoolErrands] +// can find the ones a closing process must join. It is keyed by the profile +// directory rather than carried on a process because [wirePoolIndex] is a plain +// function several doors call, and stopping takes the entry out — so a table +// answering a lookup per wiring never grows. +var ( + poolErrandsMu sync.Mutex + poolErrandSet = map[string]*poolErrands{} +) + +// poolErrandsStart seats the tracker for a profile and answers it, so the +// errands wired below register on one context and one WaitGroup. A profile +// wired twice — the host road assembles its options once per launch on the +// same profile — keeps the tracker it has: replacing it would orphan the first +// wiring's errands, which is the leak this tracker exists to close. +func poolErrandsStart(profileDir string) *poolErrands { + poolErrandsMu.Lock() + defer poolErrandsMu.Unlock() + if held := poolErrandSet[profileDir]; held != nil { + return held + } + ctx, cancel := context.WithCancel(context.Background()) + held := &poolErrands{ctx: ctx, cancel: cancel} + poolErrandSet[profileDir] = held + return held +} + +// stopPoolErrands cancels the profile's start-up errands and waits for them to +// return, so that once a process closes nothing it started is still writing +// under the profile. +// +// IT IS THE HALF THE ERRANDS' OWN BUDGETS DO NOT KEEP. A budget bounds one +// fetch, not the life of the goroutine: an unstopped refresh whose relay does +// not answer is still running when the process — or, in a test, the temporary +// profile — is gone, which is a write into a directory nobody owns. It is +// idempotent: a profile with no live errands is a no-op. +func stopPoolErrands(profileDir string) { + poolErrandsMu.Lock() + held := poolErrandSet[profileDir] + delete(poolErrandSet, profileDir) + poolErrandsMu.Unlock() + if held == nil { + return + } + held.cancel() + held.wg.Wait() +} + +// poolErrandGo starts one errand on the profile's tracker, so [stopPoolErrands] +// waits for it, through the same [poolRefreshGo] seam every pool goroutine +// starts through. A profile wired without a tracker — a bare +// [startPoolIndexRefresh] in a test — is the plain seam with nothing to join. +func poolErrandGo(profileDir, scope string, fn func()) { + poolErrandsMu.Lock() + held := poolErrandSet[profileDir] + if held != nil { + held.wg.Add(1) + } + poolErrandsMu.Unlock() + if held == nil { + poolRefreshGo(scope, fn) + return + } + poolRefreshGo(scope, func() { + defer held.wg.Done() + fn() + }) +} + // startPoolIndexRefresh is the loader's own tail: it starts the background // fetch that keeps the cache fresh, and starts NOTHING when the mode forbids // reading or the build carries no public key to check a fetched document under @@ -80,7 +169,7 @@ func startPoolIndexRefresh(ctx context.Context, profileDir string, cfg poolcfg.C if !cfg.CanRead() || len(keys) == 0 { return } - poolRefreshGo("pool/index", func() { refreshPoolIndex(ctx, profileDir, cfg, keys) }) + poolErrandGo(profileDir, "pool/index", func() { refreshPoolIndex(ctx, profileDir, cfg, keys) }) } // refreshPoolIndex fetches a fresh index and lets the puller's own cache keep @@ -142,13 +231,16 @@ func wirePoolIndex(profileDir string) { cfg := config.ModelPoolAt(profileDir) config.AutoIndex = poolIndexFor(profileDir, cfg, time.Now) config.AutoOwnCells = poolOwnCellsFor(profileDir, cfg) - startPoolIndexRefresh(context.Background(), profileDir, cfg, poolTrustedKeys(cfg)) + // The errands below run on this profile's tracker so the process that + // seated them can join them when it closes ([stopPoolErrands]). + held := poolErrandsStart(profileDir) + startPoolIndexRefresh(held.ctx, profileDir, cfg, poolTrustedKeys(cfg)) // The rows a previous run judged and could not hand over leave at once, // on their own goroutine behind the same guard the refresh uses: a // start-up errand, bounded by its own budget, and never on the run's // path. if cfg.CanSend() { - poolRefreshGo("pool/push", func() { poolPush(context.Background(), profileDir, cfg, poolPushBudget) }) + poolErrandGo(profileDir, "pool/push", func() { poolPush(held.ctx, profileDir, cfg, poolPushBudget) }) } } diff --git a/cmd/codeaf/poolleak_test.go b/cmd/codeaf/poolleak_test.go new file mode 100644 index 0000000000..a226747c86 --- /dev/null +++ b/cmd/codeaf/poolleak_test.go @@ -0,0 +1,82 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" +) + +// A LAUNCH'S START-UP ERRANDS BELONG TO THE PROCESS THAT STARTED THEM. +// +// [wirePoolIndex] starts two fire-and-forget errands on every door: the index +// refresh, which writes doc.json and doc.json.sig under the profile's pool +// directory, and the outbox push, which opens the outbox and writes the +// install nonce there. Both run through [guard.Go], which joins nothing at +// shutdown, so a process that closed left them writing into a profile nobody +// was going to wait for. In a test whose profile is a TempDir that is the +// cleanup's own `unlinkat …/pool: directory not empty`; on a door that reopens +// on another profile it is a write into a directory the process no longer +// owns. Neither errand is on a run's path — both are start-up errands — so +// joining them at close costs the person nothing. +// +// THIS IS THE RACE, MADE CERTAIN. The endpoint below holds a fetch open until +// the test releases it, so a refresh that is not stopped is certainly still in +// flight when the close path returns — for the whole of the pull's own budget, +// which is seconds, not the microseconds a scheduling accident needs. The seam +// is read through the same [poolRefreshGo] every pool goroutine starts through, +// so the assertion is about the real errand and not a stand-in: the moment the +// close path has returned, no pool writer may still be running. The wait below +// is a fraction of that budget, so a process that closed without joining them +// fails here rather than passing a second later and racing the profile's own +// clean-up. +func TestTheLaunchesPoolWritersStopWhenTheProcessCloses(t *testing.T) { + released := make(chan struct{}) + var release sync.Once + free := func() { release.Do(func() { close(released) }) } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-released: + case <-r.Context().Done(): + } + http.Error(w, "no index for a test", http.StatusServiceUnavailable) + })) + t.Cleanup(func() { free(); server.Close() }) + // The refresh fetches the index here and, when the primary does not answer, + // falls through to the mirror here as well; the push sends here too. So + // nothing reaches the network and an unstopped refresh is blocked for the + // pull's own budget on each of the two addresses. + t.Setenv("CODEAF_MODEL_POOL_URL", server.URL+"/index.json") + t.Setenv("CODEAF_MODEL_POOL_MIRROR_URL", server.URL+"/mirror.json") + t.Setenv("CODEAF_MODEL_POOL_SUBMIT_URL", server.URL+"/rows") + + var running sync.WaitGroup + prev := poolRefreshGo + poolRefreshGo = func(scope string, fn func()) { + running.Add(1) + go func() { + defer running.Done() + fn() + }() + } + t.Cleanup(func() { poolRefreshGo = prev }) + + proc := v3TestProcess(t) + if _, err := openV3Launch(proc, v3Options{Model: "test/model", Workspace: t.TempDir()}); err != nil { + t.Fatalf("the launch every v3 door assembles through did not open: %v", err) + } + + proc.closeAll() + + done := make(chan struct{}) + go func() { running.Wait(); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("a pool writer was still running after the process closed.\n" + + "The refresh and the push write under the profile's pool directory and nothing joins them at\n" + + "shutdown, so a process that closed left them writing into a profile nobody waits for — the\n" + + "TempDir cleanup's own \"directory not empty\". They must stop when the process closes (poolindex.go).") + } +} diff --git a/docs/changes/unreleased/1124-the-pool-writers-stop-when-the-process-closes.md b/docs/changes/unreleased/1124-the-pool-writers-stop-when-the-process-closes.md new file mode 100644 index 0000000000..75d8bfdca8 --- /dev/null +++ b/docs/changes/unreleased/1124-the-pool-writers-stop-when-the-process-closes.md @@ -0,0 +1,9 @@ +--- +kind: fixed +title: a launch's pool writers stop when the process closes +pr: 1124 +surface: [chat] +invalidates: + - "The Model Pool's two start-up errands — the index refresh that writes `doc.json` beside its signature under the profile's pool directory, and the outbox push that opens the outbox and writes the install nonce there — were started fire-and-forget through `guard.Go`, which joins nothing at shutdown. A process that closed left them writing into a profile nobody was waiting for: a test whose profile is a temporary directory failed its own clean-up with `unlinkat …/pool: directory not empty`, and a door that reopened on another profile could write into a directory it no longer owned. They are now seated on one context and one `sync.WaitGroup` per profile (`cmd/codeaf/poolindex.go`), and `v3Process.closeAll` cancels and waits for them before it closes the conversations and the stores." + - "The errands' own budgets did not keep this. A budget bounds one fetch, not the life of the goroutine, so a relay that did not answer left the refresh running past the close — the failure it caused was a race that passed on the rerun. The join, not the budget, is what makes the profile quiet once the process is gone." +---