diff --git a/internal/cli/doctor.go b/internal/cli/doctor.go index c166641fb..403011b51 100644 --- a/internal/cli/doctor.go +++ b/internal/cli/doctor.go @@ -391,13 +391,11 @@ func printIndexingModes(w io.Writer) { // Go soft memory limit (#5237). Report the resolved limit + where it // came from so operators can see whether GOMEMLIMIT / the env override / // the fraction-of-RAM default is in effect. - if mb, src := daemon.MemLimitSummary(); mb > 0 { - fmt.Fprintf(w, "%s go soft mem limit: %dMB (%s; GRAFEL_DAEMON_MEMLIMIT_MB)\n", - statusOK, mb, src) - } else { - fmt.Fprintf(w, "%s go soft mem limit: unbounded (%s; GRAFEL_DAEMON_MEMLIMIT_MB)\n", - statusOK, src) - } + // #6045: report the installation-wide total plus the per-plane shares — + // in split mode two processes divide this budget. + memVal, memSrc := memLimitDescription() + fmt.Fprintf(w, "%s go soft mem limit: %s (%s; GRAFEL_DAEMON_MEMLIMIT_MB)\n", + statusOK, memVal, memSrc) } func checkRepo(w io.Writer, r registry.Repo) { diff --git a/internal/cli/status.go b/internal/cli/status.go index efd3239d5..0e8ce6651 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -122,11 +122,10 @@ func runStatus(w io.Writer, filter string, ref string, showAll bool) error { onOff(ecfg.IsIncrementalEnabled()), onOff(sched.SubprocessIndexEnabled())) // Go soft memory limit (#5237): show the resolved limit + // source so operators can see what's bounding daemon RSS. - if mb, src := daemon.MemLimitSummary(); mb > 0 { - fmt.Fprintf(w, " mem_limit: %dMB (%s)\n", mb, src) - } else { - fmt.Fprintf(w, " mem_limit: unbounded (%s)\n", src) - } + // #6045: in split mode TWO processes share this budget, so the + // line names the total and both per-plane shares. + memVal, memSrc := memLimitDescription() + fmt.Fprintf(w, " mem_limit: %s (%s)\n", memVal, memSrc) } printDaemonDetail(w, st) } @@ -237,6 +236,30 @@ func onOff(b bool) string { return "off" } +// memLimitDescription renders the Go soft memory limit for operator-facing +// surfaces (grafel status / grafel doctor) as a value + a source tag. +// +// #6045: the limit is a budget for the WHOLE INSTALLATION, but in split mode +// two processes run — serve (read plane) and engine (write plane). Reporting a +// bare single figure understated real consumption by 2x, because each process +// used to apply that figure in full. The value therefore names the total AND +// both shares, e.g. +// +// 2560MB (768MB serve + 1792MB engine) +// +// In monolith mode (GRAFEL_SPLIT_MODE=0) there is one process and the value is +// just the total. +func memLimitDescription() (value, source string) { + total, serve, engine, src, split := daemon.MemLimitPlaneSummary() + if total <= 0 { + return "unbounded", src + } + if !split { + return fmt.Sprintf("%dMB", total), src + } + return fmt.Sprintf("%dMB (%dMB serve + %dMB engine)", total, serve, engine), src +} + // humanBytes formats a byte count as a short human-readable string. We // avoid pulling go-humanize for this; the daemon's RSS reporting is the // only consumer. diff --git a/internal/cli/status_memlimit_test.go b/internal/cli/status_memlimit_test.go new file mode 100644 index 000000000..7f9d96adb --- /dev/null +++ b/internal/cli/status_memlimit_test.go @@ -0,0 +1,82 @@ +package cli + +import ( + "regexp" + "strconv" + "strings" + "testing" + + "github.com/cajasmota/grafel/internal/daemon" +) + +// TestMemLimitDescription_SplitReportsBothShares is the reporting half of +// #6045: `grafel status` advertised a single mem_limit while TWO processes +// each applied it, understating real consumption by 2x. The line must now name +// the installation total AND both per-plane shares. +func TestMemLimitDescription_SplitReportsBothShares(t *testing.T) { + t.Setenv("GOMEMLIMIT", "") + t.Setenv("GRAFEL_DAEMON_MEMLIMIT_MB", "10000") + t.Setenv(daemon.SplitModeEnvVar, "1") + + got, src := memLimitDescription() + if src == "" { + t.Error("source tag must not be empty") + } + serve, engine := daemon.SplitMemLimitMB(10000) + for _, want := range []string{ + "10000MB", + strconv.FormatInt(serve, 10) + "MB serve", + strconv.FormatInt(engine, 10) + "MB engine", + } { + if !strings.Contains(got, want) { + t.Errorf("mem_limit description %q missing %q", got, want) + } + } + // Reporting only one share is the defect: both must appear, and they must + // be different numbers than the total. + if strings.Count(got, "MB") < 3 { + t.Errorf("mem_limit description %q should carry the total plus BOTH shares", got) + } + // The printed shares must ADD UP to the printed total — that is the whole + // claim the line makes. A line whose shares each equal the total is the + // 2x-understating bug wearing a nicer format. + nums := memLimitNumbersRe.FindAllStringSubmatch(got, -1) + if len(nums) != 3 { + t.Fatalf("mem_limit description %q: want exactly 3 MB figures (total, serve, engine), got %d", got, len(nums)) + } + parse := func(s string) int64 { v, _ := strconv.ParseInt(s, 10, 64); return v } + gotTotal, gotServe, gotEngine := parse(nums[0][1]), parse(nums[1][1]), parse(nums[2][1]) + if gotServe+gotEngine != gotTotal { + t.Errorf("mem_limit description %q: printed shares %d+%d=%d do not sum to the printed total %d", + got, gotServe, gotEngine, gotServe+gotEngine, gotTotal) + } +} + +var memLimitNumbersRe = regexp.MustCompile(`(\d+)MB`) + +// TestMemLimitDescription_MonolithReportsWholeBudget: in monolith mode there is +// one process, so the line must report the whole budget and must NOT claim a +// per-plane split. +func TestMemLimitDescription_MonolithReportsWholeBudget(t *testing.T) { + t.Setenv("GOMEMLIMIT", "") + t.Setenv("GRAFEL_DAEMON_MEMLIMIT_MB", "10000") + t.Setenv(daemon.SplitModeEnvVar, "0") + + got, _ := memLimitDescription() + if !strings.Contains(got, "10000MB") { + t.Errorf("monolith mem_limit description %q must report the whole 10000MB budget", got) + } + if strings.Contains(got, "serve") || strings.Contains(got, "engine") { + t.Errorf("monolith mem_limit description %q must not advertise a per-plane split", got) + } +} + +// TestMemLimitDescription_Unbounded keeps the disabled path readable. +func TestMemLimitDescription_Unbounded(t *testing.T) { + t.Setenv("GOMEMLIMIT", "") + t.Setenv("GRAFEL_DAEMON_MEMLIMIT_MB", "off") + got, _ := memLimitDescription() + if !strings.Contains(got, "unbounded") { + t.Errorf("disabled limit: description %q should say unbounded", got) + } +} diff --git a/internal/daemon/memlimit.go b/internal/daemon/memlimit.go index 76dda3b92..10cb7168d 100644 --- a/internal/daemon/memlimit.go +++ b/internal/daemon/memlimit.go @@ -49,6 +49,109 @@ const memLimitCeilingMB int64 = 2560 // untouched by either clamp. const memLimitFloorMB int64 = 2048 +// MemLimitServeShare is the fraction of the INSTALLATION-WIDE soft memory +// limit granted to the serve (read) plane when the daemon runs split +// (ADR-0024 default). The engine (write) plane gets the remainder. +// +// Why 30/70 rather than an even split (#6045): +// +// - The engine is the write plane — scheduler, watcher, extraction, +// fbwriter — and is where the heavy allocation happens. A legitimate large +// reindex peaks at ~1-1.5GB Go heap per job (measured, see +// memLimitCeilingMB), so an even split of the 2560MB default ceiling +// (1280MB) would sit BELOW the engine's normal working set and make the +// runtime GC continuously against a limit it cannot honour. +// - serve's large data structure — the graph_cache mmap — is file-backed and +// NOT Go-heap, so GOMEMLIMIT does not account for it at all. What serve +// actually allocates on the heap is MCP request/response marshalling and +// dashboard JSON: small, bursty, short-lived. +// +// 0.30 of the 2560MB default gives serve 768MB and the engine 1792MB, which +// clears the measured per-job peak with headroom while keeping the pair's TOTAL +// equal to the single advertised figure. Both limits remain SOFT: a plane that +// briefly exceeds its share GCs harder, it is not OOM-killed. +const MemLimitServeShare = 0.30 + +// SplitMemLimitMB divides an installation-wide soft limit (MB) into the serve +// and engine shares. The two shares sum to EXACTLY totalMB — the engine +// absorbs the rounding — which is the whole point of #6045: the pair must not +// be able to exceed the number `grafel status` advertises. +// +// A non-positive totalMB means "limit disabled" and propagates as-is to both +// shares; splitting must never synthesize a limit where there was none. +func SplitMemLimitMB(totalMB int64) (serveMB, engineMB int64) { + if totalMB <= 0 { + return totalMB, totalMB + } + serveMB = int64(float64(totalMB) * MemLimitServeShare) + if serveMB < 1 { + serveMB = 1 // never hand a plane a zero limit + } + if serveMB >= totalMB { + serveMB = totalMB - 1 + } + return serveMB, totalMB - serveMB +} + +// memPlane identifies which process is applying the soft memory limit, and +// therefore which share of the installation budget it gets. +// +// The share is derived from the plane the process is ACTUALLY running, not +// from an env var handed down by serve. That is deliberate: the engine child +// inherits serve's environment verbatim and defaultEngineChildCommand must not +// mutate it (see its doc comment — synthesizing env for the child is exactly +// what broke the store-root invariant once already). Both processes see the +// same env, the same settings.json and the same host RAM, so both resolve the +// identical TOTAL independently and then take their own slice of it. No +// channel, nothing to keep in sync, and a standalone `grafel engine` started +// by hand still stays inside the installation budget. +type memPlane int + +const ( + // memPlaneMonolith is the single-process daemon (escape hatch + // GRAFEL_SPLIT_MODE=0): one process, so it gets the WHOLE budget. + memPlaneMonolith memPlane = iota + // memPlaneServe is the split-mode serve (read) plane. + memPlaneServe + // memPlaneEngine is the split-mode engine (write) plane. + memPlaneEngine +) + +func (p memPlane) String() string { + switch p { + case memPlaneServe: + return "serve" + case memPlaneEngine: + return "engine" + default: + return "monolith" + } +} + +// shareOf returns this plane's slice of an installation-wide limit (MB). +func (p memPlane) shareOf(totalMB int64) int64 { + if totalMB <= 0 { + return totalMB + } + serve, engine := SplitMemLimitMB(totalMB) + switch p { + case memPlaneServe: + return serve + case memPlaneEngine: + return engine + default: + return totalMB + } +} + +// memPlaneForDaemonPlane maps the run() plane selector onto the memory plane. +func memPlaneForDaemonPlane(plane daemonPlaneMode) memPlane { + if plane == planeServeOnly { + return memPlaneServe + } + return memPlaneMonolith +} + // applyMemoryLimit sets a conservative Go soft memory limit (GOMEMLIMIT) so // the runtime collects more aggressively as it nears the cap, bounding the // daemon's peak footprint (#3648, tightened in #5237). @@ -65,27 +168,40 @@ const memLimitFloorMB int64 = 2048 // This is intentionally a soft limit: Go will exceed it transiently rather // than OOM-killing the indexer, it just GCs harder — so a legitimate heavy // reindex still completes, it simply doesn't get to retain a multi-GB arena. -func applyMemoryLimit(logger *slog.Logger) { +// +// The resolved value is the budget for the WHOLE INSTALLATION, not for one +// process (#6045). In split mode two processes start — serve and engine — and +// each used to apply the full resolved limit, so the real ceiling was 2x the +// figure `grafel status` advertised. Each process now applies only its plane's +// share (see memPlane / SplitMemLimitMB); the monolith, being one process, +// still gets the whole thing. +func applyMemoryLimit(logger *slog.Logger, plane memPlane) { if logger == nil { logger = slog.Default() } // Respect an explicit GOMEMLIMIT — the runtime already applied it at // startup; re-setting from here would clobber the operator's choice. + // NOTE: this path is genuinely per-process — the runtime read the env var + // before main() in BOTH planes and we cannot retroactively split it. That + // is the operator's explicit choice, and the log names the plane so the + // doubling is at least visible. if v := os.Getenv("GOMEMLIMIT"); v != "" && v != "off" { - logger.Info("daemon: GOMEMLIMIT already set by runtime env; not overriding (#3648)", "gomemlimit", v) + logger.Info("daemon: GOMEMLIMIT already set by runtime env; not overriding (#3648)", + "gomemlimit", v, "plane", plane.String()) return } - limitMB, source := resolveMemLimitMB() - if limitMB <= 0 { - logger.Info("daemon: Go soft memory limit disabled (#3648)", "source", source) + totalMB, source := resolveMemLimitMB() + if totalMB <= 0 { + logger.Info("daemon: Go soft memory limit disabled (#3648)", "source", source, "plane", plane.String()) return } + limitMB := plane.shareOf(totalMB) limitBytes := limitMB * 1024 * 1024 debug.SetMemoryLimit(limitBytes) - logger.Info("daemon: applied Go soft memory limit (#3648)", - "limit_mb", limitMB, "source", source) + logger.Info("daemon: applied Go soft memory limit (#3648, split #6045)", + "limit_mb", limitMB, "total_mb", totalMB, "plane", plane.String(), "source", source) } // resolveMemLimitMB returns the soft-limit in MB and a short tag describing @@ -152,3 +268,29 @@ func MemLimitSummary() (mb int64, source string) { } return resolveMemLimitMB() } + +// MemLimitPlaneSummary reports the INSTALLATION-WIDE soft memory limit and how +// it is divided across the running processes, for operator-facing surfaces +// (grafel status / doctor). Added for #6045, where status advertised a single +// figure that two processes each applied in full. +// +// - totalMB is the whole-installation budget (<=0 means disabled/unbounded). +// - split reports whether the daemon runs the serve/engine process split. +// - When split, serveMB+engineMB == totalMB exactly. +// - When NOT split (monolith escape hatch GRAFEL_SPLIT_MODE=0) there is one +// process: serveMB == totalMB and engineMB == 0. +// +// Like MemLimitSummary this reads the local process env, which a co-located +// daemon shares. +func MemLimitPlaneSummary() (totalMB, serveMB, engineMB int64, source string, split bool) { + totalMB, source = MemLimitSummary() + split = SplitModeEnabled() + if totalMB <= 0 { + return totalMB, 0, 0, source, split + } + if !split { + return totalMB, totalMB, 0, source, false + } + serveMB, engineMB = SplitMemLimitMB(totalMB) + return totalMB, serveMB, engineMB, source, true +} diff --git a/internal/daemon/memlimit_plane_e2e_test.go b/internal/daemon/memlimit_plane_e2e_test.go new file mode 100644 index 000000000..ca19e8ca3 --- /dev/null +++ b/internal/daemon/memlimit_plane_e2e_test.go @@ -0,0 +1,355 @@ +//go:build darwin || linux + +package daemon_test + +import ( + "context" + "log/slog" + "math" + "os" + "os/exec" + "regexp" + "runtime/debug" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/cajasmota/grafel/internal/daemon" + "github.com/cajasmota/grafel/internal/daemon/proto" +) + +// memLimitTotalForPlaneTests is a deterministic installation-wide budget +// pinned via GRAFEL_DAEMON_MEMLIMIT_MB so these tests do not depend on the +// host's RAM. +const memLimitTotalForPlaneTests = 10000 + +// pinProcessMemLimit saves + restores the process-global Go soft memory limit +// and RESETS it to the Go default for the duration of the test. +// +// The reset matters: sibling tests in this package start in-process daemons, +// which call applyMemoryLimit and leave the limit set. Without the reset, +// waitForRuntimeMemLimit would return that stale value immediately and the +// test would pass or fail on someone else's number — the whole suite would be +// a fixture that cannot exhibit what it claims to check. +func pinProcessMemLimit(t *testing.T) { + t.Helper() + prev := debug.SetMemoryLimit(math.MaxInt64) + t.Cleanup(func() { debug.SetMemoryLimit(prev) }) +} + +// waitForRuntimeMemLimit polls until the process-global soft limit moves off +// the Go default (math.MaxInt64), i.e. until the daemon under test has run +// applyMemoryLimit. Returns the limit in MB, or -1 on timeout. +func waitForRuntimeMemLimit(timeout time.Duration) int64 { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if v := debug.SetMemoryLimit(-1); v != math.MaxInt64 { + return v / 1024 / 1024 + } + time.Sleep(10 * time.Millisecond) + } + return -1 +} + +// TestRunServe_AppliesServeShareNotWholeBudget binds the fix to the path that +// actually runs (#6045). With split mode ON, RunServe's in-process serve plane +// must set GOMEMLIMIT to its SHARE of the installation budget — not the whole +// advertised figure, which is what made the real ceiling 2x what +// `grafel status` reported. +func TestRunServe_AppliesServeShareNotWholeBudget(t *testing.T) { + pinProcessMemLimit(t) + root := shortTempRoot(t) + t.Setenv(daemon.EnvRoot, root) + t.Setenv("GRAFEL_HOME", root) + t.Setenv(daemon.EnvDisableSelfDefense, "1") + t.Setenv("GRAFEL_SPLIT_MODE", "1") + t.Setenv(daemon.EnvStatusHeartbeatSeconds, "1") + t.Setenv("GOMEMLIMIT", "") + t.Setenv("GRAFEL_DAEMON_MEMLIMIT_MB", strconv.Itoa(memLimitTotalForPlaneTests)) + + // Substitute the in-binary engine helper for a real grafel binary. + restore := daemon.SetEngineChildCommandForTest(func(selfExe, root string) *exec.Cmd { + cmd := exec.Command(selfExe, "-test.run=TestEngineChildHelper", "-test.timeout=120s") + cmd.Env = append(os.Environ(), + "GRAFEL_ENGINE_CHILD_HELPER=1", + daemon.EnvRoot+"="+root, + ) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + return cmd + }) + defer restore() + + layout, err := daemon.DefaultLayout() + if err != nil { + t.Fatalf("layout: %v", err) + } + if err := daemon.EnsureLayout(layout); err != nil { + t.Fatalf("ensure: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- daemon.RunServe(ctx, daemon.ServeConfig{Config: daemon.Config{ + Layout: layout, + Index: func(a proto.IndexArgs) (string, string, error) { + return a.RepoPath + "/.grafel/graph.json", `{"ok":true}`, nil + }, + }}) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(15 * time.Second): + t.Log("RunServe did not exit within 15s") + } + }) + + gotMB := waitForRuntimeMemLimit(20 * time.Second) + if gotMB < 0 { + t.Fatal("serve never applied a Go soft memory limit") + } + wantServe, _ := daemon.SplitMemLimitMB(memLimitTotalForPlaneTests) + if gotMB != wantServe { + t.Errorf("serve plane applied %dMB, want its share %dMB of the %dMB installation budget", + gotMB, wantServe, memLimitTotalForPlaneTests) + } + if gotMB == memLimitTotalForPlaneTests { + t.Errorf("serve plane took the WHOLE %dMB budget — this is the #6045 per-process doubling", + memLimitTotalForPlaneTests) + } +} + +// TestRunServe_MonolithGetsWholeBudget: with the GRAFEL_SPLIT_MODE=0 escape +// hatch there is exactly ONE process, so it must keep the whole installation +// budget. Halving it here would be a silent regression for every operator on +// the documented escape hatch. +func TestRunServe_MonolithGetsWholeBudget(t *testing.T) { + pinProcessMemLimit(t) + root := shortTempRoot(t) + t.Setenv(daemon.EnvRoot, root) + t.Setenv("GRAFEL_HOME", root) + t.Setenv(daemon.EnvDisableSelfDefense, "1") + t.Setenv("GRAFEL_SPLIT_MODE", "0") + t.Setenv("GOMEMLIMIT", "") + t.Setenv("GRAFEL_DAEMON_MEMLIMIT_MB", strconv.Itoa(memLimitTotalForPlaneTests)) + + layout, err := daemon.DefaultLayout() + if err != nil { + t.Fatalf("layout: %v", err) + } + if err := daemon.EnsureLayout(layout); err != nil { + t.Fatalf("ensure: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- daemon.RunServe(ctx, daemon.ServeConfig{Config: daemon.Config{ + Layout: layout, + Index: func(a proto.IndexArgs) (string, string, error) { + return a.RepoPath + "/.grafel/graph.json", `{"ok":true}`, nil + }, + }}) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(15 * time.Second): + t.Log("RunServe (monolith) did not exit within 15s") + } + }) + + gotMB := waitForRuntimeMemLimit(20 * time.Second) + if gotMB < 0 { + t.Fatal("monolith serve never applied a Go soft memory limit") + } + if gotMB != memLimitTotalForPlaneTests { + t.Errorf("monolith applied %dMB, want the whole %dMB budget (one process, no split)", + gotMB, memLimitTotalForPlaneTests) + } +} + +// syncBuf is a concurrency-safe io.Writer for capturing daemon log output. +type syncBuf struct { + mu sync.Mutex + buf strings.Builder +} + +func (s *syncBuf) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncBuf) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +// runServeCapturingLog starts RunServe with a captured logger and a NON-ZERO +// RSS admission-control budget, waits for the socket, and returns the log text. +func runServeCapturingLog(t *testing.T, splitMode string) string { + t.Helper() + root := shortTempRoot(t) + t.Setenv(daemon.EnvRoot, root) + t.Setenv("GRAFEL_HOME", root) + t.Setenv(daemon.EnvDisableSelfDefense, "1") + t.Setenv("GRAFEL_SPLIT_MODE", splitMode) + t.Setenv(daemon.EnvStatusHeartbeatSeconds, "1") + + if splitMode != "0" { + restore := daemon.SetEngineChildCommandForTest(func(selfExe, root string) *exec.Cmd { + cmd := exec.Command(selfExe, "-test.run=TestEngineChildHelper", "-test.timeout=120s") + cmd.Env = append(os.Environ(), + "GRAFEL_ENGINE_CHILD_HELPER=1", + daemon.EnvRoot+"="+root, + ) + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + return cmd + }) + t.Cleanup(restore) + } + + layout, err := daemon.DefaultLayout() + if err != nil { + t.Fatalf("layout: %v", err) + } + if err := daemon.EnsureLayout(layout); err != nil { + t.Fatalf("ensure: %v", err) + } + + sb := &syncBuf{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- daemon.RunServe(ctx, daemon.ServeConfig{Config: daemon.Config{ + Layout: layout, + Logger: slog.New(slog.NewTextHandler(sb, nil)), + // A NON-ZERO budget is what makes this fixture able to exhibit the + // thing it checks: with 0 the admission-control line is never + // logged by any plane and the assertion would be vacuous. + MaxRSSBudgetMB: 2048, + // The scheduler (and therefore the admission-control line) only + // comes up when SchedulerIndex is wired. + SchedulerIndex: func(ctx context.Context, repo, ref string) error { return nil }, + Index: func(a proto.IndexArgs) (string, string, error) { + return a.RepoPath + "/.grafel/graph.json", `{"ok":true}`, nil + }, + }}) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(15 * time.Second): + t.Log("RunServe did not exit within 15s") + } + }) + waitDaemonReady(t, layout.SocketPath, 20*time.Second) + return sb.String() +} + +// TestRSSBudgetAdmissionControl_IsEnginePlaneOnly is the companion check to the +// GOMEMLIMIT split (#6045 item 3). The reporter saw +// `scheduler: RSS-budget admission control enabled budget_mb=2048` and assumed +// it was doubled the same way. It is NOT: the budget belongs to the scheduler, +// which lives in the engine plane, and split-mode serve skips the engine plane +// entirely — so exactly ONE process ever arms it. This test pins that so a +// future change that starts the engine plane inside serve cannot silently +// reintroduce the doubling for the RSS budget. +func TestRSSBudgetAdmissionControl_IsEnginePlaneOnly(t *testing.T) { + const marker = "RSS-budget admission control enabled" + + // Monolith: exactly one process, and it MUST arm the budget. (This half is + // what proves the fixture can exhibit the marker at all.) + if logged := runServeCapturingLog(t, "0"); !strings.Contains(logged, marker) { + t.Fatalf("monolith: expected %q in serve's log (fixture cannot exhibit the marker)\n%s", marker, logged) + } + + // Split: serve must NOT arm it — the engine child owns the scheduler. + logged := runServeCapturingLog(t, "1") + if strings.Contains(logged, marker) { + t.Errorf("split-mode serve armed the RSS budget too — that is a second per-process doubling\n%s", logged) + } +} + +// TestEngineMemLimitHelper is the subprocess entrypoint for +// TestRunEngine_AppliesEngineShare. It runs the REAL daemon.RunEngine, waits +// for the runtime soft limit to move off the Go default, prints it, and exits. +// Inert unless GRAFEL_ENGINE_MEMLIMIT_HELPER=1. +func TestEngineMemLimitHelper(t *testing.T) { + if os.Getenv("GRAFEL_ENGINE_MEMLIMIT_HELPER") != "1" { + return + } + layout, err := daemon.DefaultLayout() + if err != nil { + t.Fatalf("helper layout: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + _ = daemon.RunEngine(ctx, daemon.EngineConfig{Config: daemon.Config{Layout: layout}}) + }() + mb := waitForRuntimeMemLimit(30 * time.Second) + // Deliberately os.Exit: RunEngine blocks until SIGTERM and we only need + // the limit it applied. + os.Stdout.WriteString("ENGINE_MEMLIMIT_MB=" + strconv.FormatInt(mb, 10) + "\n") + os.Exit(0) +} + +var engineMemLimitRe = regexp.MustCompile(`ENGINE_MEMLIMIT_MB=(-?\d+)`) + +// TestRunEngine_AppliesEngineShare binds the engine half of the fix to the +// path that actually runs: a real daemon.RunEngine in a separate process must +// apply its SHARE, not the whole installation budget. +func TestRunEngine_AppliesEngineShare(t *testing.T) { + root := shortTempRoot(t) + exe, err := os.Executable() + if err != nil { + t.Fatalf("self exe: %v", err) + } + cmd := exec.Command(exe, "-test.run=TestEngineMemLimitHelper", "-test.timeout=90s") + cmd.Env = append(os.Environ(), + "GRAFEL_ENGINE_MEMLIMIT_HELPER=1", + "GRAFEL_ENGINE_CHILD_HELPER=", + daemon.EnvRoot+"="+root, + "GRAFEL_HOME="+root, + daemon.EnvDisableSelfDefense+"=1", + "GOMEMLIMIT=", + "GRAFEL_DAEMON_MEMLIMIT_MB="+strconv.Itoa(memLimitTotalForPlaneTests), + ) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("engine helper subprocess: %v\n%s", err, out) + } + m := engineMemLimitRe.FindStringSubmatch(string(out)) + if m == nil { + t.Fatalf("helper did not report a limit\n%s", out) + } + gotMB, _ := strconv.ParseInt(m[1], 10, 64) + _, wantEngine := daemon.SplitMemLimitMB(memLimitTotalForPlaneTests) + if gotMB != wantEngine { + t.Errorf("engine plane applied %dMB, want its share %dMB of the %dMB installation budget", + gotMB, wantEngine, memLimitTotalForPlaneTests) + } + if gotMB == memLimitTotalForPlaneTests { + t.Errorf("engine plane took the WHOLE %dMB budget — this is the #6045 per-process doubling", + memLimitTotalForPlaneTests) + } + // The log line the reporter saw must now carry the plane + the total. + logged := string(out) + for _, want := range []string{"plane=engine", "total_mb=" + strconv.Itoa(memLimitTotalForPlaneTests)} { + if !strings.Contains(logged, want) { + t.Errorf("engine soft-limit log line missing %q\n%s", want, logged) + } + } +} diff --git a/internal/daemon/memlimit_split_test.go b/internal/daemon/memlimit_split_test.go new file mode 100644 index 000000000..9c8fb3e09 --- /dev/null +++ b/internal/daemon/memlimit_split_test.go @@ -0,0 +1,206 @@ +package daemon + +import ( + "bytes" + "log/slog" + "runtime/debug" + "strconv" + "strings" + "testing" +) + +// pinRuntimeMemLimit saves and restores the process-global Go soft memory +// limit so a test that calls applyMemoryLimit cannot leak a tight (or lax) +// limit into sibling tests. +func pinRuntimeMemLimit(t *testing.T) { + t.Helper() + prev := debug.SetMemoryLimit(-1) // -1 reads without changing + debug.SetMemoryLimit(prev) + t.Cleanup(func() { debug.SetMemoryLimit(prev) }) +} + +// TestSplitMemLimitMB_SharesSumToTotal is the core arithmetic of #6045: the +// installation gets ONE budget, split across the two processes. serve+engine +// must equal the advertised total EXACTLY (engine absorbs the rounding), and +// neither plane may be handed the whole thing. +// +// This is the test that fails if the split is reverted to per-process-full. +func TestSplitMemLimitMB_SharesSumToTotal(t *testing.T) { + // 2MB is a degenerate lower bound (the real floor is memLimitFloorMB); + // it is here only to prove the arithmetic never hands one plane everything. + for _, total := range []int64{2, 2048, 2560, 3000, 4096, 10000} { + serve, engine := SplitMemLimitMB(total) + if serve+engine != total { + t.Errorf("total=%d: serve(%d)+engine(%d)=%d, want exactly %d", + total, serve, engine, serve+engine, total) + } + if serve >= total { + t.Errorf("total=%d: serve share %d must be strictly less than the total (per-process-full is the bug)", total, serve) + } + if engine >= total { + t.Errorf("total=%d: engine share %d must be strictly less than the total (per-process-full is the bug)", total, engine) + } + if serve < 0 || engine < 0 { + t.Errorf("total=%d: negative share serve=%d engine=%d", total, serve, engine) + } + } +} + +// TestSplitMemLimitMB_EngineGetsMajority pins the RATIO, not just the sum: +// the engine is the write plane (scheduler/extraction/fbwriter) and does the +// heavy allocation, while serve's large data (graph_cache mmap) is off-heap +// and therefore not counted by GOMEMLIMIT at all. A 50/50 split would starve +// the engine, whose measured per-job peak heap is 1-1.5GB. +func TestSplitMemLimitMB_EngineGetsMajority(t *testing.T) { + if MemLimitServeShare >= 0.5 { + t.Fatalf("MemLimitServeShare=%v: the engine (write plane) must get the majority", MemLimitServeShare) + } + const total = int64(2560) + serve, engine := SplitMemLimitMB(total) + if engine <= serve { + t.Fatalf("total=%d: engine share %d must exceed serve share %d", total, engine, serve) + } + wantServe := int64(float64(total) * MemLimitServeShare) + if serve != wantServe { + t.Errorf("serve share = %d, want %d (%.0f%% of %d)", serve, wantServe, MemLimitServeShare*100, total) + } + // The engine must still clear the measured 1.5GB per-job peak on the + // default 2560MB ceiling. + if engine < 1536 { + t.Errorf("engine share %dMB is below the measured 1536MB per-job peak heap", engine) + } +} + +// TestSplitMemLimitMB_DisabledStaysDisabled: a non-positive total means the +// daemon-applied limit is off; splitting must not synthesize one. +func TestSplitMemLimitMB_DisabledStaysDisabled(t *testing.T) { + for _, total := range []int64{0, -1} { + serve, engine := SplitMemLimitMB(total) + if serve > 0 || engine > 0 { + t.Errorf("total=%d: want both shares <=0, got serve=%d engine=%d", total, serve, engine) + } + } +} + +// TestMemPlaneShareOf_MonolithGetsWholeBudget: in monolith mode +// (GRAFEL_SPLIT_MODE=0) there is exactly ONE process, so it must get the whole +// installation budget. This is the test that fails if the split is applied +// unconditionally and monolith is accidentally halved. +func TestMemPlaneShareOf_MonolithGetsWholeBudget(t *testing.T) { + const total = int64(2560) + if got := memPlaneMonolith.shareOf(total); got != total { + t.Errorf("monolith share = %d, want the whole budget %d", got, total) + } + serve, engine := SplitMemLimitMB(total) + if got := memPlaneServe.shareOf(total); got != serve { + t.Errorf("serve plane share = %d, want %d", got, serve) + } + if got := memPlaneEngine.shareOf(total); got != engine { + t.Errorf("engine plane share = %d, want %d", got, engine) + } +} + +// applyAndCapture runs applyMemoryLimit for a plane against a JSON logger and +// returns the resulting runtime limit (bytes) plus the captured log text. +func applyAndCapture(t *testing.T, plane memPlane) (int64, string) { + t.Helper() + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, nil)) + applyMemoryLimit(logger, plane) + return debug.SetMemoryLimit(-1), buf.String() +} + +// TestApplyMemoryLimit_PerPlaneShare exercises the REAL applyMemoryLimit and +// asserts each plane sets the runtime limit to its SHARE, not the total — and +// that the log line says so. #6045 is a bug visible in the log, so the log +// content is part of the contract. +func TestApplyMemoryLimit_PerPlaneShare(t *testing.T) { + pinRuntimeMemLimit(t) + t.Setenv("GOMEMLIMIT", "") + t.Setenv(memLimitEnv, "10000") // deterministic total, no host-RAM dependency + + serveMB, engineMB := SplitMemLimitMB(10000) + + cases := []struct { + name string + plane memPlane + wantMB int64 + }{ + {"serve", memPlaneServe, serveMB}, + {"engine", memPlaneEngine, engineMB}, + {"monolith", memPlaneMonolith, 10000}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + gotBytes, logged := applyAndCapture(t, tc.plane) + if want := tc.wantMB * 1024 * 1024; gotBytes != want { + t.Errorf("runtime limit = %d bytes (%dMB), want %d bytes (%dMB)", + gotBytes, gotBytes/1024/1024, want, tc.wantMB) + } + // The log must report the applied share, the installation total, + // and which plane applied it. + for _, want := range []string{ + `"limit_mb":` + strconv.FormatInt(tc.wantMB, 10), + `"total_mb":10000`, + `"plane":"` + tc.plane.String() + `"`, + } { + if !strings.Contains(logged, want) { + t.Errorf("log line missing %s\ngot: %s", want, logged) + } + } + }) + } +} + +// TestApplyMemoryLimit_DisabledPerPlane: with the limit disabled, no plane may +// set a limit (and the disabled log line must not claim one was applied). +func TestApplyMemoryLimit_DisabledPerPlane(t *testing.T) { + pinRuntimeMemLimit(t) + t.Setenv("GOMEMLIMIT", "") + t.Setenv(memLimitEnv, "4096") + applyMemoryLimit(nil, memPlaneMonolith) + before := debug.SetMemoryLimit(-1) + + t.Setenv(memLimitEnv, "off") + for _, p := range []memPlane{memPlaneMonolith, memPlaneServe, memPlaneEngine} { + applyMemoryLimit(nil, p) + if got := debug.SetMemoryLimit(-1); got != before { + t.Errorf("plane %s: disabled path changed the limit %d -> %d", p, before, got) + } + } +} + +// TestMemLimitPlaneSummary_ReportsBothShares backs the operator-facing surface: +// grafel status must be able to print the effective total AND both shares. +// Reporting only one share (the current 2x-understating behaviour) fails here. +func TestMemLimitPlaneSummary_ReportsBothShares(t *testing.T) { + t.Setenv("GOMEMLIMIT", "") + t.Setenv(memLimitEnv, "10000") + + t.Setenv(SplitModeEnvVar, "1") + total, serve, engine, _, split := MemLimitPlaneSummary() + if !split { + t.Fatal("split mode on: want split=true") + } + if total != 10000 { + t.Errorf("total = %d, want 10000", total) + } + if serve+engine != total { + t.Errorf("shares %d+%d do not sum to the advertised total %d", serve, engine, total) + } + if serve == total || engine == total { + t.Errorf("a share equals the total (serve=%d engine=%d total=%d): status would understate real consumption by 2x", serve, engine, total) + } + + t.Setenv(SplitModeEnvVar, "0") + total, serve, engine, _, split = MemLimitPlaneSummary() + if split { + t.Fatal("monolith mode: want split=false") + } + if total != 10000 || serve != 10000 { + t.Errorf("monolith: total=%d serve=%d, want the whole 10000 in one process", total, serve) + } + if engine != 0 { + t.Errorf("monolith: engine share = %d, want 0 (no engine process)", engine) + } +} diff --git a/internal/daemon/memlimit_test.go b/internal/daemon/memlimit_test.go index 9b5854a4e..85ef048d6 100644 --- a/internal/daemon/memlimit_test.go +++ b/internal/daemon/memlimit_test.go @@ -149,15 +149,15 @@ func TestApplyMemoryLimit_DoesNotPanic(t *testing.T) { debug.SetMemoryLimit(prev) t.Cleanup(func() { debug.SetMemoryLimit(prev) }) - t.Setenv("GOMEMLIMIT", "") // don't defer to a real env limit - t.Setenv(memLimitEnv, "8192") // 8GB soft limit - applyMemoryLimit(nil) // nil logger → slog.Default() + t.Setenv("GOMEMLIMIT", "") // don't defer to a real env limit + t.Setenv(memLimitEnv, "8192") // 8GB soft limit + applyMemoryLimit(nil, memPlaneMonolith) // nil logger → slog.Default() if got := debug.SetMemoryLimit(-1); got != 8192*1024*1024 { t.Errorf("applyMemoryLimit(8192MB): runtime limit = %d, want %d", got, int64(8192)*1024*1024) } t.Setenv(memLimitEnv, "off") // disabled path must not change the limit - applyMemoryLimit(nil) + applyMemoryLimit(nil, memPlaneMonolith) if got := debug.SetMemoryLimit(-1); got != 8192*1024*1024 { t.Errorf("disabled path changed the limit to %d", got) } diff --git a/internal/daemon/server.go b/internal/daemon/server.go index d1c918b0b..d3f45ebc3 100644 --- a/internal/daemon/server.go +++ b/internal/daemon/server.go @@ -423,8 +423,10 @@ func RunEngine(ctx context.Context, cfg EngineConfig) error { // #3648: match Run's conservative Go soft memory limit — the engine is the // heavy plane (extraction/reindex/fbwriter), exactly the workload the cap - // bounds. - applyMemoryLimit(logger) + // bounds. #6045: the resolved limit is the budget for the WHOLE + // installation, so the engine takes only its (majority) share of it — serve + // takes the rest in its own process. + applyMemoryLimit(logger, memPlaneEngine) if err := EnsureLayout(cfg.Layout); err != nil { return fmt.Errorf("engine: ensure layout: %w", err) @@ -541,7 +543,12 @@ func run(ctx context.Context, cfg Config, plane daemonPlaneMode) error { // 16GB host during concurrent reindex bursts. Combined with the // scheduler's idle FreeOSMemory trigger this attacks both the PEAK // (GOMEMLIMIT) and the idle RETAINED arena (FreeOSMemory). - applyMemoryLimit(logger) + // + // #6045: the resolved limit is an INSTALLATION budget. In split mode this + // is the serve plane and a sibling engine process takes the rest, so serve + // applies only its share; in the monolith there is one process and it keeps + // the whole budget. + applyMemoryLimit(logger, memPlaneForDaemonPlane(plane)) // Layer 1 self-defense: refuse to start if a canonical (non-/tmp) daemon // is already running and this binary lives under /tmp. This prevents the