diff --git a/internal/cli/status.go b/internal/cli/status.go index 7fb2aebd2..efd3239d5 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -7,7 +7,6 @@ import ( "os" "path/filepath" "sort" - "strings" "time" "github.com/spf13/cobra" @@ -129,130 +128,7 @@ func runStatus(w io.Writer, filter string, ref string, showAll bool) error { fmt.Fprintf(w, " mem_limit: unbounded (%s)\n", src) } } - if st.WatcherRepos > 0 || st.WatcherEvents > 0 { - fmt.Fprintf(w, " watcher: repos=%d dirs=%d events=%d dropped=%d", - st.WatcherRepos, st.WatcherDirs, st.WatcherEvents, st.WatcherDropped) - // PH2a (#2096): show pause/resume slot counts when available. - if st.WatcherActiveSlots > 0 || st.WatcherPausedSlots > 0 { - fmt.Fprintf(w, " slots_active=%d slots_paused=%d", - st.WatcherActiveSlots, st.WatcherPausedSlots) - } - fmt.Fprintln(w) - } - if st.QueueLen > 0 || len(st.IndexInFlight) > 0 || - len(st.PendingAlgo) > 0 || len(st.PendingLinks) > 0 { - fmt.Fprintf(w, " scheduler: queue=%d in_flight=%d pending_algo=%d pending_links=%d\n", - st.QueueLen, len(st.IndexInFlight), len(st.PendingAlgo), len(st.PendingLinks)) - } - printAnnotationStatus(w, st) - // #5954 heavy write-stage gate. Printed whenever the gate is doing - // ANYTHING (holding, deferring, or being barged) so an operator — or - // a peak-RSS measurement run — can confirm from outside the process - // that it fired, instead of inferring it from RSS shape. Silent when - // the gate is idle, which is the overwhelmingly common case. - if st.StageGateHolder != "" || len(st.StageGateDeferred) > 0 || len(st.StageGateBarging) > 0 || - st.StageGateForfeits > 0 { - fmt.Fprint(w, " stage_gate:") - if len(st.StageGateBarging) > 0 { - fmt.Fprintf(w, " barging=%s", strings.Join(st.StageGateBarging, ",")) - } - if st.StageGateHolder != "" { - fmt.Fprintf(w, " holder=%s", st.StageGateHolder) - // LIVE, unlike FORFEITS below: this holder blew the 4h - // hold-max and the gate is inside its forfeit grace right - // now. It is the state an operator staring at a stalled - // daemon is trying to identify, so it rides on the holder. - if st.StageGateForfeitedHolder { - fmt.Fprint(w, "(FORFEITED,awaiting-cancel)") - } - } - if len(st.StageGateDeferred) > 0 { - fmt.Fprintf(w, " deferred=%s", strings.Join(st.StageGateDeferred, ",")) - } - // Sticky: a forfeit is a failure, so it stays on the line for - // the life of the daemon rather than vanishing with the holder. - if st.StageGateForfeits > 0 { - fmt.Fprintf(w, " FORFEITS=%d", st.StageGateForfeits) - } - fmt.Fprintln(w) - } - if st.RSSBudgetMB > 0 { - // Two separate lines: daemon idle RSS (informational) vs. - // admission budget (delta-based predicted in-flight sum). - // These are intentionally distinct — idle RSS can exceed the - // budget without blocking jobs, because jobs are only blocked - // when sum(predicted_in_flight) + new_job_pred > budget. - // #3648: report the honest process footprint (resident set - // size), not the old MemStats.Sys mislabel. On macOS RSS - // under-counts swapped/compressed pages — note that, plus the - // Go-heap breakdown, so the number is interpretable. - fmt.Fprintf(w, " mem: footprint=%dMB heap_inuse=%dMB heap_released=%dMB go_sys=%dMB\n", - st.RSSUsedMB, - st.HeapInuseBytes/(1024*1024), - st.HeapReleasedBytes/(1024*1024), - st.SysBytes/(1024*1024)) - if st.FootprintLabel != "" { - fmt.Fprintf(w, " (footprint = %s)\n", st.FootprintLabel) - } - admHeadroom := st.RSSBudgetMB - st.AdmissionUsedMB - if admHeadroom < 0 { - admHeadroom = 0 - } - fmt.Fprintf(w, " admission: queued=%d admitted=%d predicted=%dMB / budget=%dMB (headroom=%dMB)\n", - len(st.BlockedJobs), len(st.InFlightJobs), - st.AdmissionUsedMB, st.RSSBudgetMB, admHeadroom) - if st.RebuildConcurrencyCap > 0 { - fmt.Fprintf(w, " rebuild: in_flight=%d / cap=%d\n", - st.RebuildInFlight, st.RebuildConcurrencyCap) - } - if len(st.InFlightJobs) > 0 { - for _, j := range st.InFlightJobs { - fmt.Fprintf(w, " admitted: %s (predicted=%dMB)\n", j.Path, j.PredictedMB) - } - } - if len(st.BlockedJobs) > 0 { - for _, p := range st.BlockedJobs { - fmt.Fprintf(w, " queued: %s\n", p) - } - } - } - if len(st.IndexedRepos) > 0 { - fmt.Fprintln(w, " indexed repos:") - for _, r := range st.IndexedRepos { - last := r.LastIndex - if last == "" { - last = "(never)" - } - fmt.Fprintf(w, " %s last_index=%s indexes=%d algos=%d", - r.Path, last, r.IndexCount, r.AlgoCount) - if r.LastErr != "" { - fmt.Fprintf(w, " err=%s", r.LastErr) - } - // #5727/#5729-W1: show the exact indexed commit + whether - // the on-disk graph still matches HEAD. - if r.IndexedCommitShort != "" { - fmt.Fprintf(w, " commit=%s at_head=%v", r.IndexedCommitShort, r.AtHead) - } - fmt.Fprintln(w) - } - } - if n := len(st.RecentLog); n > 0 { - start := n - 5 - if start < 0 { - start = 0 - } - fmt.Fprintln(w, " recent events:") - for _, e := range st.RecentLog[start:] { - line := fmt.Sprintf(" %s %s", e.Time, e.Kind) - if e.Repo != "" { - line += " " + e.Repo - } - if e.Msg != "" { - line += " " + e.Msg - } - fmt.Fprintln(w, line) - } - } + printDaemonDetail(w, st) } case errors.Is(err, client.ErrDaemonNotRunning): fmt.Fprintln(w, "Daemon: not running") diff --git a/internal/cli/status_daemon_detail.go b/internal/cli/status_daemon_detail.go new file mode 100644 index 000000000..872469cfb --- /dev/null +++ b/internal/cli/status_daemon_detail.go @@ -0,0 +1,157 @@ +package cli + +// status_daemon_detail.go -- the daemon-section renderer for `grafel status`, +// split out of runStatus so the print gates are unit-testable without dialing a +// real daemon (#6014). runStatus itself needs a live client.Dial, which put +// every one of these conditionals out of reach of a test -- and an unreachable +// print gate is precisely how #6014's engine-sourced rebuild count came to be +// computed correctly and then never shown to anyone. + +import ( + "fmt" + "io" + "strings" + + "github.com/cajasmota/grafel/internal/daemon/proto" +) + +// printDaemonDetail renders the per-subsystem detail lines of the daemon +// section: watcher, scheduler, annotation pass, stage gate, rebuild counters, +// memory/admission, indexed repos and the recent-event tail. Every line is +// individually gated on its subsystem being present or busy, so an idle daemon +// prints only the header lines runStatus emits directly. +func printDaemonDetail(w io.Writer, st proto.StatusReply) { + if st.WatcherRepos > 0 || st.WatcherEvents > 0 { + fmt.Fprintf(w, " watcher: repos=%d dirs=%d events=%d dropped=%d", + st.WatcherRepos, st.WatcherDirs, st.WatcherEvents, st.WatcherDropped) + // PH2a (#2096): show pause/resume slot counts when available. + if st.WatcherActiveSlots > 0 || st.WatcherPausedSlots > 0 { + fmt.Fprintf(w, " slots_active=%d slots_paused=%d", + st.WatcherActiveSlots, st.WatcherPausedSlots) + } + fmt.Fprintln(w) + } + if st.QueueLen > 0 || len(st.IndexInFlight) > 0 || + len(st.PendingAlgo) > 0 || len(st.PendingLinks) > 0 { + fmt.Fprintf(w, " scheduler: queue=%d in_flight=%d pending_algo=%d pending_links=%d\n", + st.QueueLen, len(st.IndexInFlight), len(st.PendingAlgo), len(st.PendingLinks)) + } + printAnnotationStatus(w, st) + // #5954 heavy write-stage gate. Printed whenever the gate is doing + // ANYTHING (holding, deferring, or being barged) so an operator — or + // a peak-RSS measurement run — can confirm from outside the process + // that it fired, instead of inferring it from RSS shape. Silent when + // the gate is idle, which is the overwhelmingly common case. + if st.StageGateHolder != "" || len(st.StageGateDeferred) > 0 || len(st.StageGateBarging) > 0 || + st.StageGateForfeits > 0 { + fmt.Fprint(w, " stage_gate:") + if len(st.StageGateBarging) > 0 { + fmt.Fprintf(w, " barging=%s", strings.Join(st.StageGateBarging, ",")) + } + if st.StageGateHolder != "" { + fmt.Fprintf(w, " holder=%s", st.StageGateHolder) + // LIVE, unlike FORFEITS below: this holder blew the 4h + // hold-max and the gate is inside its forfeit grace right + // now. It is the state an operator staring at a stalled + // daemon is trying to identify, so it rides on the holder. + if st.StageGateForfeitedHolder { + fmt.Fprint(w, "(FORFEITED,awaiting-cancel)") + } + } + if len(st.StageGateDeferred) > 0 { + fmt.Fprintf(w, " deferred=%s", strings.Join(st.StageGateDeferred, ",")) + } + // Sticky: a forfeit is a failure, so it stays on the line for + // the life of the daemon rather than vanishing with the holder. + if st.StageGateForfeits > 0 { + fmt.Fprintf(w, " FORFEITS=%d", st.StageGateForfeits) + } + fmt.Fprintln(w) + } + // #6014: the rebuild counters are printed OUTSIDE the RSSBudgetMB block + // below. RSSBudgetMB is populated only from an in-process scheduler + // (Service.Status' `if s.scheduler != nil` branch), and in SPLIT MODE -- the + // default since ADR-0024 -- serve has no scheduler, so it is 0 and that + // whole block never prints. Nesting this line there made the engine-sourced + // rebuild count unreachable on the one surface an operator actually reads: + // the number crossed the process boundary correctly and was then discarded + // by the print gate. RebuildConcurrencyCap is set unconditionally by + // Service.Status, so gating on it alone is both sufficient and mode-neutral. + if st.RebuildConcurrencyCap > 0 { + fmt.Fprintf(w, " rebuild: in_flight=%d / cap=%d\n", + st.RebuildInFlight, st.RebuildConcurrencyCap) + } + if st.RSSBudgetMB > 0 { + // Two separate lines: daemon idle RSS (informational) vs. + // admission budget (delta-based predicted in-flight sum). + // These are intentionally distinct — idle RSS can exceed the + // budget without blocking jobs, because jobs are only blocked + // when sum(predicted_in_flight) + new_job_pred > budget. + // #3648: report the honest process footprint (resident set + // size), not the old MemStats.Sys mislabel. On macOS RSS + // under-counts swapped/compressed pages — note that, plus the + // Go-heap breakdown, so the number is interpretable. + fmt.Fprintf(w, " mem: footprint=%dMB heap_inuse=%dMB heap_released=%dMB go_sys=%dMB\n", + st.RSSUsedMB, + st.HeapInuseBytes/(1024*1024), + st.HeapReleasedBytes/(1024*1024), + st.SysBytes/(1024*1024)) + if st.FootprintLabel != "" { + fmt.Fprintf(w, " (footprint = %s)\n", st.FootprintLabel) + } + admHeadroom := st.RSSBudgetMB - st.AdmissionUsedMB + if admHeadroom < 0 { + admHeadroom = 0 + } + fmt.Fprintf(w, " admission: queued=%d admitted=%d predicted=%dMB / budget=%dMB (headroom=%dMB)\n", + len(st.BlockedJobs), len(st.InFlightJobs), + st.AdmissionUsedMB, st.RSSBudgetMB, admHeadroom) + if len(st.InFlightJobs) > 0 { + for _, j := range st.InFlightJobs { + fmt.Fprintf(w, " admitted: %s (predicted=%dMB)\n", j.Path, j.PredictedMB) + } + } + if len(st.BlockedJobs) > 0 { + for _, p := range st.BlockedJobs { + fmt.Fprintf(w, " queued: %s\n", p) + } + } + } + if len(st.IndexedRepos) > 0 { + fmt.Fprintln(w, " indexed repos:") + for _, r := range st.IndexedRepos { + last := r.LastIndex + if last == "" { + last = "(never)" + } + fmt.Fprintf(w, " %s last_index=%s indexes=%d algos=%d", + r.Path, last, r.IndexCount, r.AlgoCount) + if r.LastErr != "" { + fmt.Fprintf(w, " err=%s", r.LastErr) + } + // #5727/#5729-W1: show the exact indexed commit + whether + // the on-disk graph still matches HEAD. + if r.IndexedCommitShort != "" { + fmt.Fprintf(w, " commit=%s at_head=%v", r.IndexedCommitShort, r.AtHead) + } + fmt.Fprintln(w) + } + } + if n := len(st.RecentLog); n > 0 { + start := n - 5 + if start < 0 { + start = 0 + } + fmt.Fprintln(w, " recent events:") + for _, e := range st.RecentLog[start:] { + line := fmt.Sprintf(" %s %s", e.Time, e.Kind) + if e.Repo != "" { + line += " " + e.Repo + } + if e.Msg != "" { + line += " " + e.Msg + } + fmt.Fprintln(w, line) + } + } +} diff --git a/internal/cli/status_daemon_detail_6014_test.go b/internal/cli/status_daemon_detail_6014_test.go new file mode 100644 index 000000000..406b1be8d --- /dev/null +++ b/internal/cli/status_daemon_detail_6014_test.go @@ -0,0 +1,81 @@ +package cli + +// status_daemon_detail_6014_test.go — #6014, the print gate. +// +// The engine-sourced rebuild count reached StatusReply.RebuildInFlight +// correctly and was then thrown away by the renderer: the `rebuild:` line lived +// inside `if st.RSSBudgetMB > 0`, and RSSBudgetMB is populated only from an +// IN-PROCESS scheduler (Service.Status' `if s.scheduler != nil` branch). In +// split mode — the default since ADR-0024 — serve has no scheduler, so +// RSSBudgetMB is 0 and the whole block, rebuild line included, never printed. +// `grafel status` showed no `rebuild:` line at all. +// +// That is the same shape as the bug being fixed: a number that is right on the +// wire and absent on the screen. So the gate is pinned here, in the one +// configuration that matters — no scheduler, no RSS budget. + +import ( + "bytes" + "strings" + "testing" + + "github.com/cajasmota/grafel/internal/daemon/proto" +) + +// TestPrintDaemonDetail_RebuildLinePrintsWithNoRSSBudget is the regression: a +// split-mode serve reply (scheduler-derived fields all zero) must still show +// the rebuild counters. Re-nesting the line under RSSBudgetMB kills this test. +func TestPrintDaemonDetail_RebuildLinePrintsWithNoRSSBudget(t *testing.T) { + var buf bytes.Buffer + printDaemonDetail(&buf, proto.StatusReply{ + // Split-mode serve: no scheduler, so every scheduler-sourced field is + // zero — RSSBudgetMB above all. + RSSBudgetMB: 0, + RebuildInFlight: 1, + RebuildConcurrencyCap: 2, + }) + out := buf.String() + + if !strings.Contains(out, "rebuild: in_flight=1 / cap=2") { + t.Errorf("`rebuild:` line missing from `grafel status` output with RSSBudgetMB=0.\n"+ + "This is the DEFAULT deployment mode: serve has no scheduler, so a rebuild line gated on the "+ + "RSS-budget block is unreachable and the engine-sourced count is computed, transmitted, and "+ + "then discarded before anyone sees it.\ngot:\n%s", out) + } + // Guard the hoist rather than the copy: the RSS/admission block must still + // be suppressed when there is no budget to report. + if strings.Contains(out, "admission:") { + t.Errorf("admission block printed with RSSBudgetMB=0; the rebuild line should have been hoisted "+ + "OUT of that block, not the block un-gated.\ngot:\n%s", out) + } +} + +// TestPrintDaemonDetail_RebuildLineStillPrintsWithRSSBudget: the monolith path +// (scheduler attached, budget reported) must be unchanged by the hoist. +func TestPrintDaemonDetail_RebuildLineStillPrintsWithRSSBudget(t *testing.T) { + var buf bytes.Buffer + printDaemonDetail(&buf, proto.StatusReply{ + RSSBudgetMB: 4096, + RebuildInFlight: 3, + RebuildConcurrencyCap: 4, + }) + out := buf.String() + if !strings.Contains(out, "rebuild: in_flight=3 / cap=4") { + t.Errorf("`rebuild:` line missing in monolith mode (RSSBudgetMB > 0).\ngot:\n%s", out) + } + if !strings.Contains(out, "admission:") { + t.Errorf("admission block missing with RSSBudgetMB > 0 — the hoist changed monolith output.\ngot:\n%s", out) + } +} + +// TestPrintDaemonDetail_NoRebuildLineWithoutCap: RebuildConcurrencyCap is set +// unconditionally by Service.Status, so a zero means the reply did not come +// from a daemon that reports it (an old daemon, or a zero-valued fixture). +// Printing `cap=0` there would be noise, not information. +func TestPrintDaemonDetail_NoRebuildLineWithoutCap(t *testing.T) { + var buf bytes.Buffer + printDaemonDetail(&buf, proto.StatusReply{}) + if out := buf.String(); strings.Contains(out, "rebuild:") { + t.Errorf("`rebuild:` line printed with RebuildConcurrencyCap=0.\ngot:\n%s", out) + } +} diff --git a/internal/daemon/engine_status.go b/internal/daemon/engine_status.go index 0e5e07e5e..81a34d21e 100644 --- a/internal/daemon/engine_status.go +++ b/internal/daemon/engine_status.go @@ -135,6 +135,52 @@ func GroupAlgoInFlightFromStatusFile() int { return f.EngineGroupAlgoInFlight } +// RebuildInFlightFromStatusFile reports how many GROUP REBUILDS the ambient +// engine is running, read from its liveness sidecar by a process that has none +// of its own — a SPLIT-MODE serve answering Service.Status (#6014). +// +// This is the exact analogue of GroupAlgoInFlightFromStatusFile, and shares its +// file and its freshness rule, so both modes converge on one path. It exists +// because Service.rebuildInFlight / groupsActiveCount are STRUCTURALLY zero in +// split mode: their increments sit below Service.Rebuild's split-mode early +// return, and the rebuild itself runs in the engine. Serve therefore has no +// local information at all about a running rebuild, so StatusReply.RebuildInFlight +// was a constant 0 — a working rebuild was indistinguishable from a dropped one. +// +// The RPC field was only half the problem. `grafel status` printed its +// `rebuild: in_flight=` line inside a block gated on StatusReply.RSSBudgetMB, +// which is itself populated only from an in-process scheduler — so in split +// mode the line did not print AT ALL, and a correct value here would still have +// been invisible. That gate was hoisted in the same change (internal/cli's +// printDaemonDetail); this accessor is only observable because of it. +// +// LATENCY, honestly: the value is published by a plain 5s ticker +// (startEngineLivenessHeartbeat, defaultStatusHeartbeatInterval) with no +// state-change coalescing — unlike startStatusWriter, which has a notify seam. +// So a rebuild that has just started still reads 0 here for up to one heartbeat +// interval. Group rebuilds are multi-minute, so a ≤5s false zero at the very +// start is acceptable; it is called out because it is this same bug in +// miniature, and anyone tightening it should add a notify on +// rebuildWorker.submit rather than shortening the tick. +// +// 0 when the sidecar is missing or stale ("unknown" reported as "none"), the +// same conservative default as the rest of this file. A stale sidecar's +// last-known count must NEVER be reported as live: this surface exists to +// answer "is a rebuild running RIGHT NOW", so a phantom non-zero would mislead +// exactly the reader it was built for — no better than the wrong zero it +// replaces. +func RebuildInFlightFromStatusFile() int { + layout, err := DefaultLayout() + if err != nil { + return 0 + } + f, fresh := EngineLivenessStatus(layout.Root) + if !fresh || f == nil { + return 0 + } + return f.EngineRebuildInFlight +} + // FlushRepoStatusFile synchronously recomputes and writes repoPath's per-repo // status-plane sidecar from the CURRENT indexstate + the on-disk graph.fb, right // now — the same work the async statusWriter goroutine would do on its next diff --git a/internal/daemon/requests_drain.go b/internal/daemon/requests_drain.go index eb7638494..a750c49a8 100644 --- a/internal/daemon/requests_drain.go +++ b/internal/daemon/requests_drain.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "sync" + "sync/atomic" "time" "github.com/cajasmota/grafel/internal/daemon/proto" @@ -76,6 +77,26 @@ const maxRebuildAttempts = 3 // DIFFERENT groups still proceed concurrently. var engineGroupRebuildGuard groupRebuildGuard +// engineRebuildInFlight counts the group rebuilds this ENGINE process is +// running right now (#6014). One entry per rebuildWorker.submit goroutine, i.e. +// per group, so it is simultaneously "rebuilds in flight" and "groups actively +// rebuilding" — the engine guard is per-group and admits one at a time. +// +// WHY A COUNTER AT ALL. In split mode (the default, ADR-0024) the rebuild runs +// HERE, in the engine, while `grafel status` is answered by SERVE. Serve's own +// Service.rebuildInFlight can never be non-zero in that mode — its increment is +// below the split-mode early return — so serve has literally no local +// information from which to report a running rebuild. This counter is the real, +// work-derived number the heartbeat publishes onto the engine-liveness sidecar +// for serve to read. Serve never invents a value of its own. +var engineRebuildInFlight atomic.Int64 + +// EngineRebuildInFlightCount reports how many group rebuilds this process is +// running. Read by the engine-liveness heartbeat (statuswriter.go); meaningful +// only in the process that actually applies rebuilds (the engine, or a monolith +// draining its own queue). +func EngineRebuildInFlightCount() int { return int(engineRebuildInFlight.Load()) } + // groupRebuildGuard is a per-group capacity-1 semaphore keyed by group name. type groupRebuildGuard struct { sems sync.Map // group name -> chan struct{} (cap 1) @@ -274,8 +295,15 @@ func (w *rebuildWorker) submit(root, dir, group string) { w.wg.Add(1) w.mu.Unlock() + // #6014: the process-global mirror of w.active, published by the engine's + // liveness heartbeat so a split-mode serve can report a running rebuild. + // Incremented here — on the claim, in lockstep with w.active — rather than + // inside runGroup, so the count can never disagree with the guard. + engineRebuildInFlight.Add(1) + go func() { defer func() { + engineRebuildInFlight.Add(-1) w.mu.Lock() delete(w.active, group) w.mu.Unlock() diff --git a/internal/daemon/service.go b/internal/daemon/service.go index 0982025d2..1f059963d 100644 --- a/internal/daemon/service.go +++ b/internal/daemon/service.go @@ -475,6 +475,23 @@ func (s *Service) Status(_ *proto.StatusArgs, reply *proto.StatusReply) error { // enough to answer "is an overlay recompute in flight" — which is the // question behind "why are my communities missing". reply.GroupAlgoInFlight = GroupAlgoInFlightFromStatusFile() + // #6014: the REBUILD counters have the same problem and the same cure. + // s.rebuildInFlight / s.groupsActiveCount are incremented below + // Service.Rebuild's split-mode early return, so in the default mode they + // are a structural zero — "no rebuild is running" was reported whether + // or not one was. The engine's per-group rebuild guard is the real + // count; it rides the same sidecar. + // + // Assigned only when non-zero: a fresh sidecar reporting 0 tells us + // nothing the local 0 did not already say, and this way the branch can + // never clobber a genuine local count in the (test-only) shape where a + // monolith has no scheduler but does rebuild in-process. + if n := RebuildInFlightFromStatusFile(); n > 0 { + reply.RebuildInFlight = n + // The engine guard is per-group and admits one rebuild per group at + // a time, so the in-flight count IS the active-group count. + reply.RebuildGroupsActive = n + } } return nil } @@ -622,6 +639,24 @@ func (s *Service) Rebuild(args *proto.RebuildArgs, reply *proto.RebuildReply) er s.progressBroker.ClearTerminal(args.Group) } + // #6014 — s.inFlight is serve's count of RPCs currently executing IN THIS + // PROCESS (StatusReply.InFlight, printed as `in_flight=` by `grafel + // status`). It used to be incremented further down, below the split-mode + // early return, which made it structurally unreachable in the default mode. + // That was invisible for the fire-and-forget path (nothing IS in flight + // there — the RPC returns as soon as the request file lands) but a real lie + // for WaitForCompletion, which BLOCKS this handler in + // awaitRebuildCompletion for the entire multi-minute rebuild while + // reporting in_flight=0. + // + // Incrementing at the top of the handler — before any branch — is both the + // fix and the honest definition: the RPC is in flight for exactly as long + // as this function is on the stack, in every mode. It says nothing about + // whether the ENGINE is rebuilding; that is RebuildInFlight's job, sourced + // from the engine's liveness sidecar (see Status). + atomic.AddInt64(&s.inFlight, 1) + defer atomic.AddInt64(&s.inFlight, -1) + // ADR-0024 PR6 prerequisite (epic #5729): in split mode this RPC is // answered by the SERVE process. Unlike Service.Index (whose split-mode // fast path is gated by s.scheduler being nil there), s.rebuild here is @@ -682,9 +717,6 @@ func (s *Service) Rebuild(args *proto.RebuildArgs, reply *proto.RebuildReply) er return s.awaitRebuildCompletion(dir, id) } - atomic.AddInt64(&s.inFlight, 1) - defer atomic.AddInt64(&s.inFlight, -1) - // Per-group single-flight (#2097 + #5681). Load-or-store a capacity-1 // semaphore for this group so concurrent Rebuild RPCs targeting the same // group are serialised rather than racing on the same output files AND, diff --git a/internal/daemon/status_rebuild_inflight_6014_test.go b/internal/daemon/status_rebuild_inflight_6014_test.go new file mode 100644 index 000000000..753fcfc17 --- /dev/null +++ b/internal/daemon/status_rebuild_inflight_6014_test.go @@ -0,0 +1,266 @@ +package daemon + +// status_rebuild_inflight_6014_test.go — #6014. +// +// `grafel status` reported rebuild_in_flight / in_flight as a STRUCTURAL ZERO +// in split mode (the default since ADR-0024). Two independent defects produced +// it, and both are pinned here: +// +// 1. THE WORK MOVED PROCESSES. Service.Rebuild's rebuildInFlight / +// groupsActiveCount increments sit BELOW the `if SplitModeEnabled()` early +// return, on the monolith path. In split mode the rebuild runs in the +// ENGINE process, so serve has no local counter that could ever be +// non-zero. The only honest fix is cross-process: the engine publishes its +// live rebuild count onto the liveness sidecar it already heartbeats, and +// serve reads it — exactly the route #6002 established for +// GroupAlgoInFlight and #5954 for the stage gate. Serve must NOT invent a +// number of its own; a wrong non-zero is no better than a wrong zero. +// +// Getting the number into the RPC reply is necessary but NOT sufficient: +// `grafel status` gated its `rebuild:` line on RSSBudgetMB, which only an +// in-process scheduler populates, so in split mode the line never printed +// regardless. That gate is pinned separately, in internal/cli's +// status_daemon_detail_6014_test.go — nothing in THIS file would notice it +// regressing. +// +// 2. THE RPC ITSELF WAS UNCOUNTED. Service.Rebuild's `s.inFlight` increment +// also sat below the split-mode branch, so a WaitForCompletion rebuild — +// which BLOCKS the handler for the whole rebuild — was invisible in +// StatusReply.InFlight, the "in_flight=" field `grafel status` prints. +// That one is purely serve-local and needs no sidecar: the RPC is in +// flight in serve, in serve's own memory, for as long as it blocks. + +import ( + "sync" + "testing" + "time" + + "github.com/cajasmota/grafel/internal/daemon/proto" + "github.com/cajasmota/grafel/internal/daemon/requests" + "github.com/cajasmota/grafel/internal/statusfile" +) + +// TestStatus_RebuildInFlightFromEngineSidecarInSplitMode is the READ half: a +// serve process with no scheduler must surface the engine's live rebuild count. +// Without it, "is a rebuild running right now?" is answered `0` in the default +// deployment mode no matter what the engine is doing. +func TestStatus_RebuildInFlightFromEngineSidecarInSplitMode(t *testing.T) { + t.Setenv("GRAFEL_HOME", t.TempDir()) + svc := newStageGateTestService(t) + if svc.scheduler != nil { + t.Fatal("precondition: this test models a serve process with no scheduler") + } + + writeStageGateLivenessFixture(t, &statusfile.File{ + HeartbeatAt: time.Now().UTC(), + EngineRebuildInFlight: 2, + }) + + var reply proto.StatusReply + if err := svc.Status(&proto.StatusArgs{}, &reply); err != nil { + t.Fatalf("Status RPC: %v", err) + } + if reply.RebuildInFlight != 2 { + t.Errorf("RebuildInFlight = %d, want 2: in split mode the engine's liveness sidecar is the ONLY "+ + "route by which a running rebuild can reach `grafel status`", reply.RebuildInFlight) + } + if reply.RebuildGroupsActive != 2 { + t.Errorf("RebuildGroupsActive = %d, want 2: the engine's guard is per-group, so its in-flight "+ + "count IS the number of groups actively rebuilding", reply.RebuildGroupsActive) + } +} + +// TestStatus_RebuildInFlightOmittedWhenSidecarStale: a stale heartbeat means the +// engine is down, starting, or wedged. Reporting its last-known rebuild count as +// LIVE would be a fabricated non-zero — the precise failure mode #6014 warns +// against, and worse than reporting nothing. +func TestStatus_RebuildInFlightOmittedWhenSidecarStale(t *testing.T) { + t.Setenv("GRAFEL_HOME", t.TempDir()) + svc := newStageGateTestService(t) + + writeStageGateLivenessFixture(t, &statusfile.File{ + HeartbeatAt: time.Now().UTC().Add(-24 * time.Hour), + EngineRebuildInFlight: 3, + }) + + var reply proto.StatusReply + if err := svc.Status(&proto.StatusArgs{}, &reply); err != nil { + t.Fatalf("Status RPC: %v", err) + } + if reply.RebuildInFlight != 0 || reply.RebuildGroupsActive != 0 { + t.Fatalf("stale sidecar reported as a live rebuild: in_flight=%d groups=%d; want 0 (unknown)", + reply.RebuildInFlight, reply.RebuildGroupsActive) + } +} + +// TestRebuildInFlightFromStatusFile_FreshVsStale binds the READER'S OWN +// freshness guard directly. +// +// The Status-RPC test above cannot do it: the split-mode branch it exercises is +// itself entered via `StageGateFromStatusFile() ok`, which already encodes the +// same freshness rule — so deleting RebuildInFlightFromStatusFile's guard leaves +// that test green while the function is, on its own terms, wrong. Any other +// caller (and there will be one) would then read a dead engine's last-known +// count as live. Pinning the accessor directly is what makes the guard real +// rather than decorative. +func TestRebuildInFlightFromStatusFile_FreshVsStale(t *testing.T) { + t.Setenv("GRAFEL_HOME", t.TempDir()) + + writeStageGateLivenessFixture(t, &statusfile.File{ + HeartbeatAt: time.Now().UTC(), + EngineRebuildInFlight: 4, + }) + if got := RebuildInFlightFromStatusFile(); got != 4 { + t.Errorf("fresh sidecar: RebuildInFlightFromStatusFile = %d, want 4", got) + } + + writeStageGateLivenessFixture(t, &statusfile.File{ + HeartbeatAt: time.Now().UTC().Add(-24 * time.Hour), + EngineRebuildInFlight: 4, + }) + if got := RebuildInFlightFromStatusFile(); got != 0 { + t.Errorf("stale sidecar: RebuildInFlightFromStatusFile = %d, want 0 — a dead engine's last-known "+ + "count reported as live is a fabricated non-zero, no better than the wrong zero it replaces", got) + } +} + +// TestEngineLivenessHeartbeat_PublishesRebuildInFlight is the WRITE half. A +// field the reader handles but the heartbeat never stamps is dead in the default +// deployment mode while looking fully wired. +func TestEngineLivenessHeartbeat_PublishesRebuildInFlight(t *testing.T) { + root := t.TempDir() + + engineRebuildInFlight.Add(2) + defer engineRebuildInFlight.Add(-2) + + stop := startEngineLivenessHeartbeat(root, 0, nil, nil, nil) + stop() // writeOnce runs synchronously before the ticker loop + + f, err := statusfile.Read(EngineLivenessStatusKey(root)) + if err != nil { + t.Fatalf("read engine liveness sidecar: %v", err) + } + if f.EngineRebuildInFlight != 2 { + t.Errorf("EngineRebuildInFlight = %d, want 2: the heartbeat is the ONLY writer of the split-mode route", + f.EngineRebuildInFlight) + } +} + +// TestRebuildWorker_TracksInFlightCount pins the SOURCE of the published number +// to real work: the counter must rise while rebuildFn is genuinely executing and +// return to zero once it returns. This is what makes the reported value derived +// rather than plausible. +func TestRebuildWorker_TracksInFlightCount(t *testing.T) { + if got := EngineRebuildInFlightCount(); got != 0 { + t.Fatalf("precondition: EngineRebuildInFlightCount = %d, want 0", got) + } + root := t.TempDir() + t.Setenv(EnvRoot, root) + + entered := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + w := newRebuildWorker(func(proto.RebuildArgs) ([]string, string, error) { + once.Do(func() { close(entered) }) + <-release + return nil, "", nil + }, nil) + + dir := requestsDirForGroup("acme") + payload := []byte(`{"group":"acme"}`) + if _, err := requests.Write(dir, requests.Record{Kind: requests.KindRebuild, Payload: payload}); err != nil { + t.Fatalf("queue rebuild request: %v", err) + } + + w.submit(root, dir, "acme") + select { + case <-entered: + case <-time.After(10 * time.Second): + t.Fatal("rebuildFn never ran") + } + if got := EngineRebuildInFlightCount(); got != 1 { + t.Errorf("EngineRebuildInFlightCount = %d while a rebuild is genuinely executing, want 1: "+ + "the published number must be derived from real work, not fabricated", got) + } + close(release) + w.waitIdle() + if got := EngineRebuildInFlightCount(); got != 0 { + t.Errorf("EngineRebuildInFlightCount = %d after the rebuild returned, want 0 (leaked counter)", got) + } +} + +// TestStatus_InFlightCountsBlockingSplitModeRebuildRPC: the WaitForCompletion +// rebuild blocks the serve handler for the entire rebuild, yet +// StatusReply.InFlight — the `in_flight=` field `grafel status` prints — +// reported 0 for the whole time, because the increment sat below the split-mode +// branch. This is serve-local truth about a serve-local RPC; no sidecar needed. +func TestStatus_InFlightCountsBlockingSplitModeRebuildRPC(t *testing.T) { + t.Setenv(SplitModeEnvVar, "1") + root := t.TempDir() + t.Setenv(EnvRoot, root) + t.Setenv("GRAFEL_HOME", t.TempDir()) + + // Keep the completion wait honest but fast: the engine is "alive" and we + // poll frequently, so the handler blocks until we write the terminal ack. + origAlive := rebuildEngineAliveFn + origInterval := rebuildWaitInterval + rebuildEngineAliveFn = func() bool { return true } + rebuildWaitInterval = time.Millisecond + t.Cleanup(func() { + rebuildEngineAliveFn = origAlive + rebuildWaitInterval = origInterval + }) + + svc := newService(nil, func(proto.RebuildArgs) ([]string, string, error) { + t.Error("split mode must never call s.rebuild in serve") + return nil, "", nil + }, nil, "", make(chan struct{}, 1), nil, 1) + + done := make(chan error, 1) + go func() { + var reply proto.RebuildReply + done <- svc.Rebuild(&proto.RebuildArgs{Group: "acme", WaitForCompletion: true}, &reply) + }() + + dir := requestsDirForGroup("acme") + // Wait for the handler to be genuinely blocked in the completion wait: the + // request file is on disk and no ack has been written. + var id string + deadline := time.Now().Add(10 * time.Second) + for { + recs, err := requests.ListPending(dir) + if err == nil && len(recs) == 1 { + id = recs[0].ID + break + } + if time.Now().After(deadline) { + t.Fatalf("rebuild request never landed on disk (err=%v)", err) + } + time.Sleep(2 * time.Millisecond) + } + + var reply proto.StatusReply + if err := svc.Status(&proto.StatusArgs{}, &reply); err != nil { + t.Fatalf("Status RPC: %v", err) + } + inFlight := reply.InFlight + + // Release the blocked handler with a terminal OK ack. + if err := requests.WriteAck(dir, id, requests.Ack{Status: requests.StatusOK}); err != nil { + t.Fatalf("write terminal ack: %v", err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("Rebuild: %v", err) + } + case <-time.After(30 * time.Second): + t.Fatal("Rebuild never returned after the terminal ack") + } + + if inFlight < 1 { + t.Errorf("StatusReply.InFlight = %d while a WaitForCompletion rebuild RPC was BLOCKED in this "+ + "process, want >= 1: `in_flight=` is serve's own RPC count and the RPC was demonstrably in "+ + "flight", inFlight) + } +} diff --git a/internal/daemon/statuswriter.go b/internal/daemon/statuswriter.go index db28984f6..7f8df23f4 100644 --- a/internal/daemon/statuswriter.go +++ b/internal/daemon/statuswriter.go @@ -392,10 +392,14 @@ func startEngineLivenessHeartbeat(root string, interval time.Duration, warmingFn ParseInFlight: snap.ParseInFlight, EngineInFlight: snap.InFlight, EngineGroupAlgoInFlight: snap.GroupAlgoInFlight, - EngineBusyStartedAt: snap.StartedAt, - ConcurrencyActive: conc.Active, - ConcurrencyQueued: conc.Queued, - ConcurrencyCap: conc.Cap, + // #6014: group rebuilds bypass the scheduler, so indexstate knows + // nothing about them. This is the only publication of "a rebuild is + // running in this engine" that crosses the process boundary. + EngineRebuildInFlight: EngineRebuildInFlightCount(), + EngineBusyStartedAt: snap.StartedAt, + ConcurrencyActive: conc.Active, + ConcurrencyQueued: conc.Queued, + ConcurrencyCap: conc.Cap, } populateProcessMetrics(f) // CPU% is a delta across successive heartbeat writes (the first tick diff --git a/internal/dashboard/store.go b/internal/dashboard/store.go index b64ea62c5..28c72aac8 100644 --- a/internal/dashboard/store.go +++ b/internal/dashboard/store.go @@ -224,22 +224,173 @@ func (liveStore) RepoGraph(group, repo string) ([]byte, error) { // repoGraphBytes returns the graph as JSON bytes for a repo. ADR-0016 // flip-day (#808): tries graph.json first (fast raw-read), falls back -// to loading graph.fb and re-marshaling to JSON when only the binary -// graph exists. +// to loading the active FB graph and re-marshaling to JSON. +// +// #6015: the graph.json read used to be UNCONDITIONAL, so a repo that had ever +// produced one — via `--export-json`, or by predating the ADR-0016 layout — +// served that snapshot from the dashboard's RepoGraph / GroupGraph endpoints +// forever, across every subsequent reindex, with no indication it was stale. +// The read is now gated on graphJSONUsable, which is the same stat-compare the +// sibling sidecar readers (graph/descriptions, graph/flows) already perform. func repoGraphBytes(repoPath string) ([]byte, error) { + stateDir := daemon.StateDirForRepo(repoPath) jsonPath := daemon.GraphPathForRepo(repoPath) - if b, err := os.ReadFile(jsonPath); err == nil { - return b, nil + if graphJSONUsable(stateDir, jsonPath) { + if b, err := os.ReadFile(jsonPath); err == nil { + return b, nil + } } - // graph.json not found — try to load from graph.fb and re-marshal. - stateDir := daemon.StateDirForRepo(repoPath) + // No graph.json, or one superseded by a newer graph — load the active FB + // graph and re-marshal. A load error is returned rather than silently + // falling back to the stale JSON: a loud failure beats a snapshot the + // caller would have no way to know is out of date. doc, err := graph.LoadGraphFromDir(stateDir) if err != nil { - return nil, err + return nil, graphUnreadableError(stateDir, err) } return json.Marshal(doc) } +// graphUnreadableError turns a failed graph load into an error that says what +// to DO about it, rather than a bare loader message. +// +// This matters because of what #6015 changed. Before the staleness gate, a repo +// whose graph.fb had gone version-incompatible after an upgrade still rendered: +// it silently fell back to graph.json and showed an older-but-honest graph. +// That fallback is exactly the bug — a stale snapshot standing in for a current +// one, with nothing on the surface saying so — and it is gone. But +// format-version skew is TRANSIENT-BY-REINDEX, not corruption, and answering it +// with an opaque error behind an empty panel tells the user nothing about the +// one action that fixes it. +// +// So the no-fallback behaviour stays and the REASON is surfaced instead. +// graph.ReindexRequiredReason re-reads the on-disk header (segment-set aware) +// and renders the same "graph format vN incompatible — reindex required" +// wording every other consumer of FormatVersionError uses, so a user sees one +// consistent message whichever surface reported it first. Any other load +// failure is passed through unchanged. +func graphUnreadableError(stateDir string, err error) error { + var fvErr *graph.FormatVersionError + if required, reason := graph.ReindexRequiredReason(stateDir); required { + return fmt.Errorf("graph unreadable — %s", reason) + } else if errors.As(err, &fvErr) { + // The header re-read disagreed (the graph changed under us, or only a + // later segment is incompatible) but the load itself reported version + // skew. Trust the load and render the same wording. + return fmt.Errorf("graph unreadable — %s", + graph.FormatVersionReason(fvErr.Found, fvErr.Required)) + } + return err +} + +// graphJSONUsable reports whether stateDir's graph.json may be served as-is, +// i.e. whether it is STRICTLY NEWER than the active graph on disk (#6015). +// +// The comparison is deliberately `After`, never `!Before` — and the reason is +// the opposite of the obvious one. It is not that ties are an unlucky accident +// of filesystem granularity: an equal mtime is DELIBERATELY ENGINEERED. +// cmd/grafel/index.go stamps both artifacts with one timestamp on every export +// (`now := time.Now(); os.Chtimes(fbPath, now, now); os.Chtimes(outPath, now, +// now)`, #1626) precisely so a drift check cannot fire on two encodings of the +// same pass. +// +// So a tie carries two readings that mtime alone cannot separate: "co-written +// by one index pass, and current", or "an older export that a later graph write +// happened to land on". The first is common, the second is vanishingly rare — +// but only the second is a CORRECTNESS question, and the cost of being wrong is +// asymmetric: a needless re-marshal is slow, a silently-stale graph is wrong. +// Ties therefore resolve to regenerate. +// +// CONSEQUENCE, stated plainly because it is a real cost and not a theoretical +// one: for a SINGLE-FILE repo that also exports graph.json, the Chtimes above +// makes the mtimes equal every time, so this returns false every time and the +// raw-read fast path is effectively DISABLED — every dashboard graph request +// re-loads and re-marshals, and GroupGraph does so once per repo in the group. +// SEGMENT-SET repos are unaffected: there fbPath is the gen DIRECTORY (see +// fbwriter.writeSegments' `return genDir, nil`), so Chtimes stamps the dir +// while this gate reads manifest.json — written before the pointer flip, hence +// strictly older — and the JSON stays newer and served. The behaviour is +// genuinely asymmetric between the two layouts. +// +// That asymmetry is accepted here only because the blast radius is narrow: +// FB-only is the default, so this touches repos that opted into `--export-json` +// (and the graph.json written by `grafel quality` / xrepo_verify), never the +// 427k-entity reference corpus. Removing it properly needs a generation +// identity stamped INTO the JSON — the shape descriptions.CurrentSourceKey +// already computes for its sidecars — rather than a timestamp comparison, which +// is a larger change than this fix should carry. +// +// Freshness is measured against graph.CurrentGraphDescriptor, not a hardcoded +// graph.fb path, so a SEGMENT-SET repo (graph./ + manifest.json, no flat +// .fb) is compared against its manifest rather than being mistaken for "no +// graph exists" and pinned to the JSON — the same defect one layout over. +// +// It returns true when graph.json is the ONLY graph on disk (GraphAbsent): a +// legacy JSON-only repo has nothing newer for it to be stale against. +func graphJSONUsable(stateDir, jsonPath string) bool { + ji, err := os.Stat(jsonPath) + // The IsDir arm is belt-and-braces, not a load-bearing guard: a directory at + // graph.json would fail os.ReadFile in repoGraphBytes and fall through to + // the loader anyway. It is kept so this predicate answers honestly on its + // own terms ("is there a graph.json FILE to serve?") rather than relying on + // its caller's error handling, but no test binds it and none should imply + // otherwise. + if err != nil || ji.IsDir() { + return false + } + desc, err := graph.CurrentGraphDescriptor(stateDir) + if err != nil { + // A corrupt/hostile segment-set manifest: a graph exists but cannot be + // dated. Refuse to serve the JSON — the load below will surface the real + // error instead of masking it with a stale snapshot. + return false + } + if desc.Kind == graph.GraphAbsent { + // graph.json is the ONLY graph on disk (a legacy JSON-only repo): there + // is nothing for it to be stale against, and the raw read is the fast + // path. LoadGraphFromDir would also fall back to this same file, but via + // an unmarshal + re-marshal round trip on every request. + return true + } + srcMod, ok := graphSourceModTime(desc) + if !ok { + // A graph exists but cannot be dated (it was replaced or removed between + // the descriptor resolve and the stat). Undatable is treated exactly like + // newer: refuse the JSON and let the load below produce either the real + // graph or a real error. There is no reading of "I could not check" that + // justifies serving a possibly-superseded snapshot. + return false + } + return ji.ModTime().After(srcMod) +} + +// graphSourceModTime returns the commit mtime of the active graph desc +// describes, and ok=false when desc names no graph (GraphAbsent) or the graph +// cannot be stat'd. Split out so repoGraphBytes' gate reads as one comparison +// and each layout's commit point is named explicitly. +func graphSourceModTime(desc graph.GraphDescriptor) (time.Time, bool) { + switch desc.Kind { + case graph.GraphSingleFile: + fi, err := os.Stat(desc.Path) + if err != nil { + return time.Time{}, false + } + return fi.ModTime(), true + case graph.GraphSegmentSet: + // manifest.json is the segment set's atomic commit point; the gen dir is + // the fallback when the manifest itself cannot be stat'd. + fi, err := os.Stat(filepath.Join(desc.GenDir, graph.ManifestFileName)) + if err != nil { + if fi, err = os.Stat(desc.GenDir); err != nil { + return time.Time{}, false + } + } + return fi.ModTime(), true + default: // graph.GraphAbsent + return time.Time{}, false + } +} + func (liveStore) CreateGroup(name string) (GroupSummary, error) { if name == "" { return GroupSummary{}, errors.New("group name required") diff --git a/internal/dashboard/store_graphjson_staleness_6015_test.go b/internal/dashboard/store_graphjson_staleness_6015_test.go new file mode 100644 index 000000000..578d14807 --- /dev/null +++ b/internal/dashboard/store_graphjson_staleness_6015_test.go @@ -0,0 +1,433 @@ +package dashboard + +// store_graphjson_staleness_6015_test.go — #6015. +// +// repoGraphBytes read graph.json UNCONDITIONALLY and only fell back to the +// FlatBuffers graph when the JSON was absent. So any repo that ever produced a +// graph.json — via `--export-json`, or simply by predating the ADR-0016 layout +// — pinned the dashboard's RepoGraph / GroupGraph endpoints to that snapshot +// FOREVER, across every subsequent reindex, with nothing on the surface to say +// it was stale. Its sibling sidecar readers (graph/descriptions, graph/flows) +// already stat-compare before use; store.go was the odd one out. +// +// The fix is a stat gate, and these tests pin all three of its edges: +// +// - STALE: graph.fb newer than graph.json ⇒ serve the FB-derived graph. +// - FRESH: graph.json newer ⇒ serve its bytes VERBATIM. This is the +// performance half and it is not optional: on the reference corpus (427k +// entities / 1.85M relationships) regenerating from FB on every request +// would be a far worse regression than the staleness it cures. +// - TIE: equal mtimes ⇒ treat the JSON as stale. graph.fb is committed by an +// atomic rename (internal/atomicfile), which carries the temp file's mtime, +// and filesystem timestamp granularity is not guaranteed to be finer than +// the write. An `After` comparison — never `!Before` — is what stops a +// same-instant rebuild from being mistaken for "the JSON is still current". + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/cajasmota/grafel/internal/daemon" + "github.com/cajasmota/grafel/internal/graph" + fb "github.com/cajasmota/grafel/internal/graph/fbgraph" + "github.com/cajasmota/grafel/internal/graph/fbwriter" +) + +// staleJSONRepo is the marker baked into the graph.json fixture. It can never +// appear in an FB-derived result, so its presence/absence in the served bytes +// identifies WHICH source answered — the only thing these tests care about. +const staleJSONRepo = "stale-json-snapshot" + +// freshFBRepo is the marker baked into the graph.fb fixture. +const freshFBRepo = "fresh-fb-graph" + +// newGraphJSONFixtureRepo returns a repo path whose state dir exists and is +// isolated to this test, resolved through the SAME daemon.StateDirForRepo +// derivation production uses. +func newGraphJSONFixtureRepo(t *testing.T) (repoPath, stateDir string) { + t.Helper() + t.Setenv(daemon.EnvRoot, t.TempDir()) + repoPath = t.TempDir() + stateDir = daemon.StateDirForRepo(repoPath) + if stateDir == "" { + t.Fatal("StateDirForRepo returned empty") + } + if err := os.MkdirAll(stateDir, 0o755); err != nil { + t.Fatalf("mkdir state dir: %v", err) + } + return repoPath, stateDir +} + +// writeGraphJSONFixture writes a graph.json carrying staleJSONRepo, stamped +// with mtime. +// +// The bytes are deliberately INDENTED. A plain json.Marshal fixture would be +// byte-identical to what an unmarshal + re-marshal round trip produces, so the +// "served verbatim" assertions could not tell the raw-read fast path from a +// regeneration and would silently stop testing anything. Indentation is the +// cheapest property no re-marshal in this code path reproduces. +func writeGraphJSONFixture(t *testing.T, repoPath string, mtime time.Time) { + t.Helper() + p := daemon.GraphPathForRepo(repoPath) + b, err := json.MarshalIndent(&graph.Document{Repo: staleJSONRepo}, "", " ") + if err != nil { + t.Fatalf("marshal graph.json fixture: %v", err) + } + if err := os.WriteFile(p, b, 0o644); err != nil { + t.Fatalf("write graph.json: %v", err) + } + if err := os.Chtimes(p, mtime, mtime); err != nil { + t.Fatalf("chtimes graph.json: %v", err) + } +} + +// writeFlatGraphFBFixture writes a legacy flat graph.fb carrying freshFBRepo, +// stamped with mtime. Flat (not gen-pointer) on purpose: #6015 is live TODAY on +// flat graphs and does not need segments to manifest. +func writeFlatGraphFBFixture(t *testing.T, stateDir string, mtime time.Time) { + t.Helper() + p := filepath.Join(stateDir, "graph.fb") + doc := &graph.Document{Repo: freshFBRepo, Entities: []graph.Entity{ + {ID: "aa1", QualifiedName: "p.A", Kind: "function", Name: "A"}, + }} + if err := fbwriter.WriteAtomic(p, doc); err != nil { + t.Fatalf("write graph.fb: %v", err) + } + if err := os.Chtimes(p, mtime, mtime); err != nil { + t.Fatalf("chtimes graph.fb: %v", err) + } +} + +// servedRepo unmarshals whatever repoGraphBytes returned far enough to read the +// `repo` field, which identifies the source. +func servedRepo(t *testing.T, b []byte) string { + t.Helper() + var doc struct { + Repo string `json:"repo"` + } + if err := json.Unmarshal(b, &doc); err != nil { + t.Fatalf("unmarshal served graph: %v", err) + } + return doc.Repo +} + +// TestRepoGraphBytes_StaleGraphJSONIgnored is the bug itself: N reindexes have +// written a newer graph.fb and the dashboard is still handing out the old JSON. +func TestRepoGraphBytes_StaleGraphJSONIgnored(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + base := time.Now().Add(-time.Hour) + writeGraphJSONFixture(t, repoPath, base) + writeFlatGraphFBFixture(t, stateDir, base.Add(30*time.Minute)) + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + if got := servedRepo(t, b); got != freshFBRepo { + t.Errorf("served repo = %q, want %q: graph.fb is 30 minutes newer than graph.json, so the JSON "+ + "snapshot is a stale artifact of an earlier index and must not be served", got, freshFBRepo) + } +} + +// TestRepoGraphBytes_FreshGraphJSONServedVerbatim is the performance half. A +// gate that ignored graph.json unconditionally would "fix" staleness by +// re-marshalling a 427k-entity graph on every dashboard request. Byte-identity +// with the on-disk file is the assertion, because it proves NO regeneration +// happened — a re-marshal could never reproduce this fixture's content. +func TestRepoGraphBytes_FreshGraphJSONServedVerbatim(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + base := time.Now().Add(-time.Hour) + writeFlatGraphFBFixture(t, stateDir, base) + writeGraphJSONFixture(t, repoPath, base.Add(30*time.Minute)) + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + want, err := os.ReadFile(daemon.GraphPathForRepo(repoPath)) + if err != nil { + t.Fatalf("read graph.json: %v", err) + } + if string(b) != string(want) { + t.Errorf("served bytes are not the on-disk graph.json (served repo=%q): a graph.json NEWER than "+ + "graph.fb must be served as-is, or every request re-marshals the whole graph", + servedRepo(t, b)) + } +} + +// TestRepoGraphBytes_EqualMtimeTreatedAsStale: graph.fb lands by atomic rename, +// which preserves the temp file's mtime, and nothing guarantees the filesystem +// resolves the two writes to different timestamps. A `!Before` comparison would +// call this JSON current and serve a snapshot from a graph generation that has +// already been replaced. Ties must resolve to "regenerate". +func TestRepoGraphBytes_EqualMtimeTreatedAsStale(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + same := time.Now().Add(-time.Hour).Truncate(time.Second) + writeGraphJSONFixture(t, repoPath, same) + writeFlatGraphFBFixture(t, stateDir, same) + + ji, err := os.Stat(daemon.GraphPathForRepo(repoPath)) + if err != nil { + t.Fatal(err) + } + fi, err := os.Stat(filepath.Join(stateDir, "graph.fb")) + if err != nil { + t.Fatal(err) + } + if !ji.ModTime().Equal(fi.ModTime()) { + t.Fatalf("precondition: fixture mtimes differ (%v vs %v); this test is meaningless without a tie", + ji.ModTime(), fi.ModTime()) + } + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + if got := servedRepo(t, b); got != freshFBRepo { + t.Errorf("served repo = %q, want %q: on an mtime TIE the JSON must lose. The staleness check "+ + "cannot be fooled into serving a superseded snapshot by an equal-mtime write", got, freshFBRepo) + } +} + +// TestRepoGraphBytes_NoFBGraphServesJSON: a legacy repo with only a graph.json +// and no FB graph at all must keep working. The gate answers "is the JSON older +// than the graph?", and with no graph there is nothing for it to be older than. +// +// The assertion is byte-identity, not merely "the right content". Content alone +// cannot distinguish the two routes here: graph.LoadGraphFromDir ALSO falls back +// to graph.json when no FB graph exists, so a GraphAbsent branch that wrongly +// returned "not usable" would still produce the same graph — just via an +// unmarshal + re-marshal round trip on every single request. Byte-identity is +// what pins the fast path. +func TestRepoGraphBytes_NoFBGraphServesJSON(t *testing.T) { + repoPath, _ := newGraphJSONFixtureRepo(t) + writeGraphJSONFixture(t, repoPath, time.Now().Add(-time.Hour)) + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + if got := servedRepo(t, b); got != staleJSONRepo { + t.Fatalf("served repo = %q, want %q: with no FB graph on disk the JSON is the only graph there is", + got, staleJSONRepo) + } + want, err := os.ReadFile(daemon.GraphPathForRepo(repoPath)) + if err != nil { + t.Fatalf("read graph.json: %v", err) + } + if string(b) != string(want) { + t.Errorf("served bytes are not the on-disk graph.json: with no graph to be stale against, the " + + "JSON must take the raw-read fast path, not a re-marshal on every request") + } +} + +// TestRepoGraphBytes_SegmentSetKeepsJSONFastPathWhenManifestOlder pins the +// LAYOUT ASYMMETRY that graphJSONUsable's doc comment describes, so the comment +// is a checked claim rather than prose that can quietly go stale. +// +// cmd/grafel/index.go stamps graph.json and the FB artifact with one identical +// mtime. For a single-file repo the FB artifact IS the .fb file, so the tie +// resolves to "regenerate" and the raw-read fast path is disabled (that is +// TestRepoGraphBytes_EqualMtimeTreatedAsStale, and it is a real cost). For a +// SEGMENT SET the stamped artifact is the gen DIRECTORY while this gate reads +// manifest.json — written earlier, before the pointer flip — so the JSON is +// strictly newer and the fast path survives. Same indexer, two behaviours. +func TestRepoGraphBytes_SegmentSetKeepsJSONFastPathWhenManifestOlder(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + base := time.Now().Add(-time.Hour) + // Manifest older than graph.json: the shape the real writer produces, since + // index.go stamps the gen dir and not the manifest inside it. + writeDashboardSegmentSetFixture(t, stateDir, 5, base) + writeGraphJSONFixture(t, repoPath, base.Add(30*time.Minute)) + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + want, err := os.ReadFile(daemon.GraphPathForRepo(repoPath)) + if err != nil { + t.Fatalf("read graph.json: %v", err) + } + if string(b) != string(want) { + t.Errorf("segment-set repo with a manifest OLDER than graph.json did not take the raw-read fast "+ + "path (served repo=%q). The documented asymmetry between the single-file and segment-set "+ + "layouts no longer holds — re-check graphJSONUsable's comment against the code", + servedRepo(t, b)) + } +} + +// TestRepoGraphBytes_CorruptSegmentSetErrorsRatherThanServingStaleJSON: when a +// graph demonstrably exists but cannot be read, the honest answer is the error. +// Falling back to graph.json here would resurrect the exact failure mode #6015 +// is about — a stale snapshot served in place of a real problem, with the +// dashboard showing a plausible topology and no way for anyone to tell. +func TestRepoGraphBytes_CorruptSegmentSetErrorsRatherThanServingStaleJSON(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + writeGraphJSONFixture(t, repoPath, time.Now().Add(-time.Hour)) + + genDir := filepath.Join(stateDir, graph.GenDirName(9)) + if err := os.MkdirAll(genDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(genDir, graph.ManifestFileName), []byte("{ not json"), 0o644); err != nil { + t.Fatal(err) + } + if err := graph.WriteCurrentPointerRaw(stateDir, graph.GenDirName(9)); err != nil { + t.Fatal(err) + } + + b, err := repoGraphBytes(repoPath) + if err == nil { + t.Fatalf("repoGraphBytes returned bytes (repo=%q) for a repo whose active segment set is corrupt; "+ + "want an error, not a stale graph.json standing in for it", servedRepo(t, b)) + } +} + +// writeOldFormatGraphFB builds a valid graph.fb and patches its on-disk version +// scalar down, fabricating a graph written by an older grafel build. Mirrors +// internal/graph/reindex_required_test.go's helper of the same name. +func writeOldFormatGraphFB(t *testing.T, stateDir string, oldVersion int, mtime time.Time) { + t.Helper() + doc := &graph.Document{Repo: freshFBRepo, Entities: []graph.Entity{ + {ID: "ent0000000000000a", Name: "foo", Kind: "function", SourceFile: "a.go"}, + }} + buf, err := fbwriter.Marshal(doc) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if !fb.GetRootAsGraph(buf, 0).MutateVersion(int32(oldVersion)) { + t.Fatal("MutateVersion returned false — slot missing?") + } + p := filepath.Join(stateDir, "graph.fb") + if err := os.WriteFile(p, buf, 0o644); err != nil { + t.Fatalf("write graph.fb: %v", err) + } + if err := os.Chtimes(p, mtime, mtime); err != nil { + t.Fatalf("chtimes graph.fb: %v", err) + } +} + +// TestRepoGraphBytes_VersionSkewSurfacesReindexRequired covers the case the +// no-fallback rule made worse before it made better. +// +// After a grafel upgrade and before a reindex, a repo's graph.fb is +// version-incompatible. Previously the dashboard silently served graph.json — +// an older-but-honest graph, which is precisely the silently-stale behaviour +// #6015 removes. Now it refuses. But format skew is transient-by-reindex, not +// corruption, so refusing with a bare loader error puts an empty panel in front +// of the user with no indication that one command fixes it. The error must name +// the remedy. +func TestRepoGraphBytes_VersionSkewSurfacesReindexRequired(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + base := time.Now().Add(-time.Hour) + writeGraphJSONFixture(t, repoPath, base) + // Newer than the JSON, so the staleness gate correctly refuses the snapshot + // and the load is reached — and then fails on version skew. + writeOldFormatGraphFB(t, stateDir, 1, base.Add(30*time.Minute)) + + b, err := repoGraphBytes(repoPath) + if err == nil { + t.Fatalf("repoGraphBytes returned bytes (repo=%q) for a version-incompatible graph.fb; the stale "+ + "graph.json must not stand in for it", servedRepo(t, b)) + } + if !strings.Contains(err.Error(), "reindex required") { + t.Errorf("error does not tell the user what to do: %v\n"+ + "want it to carry the shared FormatVersionError wording ('reindex required'), so the panel "+ + "names the one action that fixes this instead of showing an opaque failure", err) + } +} + +// TestRepoGraphBytes_UnrelatedLoadErrorPassedThrough: only version skew gets the +// remedy wording. Dressing up every load failure as "reindex required" would +// send users to reindex a repo whose problem a reindex will not fix. +func TestRepoGraphBytes_UnrelatedLoadErrorPassedThrough(t *testing.T) { + repoPath, _ := newGraphJSONFixtureRepo(t) + // No graph.json and no graph at all: LoadGraphFromDir's "neither graph.fb + // nor graph.json found" must reach the caller unchanged. + _, err := repoGraphBytes(repoPath) + if err == nil { + t.Fatal("expected an error for a repo with no graph at all") + } + if strings.Contains(err.Error(), "reindex required") { + t.Errorf("a non-version-skew load failure was rendered as a reindex prompt: %v", err) + } +} + +// TestGraphSourceModTime_PrefersManifestOverGenDir pins WHICH file dates a +// segment set. manifest.json is its atomic commit point; the gen dir's own +// mtime moves whenever anything inside it is touched and is only a fallback. +// Reading the directory instead would date the graph by an unrelated write. +func TestGraphSourceModTime_PrefersManifestOverGenDir(t *testing.T) { + stateDir := t.TempDir() + manifestMod := time.Now().Add(-3 * time.Hour).Truncate(time.Second) + writeDashboardSegmentSetFixture(t, stateDir, 3, manifestMod) + + desc, err := graph.CurrentGraphDescriptor(stateDir) + if err != nil { + t.Fatalf("CurrentGraphDescriptor: %v", err) + } + if desc.Kind != graph.GraphSegmentSet { + t.Fatalf("precondition: descriptor kind = %v, want GraphSegmentSet", desc.Kind) + } + genInfo, err := os.Stat(desc.GenDir) + if err != nil { + t.Fatal(err) + } + if genInfo.ModTime().Equal(manifestMod) { + t.Skip("gen dir and manifest happen to share an mtime; nothing to distinguish") + } + + got, ok := graphSourceModTime(desc) + if !ok { + t.Fatal("graphSourceModTime: ok=false for a well-formed segment set") + } + if !got.Equal(manifestMod) { + t.Errorf("graphSourceModTime = %v, want the MANIFEST mtime %v (gen dir is %v): a segment set is "+ + "dated by its commit point, not by whatever last touched its directory", + got, manifestMod, genInfo.ModTime()) + } +} + +// TestGraphSourceModTime_NotOKWhenUndatable: "no graph" and "a graph I cannot +// stat" both report ok=false, which repoGraphBytes' gate treats as "do not +// serve the JSON". The single-file-with-missing-path shape is reachable only as +// a TOCTOU race in production (CurrentGraphDescriptor stats before returning), +// so it is pinned here at the helper rather than end-to-end. +func TestGraphSourceModTime_NotOKWhenUndatable(t *testing.T) { + if _, ok := graphSourceModTime(graph.GraphDescriptor{Kind: graph.GraphAbsent}); ok { + t.Error("GraphAbsent: ok=true, want false — there is no graph to date") + } + missing := graph.GraphDescriptor{Kind: graph.GraphSingleFile, Path: filepath.Join(t.TempDir(), "gone.fb")} + if _, ok := graphSourceModTime(missing); ok { + t.Error("GraphSingleFile with a vanished path: ok=true, want false") + } + emptySeg := graph.GraphDescriptor{Kind: graph.GraphSegmentSet, GenDir: filepath.Join(t.TempDir(), "gone")} + if _, ok := graphSourceModTime(emptySeg); ok { + t.Error("GraphSegmentSet with a vanished gen dir: ok=true, want false") + } +} + +// TestRepoGraphBytes_StaleAgainstSegmentSet: the same gate must hold when the +// active graph is a SEGMENT SET (graph./ + manifest.json) rather than a +// flat .fb. A gate hardcoded to stat graph.fb would find nothing, conclude +// "no graph", and serve the stale JSON — the exact silent-stale failure again, +// via the layout that #5915 makes the default. +func TestRepoGraphBytes_StaleAgainstSegmentSet(t *testing.T) { + repoPath, stateDir := newGraphJSONFixtureRepo(t) + base := time.Now().Add(-time.Hour) + writeGraphJSONFixture(t, repoPath, base) + writeDashboardSegmentSetFixture(t, stateDir, 7, base.Add(30*time.Minute)) + + b, err := repoGraphBytes(repoPath) + if err != nil { + t.Fatalf("repoGraphBytes: %v", err) + } + if got := servedRepo(t, b); got == staleJSONRepo { + t.Errorf("served the stale graph.json against a NEWER segment set: the freshness source must be " + + "the active graph descriptor, not a hardcoded graph.fb path") + } +} diff --git a/internal/statusfile/statusfile.go b/internal/statusfile/statusfile.go index 4094ea160..a33ad2b8f 100644 --- a/internal/statusfile/statusfile.go +++ b/internal/statusfile/statusfile.go @@ -165,6 +165,17 @@ type File struct { EngineInFlight int `json:"engine_in_flight,omitempty"` // EngineGroupAlgoInFlight mirrors indexstate.Snapshot.GroupAlgoInFlight. EngineGroupAlgoInFlight int `json:"engine_group_algo_in_flight,omitempty"` + // EngineRebuildInFlight is the number of GROUP REBUILDS the engine has + // running right now — the size of the engine-side per-group rebuild guard + // (daemon.rebuildWorker.active), one goroutine per group (#6014). + // + // It is deliberately NOT derived from indexstate: the engine's direct + // group-rebuild path (daemonRebuildFuncCore) bypasses the scheduler, so + // indexstate.InFlight — the source of EngineInFlight above — stays 0 for + // the whole of a group rebuild. Without this field a split-mode serve has + // no data at all from which to answer "is a rebuild running?", and + // StatusReply.RebuildInFlight is a structural zero. + EngineRebuildInFlight int `json:"engine_rebuild_in_flight,omitempty"` // EngineBusyStartedAt mirrors indexstate.Snapshot.StartedAt (zero when idle). EngineBusyStartedAt time.Time `json:"engine_busy_started_at,omitempty"` // ConcurrencyActive/Queued/Cap mirror indexstate.IndexConcurrency (#5493).