diff --git a/cmd/grafel/daemon.go b/cmd/grafel/daemon.go index 1c4513cbf..ff14c907c 100644 --- a/cmd/grafel/daemon.go +++ b/cmd/grafel/daemon.go @@ -397,6 +397,14 @@ func installCapReloadHandler(store *caps.Store, logf interface{ Printf(string, . } else { logf.Printf("cpu-tune: SIGHUP reload — daemon GOMAXPROCS unchanged (=%d, host=%d)", n, runtime.NumCPU()) } + // #5970: the in-process parse gate reads the same cpu.json pin, so + // re-resolve it here too. Without this an operator who tightens + // cpu.json and SIGHUPs would move the daemon's GOMAXPROCS while the + // parse gate — the thing that actually bounds cgo — kept its + // start-up value. Raising wakes queued parses; lowering applies to + // the next acquirer and never preempts a parse already running. + parseCap := installDaemonParseCap(runtime.NumCPU(), store) + logf.Printf("cpu-tune: SIGHUP reload — in-process parse concurrency cap=%d (#5970)", parseCap) } }() } @@ -417,6 +425,83 @@ func envPositiveInt2(name string) int { return n } +// daemonParseCap resolves the cap on CONCURRENT in-process tree-sitter parses +// for a daemon process (`grafel serve` / `grafel engine`), given the host core +// count. Never returns 0 — 0 means "unbounded" to the gate, the opposite of +// what a degenerate core count must produce. +// +// WHY NOT GOMAXPROCS (#5970). Until this change the daemon sized this gate from +// runtime.GOMAXPROCS(0), i.e. 100% of its own runtime parallelism — half the +// box on a default install, the WHOLE box whenever the half-cores default did +// not apply (unknown core count, or an operator raising the pin). That is the +// reactive path: the watcher fires on the user's own `git checkout`/save, the +// daemon re-parses in-process, and the user's machine slows down while they +// type. The standing policy is 25% of machine capacity for background indexing +// (process.IndexCoreBudget), and #5972/#5973 shipped it for the CLI index path +// and the extract-subprocess fanout respectively — both of which fork, and +// neither of which touched this call site. This is the last one, and the only +// one that runs unasked during the user's normal work. +// +// GOMAXPROCS is also not a substitute for this gate in the first place: +// tree-sitter parsing is cgo, and a goroutine inside a cgo call parks in +// _Gsyscall while the runtime hands its P to another goroutine, so N concurrent +// parses occupy N OS threads whatever GOMAXPROCS says. The counting semaphore +// held across the cgo call is the real bound (see internal/indexstate/parsegate.go). +// +// OPERATOR OVERRIDE. The two surfaces that pin the daemon's CPU — the +// GRAFEL_DAEMON_GOMAXPROCS env var and cpu.json's daemon GOMAXPROCS field — +// are operator decisions about how much of the machine this daemon may use, +// and keep top precedence UNCLAMPED in both directions: either may tighten the +// gate below the budget or open it above, up to and including the whole host. +// fileVal is the cpu.json value (0 = unset), so the precedence here is the same +// env > cpu.json > default chain resolveDaemonGOMAXPROCSWith documents; only +// the trailing default differs, and deliberately: the 25% background budget, +// not the half-cores GOMAXPROCS default. That default is a bound on the +// daemon's Go-runtime parallelism (query handling included), not a statement +// about how much background parsing may draw. +func daemonParseCap(hostCPU, fileVal int) int { + if n := envPositiveInt2("GRAFEL_DAEMON_GOMAXPROCS"); n > 0 { + return n + } + if fileVal > 0 { + return fileVal + } + return process.IndexCoreBudgetFor(hostCPU) +} + +// daemonParseCapFileVal reads the cpu.json daemon-GOMAXPROCS pin from the caps +// store, returning 0 when there is no store, no file, or no value. Kept +// separate from daemonParseCap so the resolution policy stays a pure function. +func daemonParseCapFileVal(store *caps.Store) int { + if store == nil { + return 0 + } + cfg, err := store.Load() + if err != nil { + return 0 + } + return cfg.DaemonGOMAXPROCSValue() +} + +// installDaemonParseCap resolves the daemon's parse cap for hostCPU (honouring +// the cpu.json pin in store, if any) and installs it on the process-wide gate, +// returning the value now in force. +// +// Separated from runDaemonMode so the value that actually reaches +// indexstate is assertable in a test — runDaemonMode binds sockets, spawns +// children and blocks forever, so it cannot itself be unit-tested, and the two +// previous attempts at this fix both left a cap that "looked right" on a path +// no test exercised. +// +// The installed cap is the BACKGROUND ceiling. User-awaited in-process parsing +// inside this same process lifts it for its duration via +// indexstate.BeginForegroundParse (see daemonIndexFunc, daemonRebuildFuncCore). +func installDaemonParseCap(hostCPU int, store *caps.Store) int { + cap := daemonParseCap(hostCPU, daemonParseCapFileVal(store)) + indexstate.SetParseConcurrency(cap) + return cap +} + // daemonRunMode selects which of the daemon.Config-driven entrypoints // runDaemonMode hands off to at the end of its shared runtime-tune + // config-assembly prelude (ADR-0024 Phase 1: entrypoint/config carve). @@ -492,19 +577,6 @@ func runDaemonMode(argv []string, runMode daemonRunMode) error { gcLog.Printf("cpu-tune: GRAFEL_DAEMON_GOMAXPROCS=%d applied (was %d, host=%d)", gmp, prev, runtime.NumCPU()) } - // #5630: bound CONCURRENT in-process tree-sitter parses. The reactive - // incremental reindex (and the opt-out in-process full index) re-parse - // changed files INSIDE the daemon — work that escapes both the IndexGate - // (#5493, rebuild-only) and the #5602 reindex GOMAXPROCS cap (subprocess- - // only), so it can monopolise the box while index_status reports idle. Cap - // it at the daemon's effective GOMAXPROCS (the same core budget the daemon - // runtime itself uses) so an in-process parse burst cannot draw more cores - // than the daemon is allowed. Mirrors the runtime cap above; honors the same - // resolve path so a tightened GRAFEL_DAEMON_GOMAXPROCS also tightens parsing. - parseCap := runtime.GOMAXPROCS(0) - indexstate.SetParseConcurrency(parseCap) - gcLog.Printf("cpu-tune: in-process parse concurrency cap=%d (#5630)", parseCap) - // #5675: raise RLIMIT_NOFILE toward the hard limit so a worktree indexing // storm (each subscribed working tree costs ~1 fd per directory on Linux // inotify) cannot exhaust fds and crash the daemon into a KeepAlive/Restart @@ -642,6 +714,17 @@ func runDaemonMode(argv []string, runMode daemonRunMode) error { extract.SetRuntimeCaps(capStore) installCapReloadHandler(capStore, gcLog) + // #5630/#5970: bound CONCURRENT in-process tree-sitter parses at the + // BACKGROUND core budget. See installDaemonParseCap for why GOMAXPROCS was + // the wrong number here, and for the foreground exemption. Installed HERE, + // after capStore exists, so the cpu.json pin is honoured — the same + // env > cpu.json > default precedence every other daemon CPU knob has, and + // re-applied on SIGHUP by installCapReloadHandler. Nothing between process + // start and this point parses, so the gate is in force before any acquirer + // reaches it. + parseCap := installDaemonParseCap(runtime.NumCPU(), capStore) + gcLog.Printf("cpu-tune: in-process parse concurrency cap=%d of %d cores (#5970)", parseCap, runtime.NumCPU()) + // #1626: one-time sweep to relocate any pre-existing in-repo // `.grafel/` graph artifacts into the external store, so groups // that were indexed before this change don't need a full re-index and @@ -1266,6 +1349,14 @@ func daemonSchedulerGroupAlgo(ctx context.Context, group string) error { // RPC argument struct onto the existing in-process Index() entrypoint // defined in this same package. func daemonIndexFunc(args proto.IndexArgs) (string, string, error) { + // #5970 FOREGROUND EXEMPTION. This runs Index() IN-PROCESS, inside the + // daemon, synchronously under an RPC the user's CLI is blocked on — the same + // judgement WithInteractive(true) below already encodes. The daemon's parse + // gate is sized for BACKGROUND work (25% of the box, installDaemonParseCap), + // so without this hold a `grafel index` would be throttled to a quarter of + // the machine with a human waiting on it. Released on every exit path. + defer indexstate.BeginForegroundParse()() + opts := []IndexOption{ WithRepairCandidates(args.Repair), WithRepairApply(args.RepairApply), @@ -1463,6 +1554,32 @@ func daemonRebuildFuncCore( // one-shots, watcher-less daemons, tests) and when GRAFEL_STAGE_GATE=0. defer sched.BargeForeground("rebuild:" + args.Group)() + // #5970 FOREGROUND EXEMPTION, in-process half — ON THE OPT-OUT PATH ONLY. + // The barge and the per-group foreground registry below resolve caps for the + // CHILDREN this rebuild spawns; neither touches the daemon's own + // process-wide parse gate, which is sized for background work. That gate + // binds this function only when indexFn runs IN-PROCESS, i.e. under + // GRAFEL_SUBPROCESS_INDEXER=0. On the shipped default the per-repo index + // forks (runRebuildSubprocess) and the child sizes its own gate from + // --interactive, and daemonForegroundLinks does no tree-sitter parsing at + // all (internal/links has no treesitter dependency) — so a hold here would + // buy this rebuild nothing while suspending the 25% budget process-wide for + // the whole 10-12 minute window. + // + // That asymmetry is the point: BeginForegroundParse lifts the cap for EVERY + // concurrent acquirer, not just this call tree (see its doc). The + // watcher-driven reindex keeps running throughout a rebuild — the barge + // holds heavy write STAGES off, it does not gate index admission — so an + // unconditional hold here would let background parsing run uncapped for the + // duration on the one config where the hold does nothing. Bounded either + // way (the scheduler's worker pool is 2 and TryIncremental extracts + // sequentially), but "suspend the budget for 12 minutes to no benefit" is + // not a trade worth making. A per-caller exemption that leaves the cap + // standing for everyone else is the correct end state; tracked in #6022. + if !sched.SubprocessIndexEnabled() { + defer indexstate.BeginForegroundParse()() + } + // #5954 WALL TIME. The barge above tells the stage gate "hold background // stages off"; this tells every child spawn helper "a human is waiting on // THIS group, resolve foreground caps". They are separate signals on diff --git a/cmd/grafel/daemon_parse_cap_5970_test.go b/cmd/grafel/daemon_parse_cap_5970_test.go new file mode 100644 index 000000000..57a7b7cde --- /dev/null +++ b/cmd/grafel/daemon_parse_cap_5970_test.go @@ -0,0 +1,390 @@ +package main + +import ( + "go/ast" + "go/parser" + "go/printer" + "go/token" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/cajasmota/grafel/internal/daemon/caps" + "github.com/cajasmota/grafel/internal/indexstate" + "github.com/cajasmota/grafel/internal/process" +) + +// daemon_parse_cap_5970_test.go — the daemon's OWN in-process parse gate is +// sized from the background core budget (#5970). +// +// WHY THESE TESTS LOOK LIKE THIS. #5972 and #5973 both landed "the 25% budget" +// and both left cmd/grafel/daemon.go sizing its parse gate from +// runtime.GOMAXPROCS(0) — 100% of the daemon's runtime parallelism, which on a +// default install is half the box and on a GRAFEL_DAEMON_GOMAXPROCS-less, +// unknown-core host is all of it. A test that asserts process.IndexCoreBudget() +// returns 3 on a 12-core machine would have passed throughout. So these tests +// assert two things the helper-level test cannot: +// +// 1. the value that actually reaches indexstate (the gate the parse calls +// block on) when the DAEMON installs it, and +// 2. that runDaemonMode — the one function every `grafel serve` / `grafel +// engine` process runs — is what installs it, and does not compute a cap +// of its own alongside. +// +// (2) is a source-level assertion on purpose. runDaemonMode binds sockets, +// spawns children and blocks forever, so it cannot be called from a unit test; +// without the structural check the behavioural test would pass against a +// daemon that quietly kept its old cap — the exact failure mode this issue is. + +// TestDaemonParseCap_IsTheBackgroundCoreBudget pins the SIZE of the daemon's +// parse cap: the canonical 25%-of-machine background budget, not the daemon's +// GOMAXPROCS. +func TestDaemonParseCap_IsTheBackgroundCoreBudget(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + + for _, hostCPU := range []int{1, 2, 4, 8, 12, 16, 32, 64} { + want := process.IndexCoreBudgetFor(hostCPU) + if got := daemonParseCap(hostCPU, 0); got != want { + t.Errorf("daemonParseCap(hostCPU=%d) = %d, want %d (25%% budget)", hostCPU, got, want) + } + } + // The concrete number the issue is about: a 12-core laptop. + if got := daemonParseCap(12, 0); got != 3 { + t.Errorf("daemonParseCap(12) = %d, want 3", got) + } + // And it must be strictly below the daemon's own runtime parallelism on any + // host big enough for the two to differ — that gap IS the bug. + if got := daemonParseCap(12, 0); got >= 12 { + t.Errorf("daemonParseCap(12) = %d, want < 12 (must not be GOMAXPROCS-shaped)", got) + } +} + +// TestDaemonParseCap_NeverUnbounded guards the floor: 0 means "unbounded" to +// the gate, which is the opposite of what a degenerate core count must produce. +func TestDaemonParseCap_NeverUnbounded(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + for _, hostCPU := range []int{-4, 0, 1} { + if got := daemonParseCap(hostCPU, 0); got < 1 { + t.Errorf("daemonParseCap(%d) = %d, want >= 1", hostCPU, got) + } + } +} + +// TestDaemonParseCap_OperatorOverrideWinsUnclamped pins the escape hatch: an +// explicit GRAFEL_DAEMON_GOMAXPROCS is an operator decision and is honoured +// as-is, in BOTH directions — tighter than the budget and looser than it. +func TestDaemonParseCap_OperatorOverrideWinsUnclamped(t *testing.T) { + cases := []struct { + env string + hostCPU int + want int + }{ + {"2", 12, 2}, // tighter than the 25% budget (3) + {"9", 12, 9}, // looser than the budget, and below host cores + {"12", 12, 12}, // "give the daemon the whole box": never clamped back to 3 + {"0", 12, 3}, // invalid/zero falls through to the budget + {"junk", 12, 3}, + } + for _, c := range cases { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", c.env) + if got := daemonParseCap(c.hostCPU, 0); got != c.want { + t.Errorf("GRAFEL_DAEMON_GOMAXPROCS=%q daemonParseCap(%d) = %d, want %d", + c.env, c.hostCPU, got, c.want) + } + } +} + +// TestInstallDaemonParseCap_ReachesTheGate is the behavioural half of (1): the +// value the daemon computes is the value in force on indexstate's semaphore — +// the thing an actual ts_parser_parse blocks on. +func TestInstallDaemonParseCap_ReachesTheGate(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + t.Cleanup(func() { indexstate.SetParseConcurrency(0) }) + indexstate.SetParseConcurrency(0) + + installed := installDaemonParseCap(12, nil) + if installed != 3 { + t.Fatalf("installDaemonParseCap(12) reported %d, want 3", installed) + } + if got := indexstate.ParseConcurrencyCap(); got != 3 { + t.Fatalf("gate cap after install = %d, want 3", got) + } + if got := indexstate.EffectiveParseConcurrencyCap(); got != 3 { + t.Fatalf("effective gate cap after install = %d, want 3 (background, no foreground hold)", got) + } +} + +// TestRunDaemonMode_InstallsTheBackgroundParseCap is (2): the structural pin. +// It asserts against the AST of runDaemonMode — the single function `grafel +// serve` and `grafel engine` both run — that the cap installed on the daemon +// path is the one this issue fixed, and that no second, GOMAXPROCS-sized +// computation sits beside it. +func TestRunDaemonMode_InstallsTheBackgroundParseCap(t *testing.T) { + fn := funcDeclInDaemonGo(t, "runDaemonMode") + + call := findCall(fn, "installDaemonParseCap") + if call == nil { + t.Fatal("runDaemonMode does not call installDaemonParseCap: the daemon's " + + "in-process parse gate is not sized from the background core budget (#5970)") + } + // A structural "the call exists" check is value-blind: it passes against + // installDaemonParseCap(runtime.NumCPU() * 4), which on a 12-core box + // installs a cap of 12 — the original bug's exact magnitude. Pin the + // argument too: the daemon must pass the HOST core count, unscaled, and let + // daemonParseCap (unit-tested above across core counts) apply the policy. + if len(call.Args) != 2 { + t.Fatalf("installDaemonParseCap called with %d args, want 2 (hostCPU, capStore)", len(call.Args)) + } + if got := exprString(call.Args[0]); got != "runtime.NumCPU()" { + t.Errorf("runDaemonMode passes hostCPU=%s to installDaemonParseCap, want runtime.NumCPU() "+ + "— a scaled or GOMAXPROCS-derived argument re-opens #5970 with the call still in place", got) + } + // The cpu.json pin is an operator override with the same precedence as + // GRAFEL_DAEMON_GOMAXPROCS; passing a nil store here would silently drop it. + if got := exprString(call.Args[1]); got == "nil" { + t.Error("runDaemonMode passes a nil caps store to installDaemonParseCap: " + + "the cpu.json daemon-GOMAXPROCS pin would be silently ignored") + } + // A direct SetParseConcurrency here would be a second, competing cap — the + // shape of the original bug (parseCap := runtime.GOMAXPROCS(0)). + if bodyCallsSelector(fn, "indexstate", "SetParseConcurrency") { + t.Error("runDaemonMode calls indexstate.SetParseConcurrency directly; the parse " + + "cap must be resolved through installDaemonParseCap (#5970)") + } +} + +// TestForegroundInProcessIndexPathsLiftTheCap pins the foreground exemption on +// the two daemon-side entrypoints that run a USER-AWAITED index inside the +// daemon process: the synchronous `grafel index` RPC and the rebuild core (whose +// indexFn runs in-process when GRAFEL_SUBPROCESS_INDEXER=0). Capping those at +// 25% would throttle work a human is sitting and waiting for, which the policy +// explicitly exempts. +func TestForegroundInProcessIndexPathsLiftTheCap(t *testing.T) { + for _, name := range []string{"daemonIndexFunc", "daemonRebuildFuncCore"} { + fn := funcDeclInDaemonGo(t, name) + if !bodyCallsSelector(fn, "indexstate", "BeginForegroundParse") { + t.Errorf("%s does not take a foreground parse hold: user-awaited in-process "+ + "indexing would run under the background 25%% cap (#5970)", name) + } + } +} + +// TestRebuildForegroundHoldIsGatedOnTheInProcessPath is the counterweight to +// the test above. The lift is PROCESS-WIDE: while it is held, background +// parsing is uncapped too (see indexstate.BeginForegroundParse). On the shipped +// default the rebuild forks its per-repo index and its link pass does no +// tree-sitter parsing, so an unconditional hold there would suspend the 25% +// budget for the whole multi-minute rebuild window and buy that rebuild +// nothing — while the watcher-driven reindex, which the barge does not gate, +// keeps parsing. So the hold must be taken only on the in-process opt-out. +func TestRebuildForegroundHoldIsGatedOnTheInProcessPath(t *testing.T) { + fn := funcDeclInDaemonGo(t, "daemonRebuildFuncCore") + + stmt := enclosingIfStmt(fn, func(n ast.Node) bool { + sel, ok := n.(*ast.SelectorExpr) + return ok && sel.Sel.Name == "BeginForegroundParse" + }) + if stmt == nil { + t.Fatal("daemonRebuildFuncCore takes its foreground parse hold unconditionally; " + + "it must be gated on !sched.SubprocessIndexEnabled() so the default (forking) " + + "path does not suspend the background parse budget for nothing (#5970)") + } + cond := exprString(stmt.Cond) + if cond != "!sched.SubprocessIndexEnabled()" { + t.Errorf("rebuild foreground hold is gated on %q, want !sched.SubprocessIndexEnabled()", cond) + } +} + +// TestDaemonParseCap_HonoursCPUJSONPin pins the cpu.json half of the operator +// override. cpu.json is the live-mutable, SIGHUP-reloadable surface for the +// daemon's CPU knobs; a pin there that the parse gate ignores is a +// silently-ineffective operator control — the hardest kind to diagnose. +func TestDaemonParseCap_HonoursCPUJSONPin(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + + // cpu.json pin beats the 25% budget, in both directions. + if got := daemonParseCap(64, 1); got != 1 { + t.Errorf("daemonParseCap(64, cpu.json=1) = %d, want 1 (budget would be 16)", got) + } + if got := daemonParseCap(12, 10); got != 10 { + t.Errorf("daemonParseCap(12, cpu.json=10) = %d, want 10 (never clamped to the budget)", got) + } + // Unset/degenerate cpu.json falls through to the budget. + for _, fileVal := range []int{0, -1} { + if got := daemonParseCap(12, fileVal); got != 3 { + t.Errorf("daemonParseCap(12, cpu.json=%d) = %d, want the budget 3", fileVal, got) + } + } + // env still outranks cpu.json — same precedence resolveDaemonGOMAXPROCSWith + // documents for the GOMAXPROCS knob itself. + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "5") + if got := daemonParseCap(12, 10); got != 5 { + t.Errorf("daemonParseCap with env=5, cpu.json=10 = %d, want 5 (env > cpu.json)", got) + } +} + +// TestInstallDaemonParseCap_ReadsCPUJSONFromStore closes the loop from the file +// on disk to the value on the gate: the pin only counts if installDaemonParseCap +// actually loads it. +func TestInstallDaemonParseCap_ReadsCPUJSONFromStore(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + t.Cleanup(func() { indexstate.SetParseConcurrency(0) }) + indexstate.SetParseConcurrency(0) + + dir := t.TempDir() + path := filepath.Join(dir, "cpu.json") + if err := os.WriteFile(path, []byte(`{"daemon_gomaxprocs": 1}`), 0o644); err != nil { + t.Fatalf("write cpu.json: %v", err) + } + store := caps.NewStore(path) + if cfg, err := store.Load(); err != nil || cfg.DaemonGOMAXPROCSValue() != 1 { + t.Fatalf("caps store did not read the pin back (val=%v err=%v); fixture is wrong, not the code", + func() int { c, _ := store.Load(); return c.DaemonGOMAXPROCSValue() }(), err) + } + + if got := installDaemonParseCap(64, store); got != 1 { + t.Fatalf("installDaemonParseCap(64, cpu.json=1) = %d, want 1", got) + } + if got := indexstate.ParseConcurrencyCap(); got != 1 { + t.Fatalf("gate cap = %d, want 1 (the operator's cpu.json pin)", got) + } +} + +// TestForegroundLiftIsScopedNotGlobal proves the exemption is a scoped hold and +// not a permanent un-capping: after a foreground unit of work completes, the +// daemon's background ceiling is back in force. +func TestForegroundLiftIsScopedNotGlobal(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + t.Cleanup(func() { indexstate.SetParseConcurrency(0) }) + indexstate.SetParseConcurrency(0) + + installDaemonParseCap(12, nil) + release := indexstate.BeginForegroundParse() + if got := indexstate.EffectiveParseConcurrencyCap(); got != 0 { + t.Fatalf("effective cap during foreground work = %d, want 0 (uncapped)", got) + } + release() + if got := indexstate.EffectiveParseConcurrencyCap(); got != 3 { + t.Fatalf("effective cap after foreground work = %d, want the background 3", got) + } +} + +// --- AST helpers ----------------------------------------------------------- + +func funcDeclInDaemonGo(t *testing.T, name string) *ast.FuncDecl { + t.Helper() + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, "daemon.go", nil, 0) + if err != nil { + t.Fatalf("parse daemon.go: %v", err) + } + for _, d := range file.Decls { + if fd, ok := d.(*ast.FuncDecl); ok && fd.Recv == nil && fd.Name.Name == name { + return fd + } + } + t.Fatalf("func %s not found in cmd/grafel/daemon.go", name) + return nil +} + +// findCall returns the first call to the bare (non-selector) function name in +// fn's body, or nil. +func findCall(fn *ast.FuncDecl, name string) *ast.CallExpr { + var found *ast.CallExpr + ast.Inspect(fn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok || found != nil { + return found == nil + } + if id, ok := call.Fun.(*ast.Ident); ok && id.Name == name { + found = call + } + return true + }) + return found +} + +// enclosingIfStmt returns the innermost *ast.IfStmt in fn whose body contains a +// node satisfying match, or nil when no such node exists or it is not inside an +// if at all. Used to assert that a statement is CONDITIONAL, not merely present. +func enclosingIfStmt(fn *ast.FuncDecl, match func(ast.Node) bool) *ast.IfStmt { + var found *ast.IfStmt + ast.Inspect(fn.Body, func(n ast.Node) bool { + ifs, ok := n.(*ast.IfStmt) + if !ok { + return true + } + hit := false + ast.Inspect(ifs.Body, func(inner ast.Node) bool { + if inner != nil && match(inner) { + hit = true + } + return true + }) + // Keep the innermost enclosing if: a later (deeper) candidate starts + // after the one already recorded. + if hit && (found == nil || ifs.Pos() > found.Pos()) { + found = ifs + } + return true + }) + return found +} + +// exprString renders an expression back to source so an argument can be +// asserted on. +func exprString(e ast.Expr) string { + var b strings.Builder + if err := printer.Fprint(&b, token.NewFileSet(), e); err != nil { + return "" + } + return b.String() +} + +// bodyCallsSelector reports whether fn's body contains a call to pkg.name. +func bodyCallsSelector(fn *ast.FuncDecl, pkg, name string) bool { + found := false + ast.Inspect(fn.Body, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + // Match both `pkg.Name(...)` and `defer pkg.Name(...)()`. + fun := call.Fun + if inner, ok := fun.(*ast.CallExpr); ok { + fun = inner.Fun + } + sel, ok := fun.(*ast.SelectorExpr) + if !ok { + return true + } + id, ok := sel.X.(*ast.Ident) + if ok && id.Name == pkg && sel.Sel.Name == name { + found = true + } + return true + }) + return found +} + +// TestDaemonParseCapDocumentsWhyNotGOMAXPROCS is a cheap guard against the +// regression returning by copy-paste: the daemon must not derive its parse cap +// from runtime.GOMAXPROCS again. Checked on the resolved value, using the real +// host, so it fails on any machine where the two differ. +func TestDaemonParseCapDocumentsWhyNotGOMAXPROCS(t *testing.T) { + t.Setenv("GRAFEL_DAEMON_GOMAXPROCS", "") + if runtime.NumCPU() < 8 { + t.Skipf("host has %d cores; budget and GOMAXPROCS can legitimately coincide", runtime.NumCPU()) + } + got := daemonParseCap(runtime.NumCPU(), 0) + if got >= runtime.GOMAXPROCS(0) { + t.Fatalf("daemonParseCap(%d) = %d >= GOMAXPROCS %d: the cap is GOMAXPROCS-shaped again (#5970)", + runtime.NumCPU(), got, runtime.GOMAXPROCS(0)) + } + if got != process.IndexCoreBudget() { + t.Fatalf("daemonParseCap(NumCPU) = %d, want process.IndexCoreBudget() = %d", + got, process.IndexCoreBudget()) + } +} diff --git a/internal/indexstate/parse_foreground_5970_test.go b/internal/indexstate/parse_foreground_5970_test.go new file mode 100644 index 000000000..d92f9e246 --- /dev/null +++ b/internal/indexstate/parse_foreground_5970_test.go @@ -0,0 +1,96 @@ +package indexstate + +import ( + "testing" + "time" +) + +// parse_foreground_5970_test.go — the FOREGROUND exemption on the in-process +// parse gate (#5970). +// +// The daemon installs a BACKGROUND cap at startup (25% of the box). But the +// same process also runs user-awaited in-process indexing: the synchronous +// `grafel index` RPC, and the rebuild path when the subprocess indexer is +// opted out. Those must not be throttled — the standing policy caps background +// work only. Since the gate is a single process-wide semaphore, the exemption +// has to be a scoped, refcounted lift rather than a second cap. + +// TestBeginForegroundParse_LiftsTheCapWhileHeld pins the core contract: while a +// foreground hold is live the gate is unbounded, and the configured background +// cap is restored — not lost — when the hold releases. +func TestBeginForegroundParse_LiftsTheCapWhileHeld(t *testing.T) { + t.Cleanup(func() { SetParseConcurrency(0) }) + SetParseConcurrency(3) + + if got := EffectiveParseConcurrencyCap(); got != 3 { + t.Fatalf("effective cap before hold = %d, want 3", got) + } + release := BeginForegroundParse() + if got := EffectiveParseConcurrencyCap(); got != 0 { + t.Fatalf("effective cap during foreground hold = %d, want 0 (unbounded)", got) + } + // The CONFIGURED cap is untouched: ensureParseConcurrencyDefault and the + // status plane both read it and must still see the daemon's real ceiling. + if got := ParseConcurrencyCap(); got != 3 { + t.Fatalf("configured cap during hold = %d, want 3 (unchanged)", got) + } + release() + if got := EffectiveParseConcurrencyCap(); got != 3 { + t.Fatalf("effective cap after release = %d, want 3 (background cap restored)", got) + } +} + +// TestBeginForegroundParse_Refcounted proves nested/concurrent foreground units +// of work do not un-lift each other: the cap returns only when the LAST hold +// releases, and a double release is inert. +func TestBeginForegroundParse_Refcounted(t *testing.T) { + t.Cleanup(func() { SetParseConcurrency(0) }) + SetParseConcurrency(2) + + r1 := BeginForegroundParse() + r2 := BeginForegroundParse() + r1() + if got := EffectiveParseConcurrencyCap(); got != 0 { + t.Fatalf("effective cap with one hold still live = %d, want 0", got) + } + r1() // idempotent: must not drop r2's hold + if got := EffectiveParseConcurrencyCap(); got != 0 { + t.Fatalf("effective cap after double-release of r1 = %d, want 0", got) + } + r2() + if got := EffectiveParseConcurrencyCap(); got != 2 { + t.Fatalf("effective cap after last release = %d, want 2", got) + } +} + +// TestBeginForegroundParse_UnblocksQueuedWaiters is the behavioural half: a +// parse queued behind a full background gate must proceed the moment foreground +// work lifts the cap, not sit blocked until the in-flight parse finishes. +func TestBeginForegroundParse_UnblocksQueuedWaiters(t *testing.T) { + t.Cleanup(func() { SetParseConcurrency(0) }) + SetParseConcurrency(1) + + AcquireParseSlot() // occupy the single background slot + defer ReleaseParseSlot() + + got := make(chan struct{}) + go func() { + AcquireParseSlot() + close(got) + ReleaseParseSlot() + }() + + select { + case <-got: + t.Fatal("second parse acquired a slot while the cap-1 gate was full") + case <-time.After(50 * time.Millisecond): + } + + release := BeginForegroundParse() + defer release() + select { + case <-got: + case <-time.After(2 * time.Second): + t.Fatal("foreground lift did not wake the queued parse") + } +} diff --git a/internal/indexstate/parsegate.go b/internal/indexstate/parsegate.go index 9bd31fa13..07e375b7e 100644 --- a/internal/indexstate/parsegate.go +++ b/internal/indexstate/parsegate.go @@ -39,6 +39,11 @@ var ( // parseGateWaiters is a FIFO of blocked acquirers, each woken with a slot // already charged to it. parseGateWaiters []chan struct{} + // parseGateForeground counts live FOREGROUND holds (#5970). While it is + // > 0 the gate is unbounded regardless of parseGateCap: the configured cap + // expresses the BACKGROUND core budget, and user-awaited work is exempt + // from it by policy. See BeginForegroundParse. + parseGateForeground int ) // SetParseConcurrency installs the daemon-wide cap on concurrent in-process @@ -57,13 +62,94 @@ func SetParseConcurrency(cap int) { parseGateMu.Unlock() } -// ParseConcurrencyCap returns the configured in-process parse cap (0 = unbounded). +// ParseConcurrencyCap returns the CONFIGURED in-process parse cap +// (0 = unbounded). It deliberately ignores any live foreground hold: callers +// like cmd/grafel's ensureParseConcurrencyDefault use it to answer "has a cap +// already been installed in this process?", and a transient foreground lift +// must not make them re-install one. Use EffectiveParseConcurrencyCap for the +// cap actually in force right now. func ParseConcurrencyCap() int { parseGateMu.Lock() defer parseGateMu.Unlock() return parseGateCap } +// EffectiveParseConcurrencyCap returns the cap acquirers are gated on RIGHT NOW +// (0 = unbounded): the configured cap, or 0 while any foreground hold is live. +func EffectiveParseConcurrencyCap() int { + parseGateMu.Lock() + defer parseGateMu.Unlock() + return effectiveParseCapLocked() +} + +// effectiveParseCapLocked is the one place the foreground exemption is applied. +// MUST hold parseGateMu. +func effectiveParseCapLocked() int { + if parseGateForeground > 0 { + return 0 // unbounded: a human is waiting on this parse + } + return parseGateCap +} + +// BeginForegroundParse marks the start of a USER-AWAITED unit of in-process +// parsing and returns its release closure (#5970). +// +// READ THIS BEFORE ADDING A CALL SITE. The lift is PROCESS-WIDE, not scoped to +// the caller. While any hold is live the gate is unbounded for EVERY acquirer, +// including background parsing that has nothing to do with the work the hold +// was taken for — a watcher-driven incremental reindex running concurrently +// with a held rebuild parses uncapped for the whole window. Because releasing +// does not preempt parses already admitted, an overshoot outlives the release +// until those parses drain. A measured probe reached 20 concurrent parses +// against a background cap of 2. +// +// The consequence for callers: a hold is only justified where the caller +// ITSELF parses in-process, and it should span the narrowest region that does. +// Taking one "just in case", or across a long window in a config where the +// caller forks its parsing to a child, buys nothing and suspends the background +// budget for everyone. The gate has no caller identity to do better; a +// per-caller exemption (acquire that bypasses the semaphore without touching +// parseGateCap) is the correct end state and is tracked in #6022. +// +// WHY THIS EXISTS. The daemon installs its parse cap ONCE at startup, sized +// from the BACKGROUND core budget (25% of the machine) because the work that +// gate was built for — the watcher-driven incremental reindex — is background +// by definition. But the same process also runs parsing a human is sitting and +// waiting for: the synchronous `grafel index` RPC, and the rebuild path's +// indexFn when the subprocess indexer is opted out (GRAFEL_SUBPROCESS_INDEXER=0). +// The standing policy caps background work only; throttling those to a quarter +// of the box would make the user wait for no benefit. Since the gate is a +// single process-wide semaphore, the exemption cannot be a second cap — it has +// to be a scoped lift of the one that exists. +// +// Refcounted, so concurrent foreground units do not un-lift each other, and the +// returned closure is idempotent, so `defer` at the top of a function is safe on +// every exit path. Lifting wakes any parses already queued behind the +// background cap; releasing does not preempt parses already running (same +// contract as SetParseConcurrency). +// +// Scope: this changes how MUCH of the machine in-process parsing may use, never +// whether it runs. It mirrors the foreground/background split the child-spawn +// paths already make (--interactive on the index child, sched's per-group +// foreground registry) for the one path that never forks. +func BeginForegroundParse() (release func()) { + parseGateMu.Lock() + parseGateForeground++ + wakeParseWaitersLocked() + parseGateMu.Unlock() + + var once sync.Once + return func() { + once.Do(func() { + parseGateMu.Lock() + if parseGateForeground > 0 { + parseGateForeground-- + } + parseGateMu.Unlock() + }) + } +} + // AcquireParseSlot blocks until an in-process parse slot is free, then returns. // The caller MUST call ReleaseParseSlot exactly once when the parse completes. // When the gate is unbounded (cap 0) it returns immediately without queueing. @@ -73,7 +159,7 @@ func ParseConcurrencyCap() int { func AcquireParseSlot() { ParseBegin() parseGateMu.Lock() - if parseGateCap <= 0 || parseGateActive < parseGateCap { + if cap := effectiveParseCapLocked(); cap <= 0 || parseGateActive < cap { parseGateActive++ parseGateMu.Unlock() return @@ -99,9 +185,10 @@ func ReleaseParseSlot() { // wakeParseWaitersLocked promotes queued waiters into free slots (FIFO). Each // promoted ticket has a slot charged to it before being signalled. MUST hold -// parseGateMu. With cap 0 (unbounded) every waiter is drained. +// parseGateMu. With an effective cap of 0 (unbounded — no cap configured, or a +// live foreground hold) every waiter is drained. func wakeParseWaitersLocked() { - for len(parseGateWaiters) > 0 && (parseGateCap <= 0 || parseGateActive < parseGateCap) { + for cap := effectiveParseCapLocked(); len(parseGateWaiters) > 0 && (cap <= 0 || parseGateActive < cap); cap = effectiveParseCapLocked() { t := parseGateWaiters[0] parseGateWaiters = parseGateWaiters[1:] parseGateActive++