pool: a launch's start-up errands stop when the process closes

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 {

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)

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) })
 	}
 }
 

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).")
+	}
+}

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."
+---