diff --git a/cmd/server/analytics_recompute_after_load_test.go b/cmd/server/analytics_recompute_after_load_test.go new file mode 100644 index 00000000..f18cbb41 --- /dev/null +++ b/cmd/server/analytics_recompute_after_load_test.go @@ -0,0 +1,454 @@ +package main + +// Analytics recomputers after the startup load. +// +// main.go starts the recomputers as soon as the FIRST load chunk is in +// memory. Their initial compute therefore only sees that chunk (the +// oldest ids), and the next compute used to wait a full interval +// (5 min by default). These tests pin the contract: +// +// - the store exposes a signal that fires only after RunStartupLoad +// has finished, background fill included; +// - every recomputer recomputes as soon as that signal fires; +// - a pass that started before the signal never ends the #1659 +// warm-up, and a snapshot served after the force timeout is +// replaced as soon as the load is done. + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/mux" +) + +func isClosedForTest(ch <-chan struct{}) bool { + select { + case <-ch: + return true + default: + return false + } +} + +func waitForTest(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("timed out waiting for: %s", what) +} + +// The startup-load signal must stay open while the background fill is +// still running (LoadComplete is already true at that point) and close +// once RunStartupLoad returns. +func TestStartupLoadDone_ClosesOnlyAfterBackgroundFill(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + createTestDBSpreadOverDays(t, dbPath, 100, 14, time.Now().UTC().Unix()) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + defer db.conn.Close() + store := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 14 * 24, HotStartupHours: 24}) + + var closedAtBgEntry, loadCompleteAtBgEntry atomic.Bool + store.bgLoaderEntryHook = func() { + closedAtBgEntry.Store(isClosedForTest(store.StartupLoadDone())) + loadCompleteAtBgEntry.Store(store.LoadComplete()) + } + if isClosedForTest(store.StartupLoadDone()) { + t.Fatal("StartupLoadDone closed before RunStartupLoad ran") + } + if err := store.RunStartupLoad(500); err != nil { + t.Fatalf("RunStartupLoad: %v", err) + } + if !loadCompleteAtBgEntry.Load() { + t.Fatal("fixture precondition: LoadComplete should already be true when the background fill starts") + } + if closedAtBgEntry.Load() { + t.Fatal("StartupLoadDone closed before the background fill ran") + } + if !isClosedForTest(store.StartupLoadDone()) { + t.Fatal("StartupLoadDone not closed after RunStartupLoad returned") + } +} + +// A failed load is terminal too: nothing more will be loaded, so the +// signal must close instead of leaving the recomputers waiting forever. +func TestStartupLoadDone_ClosesWhenLoadFails(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + createTestDBSpreadOverDays(t, dbPath, 10, 1, time.Now().UTC().Unix()) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + store := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 24, HotStartupHours: 1}) + db.conn.Close() + if err := store.RunStartupLoad(500); err == nil { + t.Fatal("fixture precondition: RunStartupLoad on a closed conn should fail") + } + if !isClosedForTest(store.StartupLoadDone()) { + t.Fatal("StartupLoadDone not closed after a failed RunStartupLoad") + } +} + +// Derived caches with their own TTL (node hash-size info 15s, clock-skew +// engine 30s) may hold values computed from the partial store. The +// recomputes that follow the signal read them, so they must be dropped. +func TestStartupLoadDone_DropsDerivedCaches(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + + store.hashSizeInfoMu.Lock() + store.hashSizeInfoCache = map[string]*hashSizeNodeInfo{"partial": {}} + store.hashSizeInfoAt = time.Now() + store.hashSizeInfoMu.Unlock() + store.clockSkew.mu.Lock() + store.clockSkew.lastComputed = time.Now() + store.clockSkew.mu.Unlock() + + store.signalStartupLoadDone() + + store.hashSizeInfoMu.Lock() + hashInfo := store.hashSizeInfoCache + store.hashSizeInfoMu.Unlock() + if hashInfo != nil { + t.Fatal("hash-size info cache survived the startup-load signal") + } + store.clockSkew.mu.RLock() + last := store.clockSkew.lastComputed + store.clockSkew.mu.RUnlock() + if !last.IsZero() { + t.Fatal("clock-skew engine still considers its pre-load result fresh") + } +} + +// The recomputer must not wait a full interval once the store is loaded. +func TestRecomputeWhenLoaded_RunsImmediately(t *testing.T) { + var loadedFlag atomic.Bool + rc := newAnalyticsRecomputer("t", time.Hour, func() interface{} { + if loadedFlag.Load() { + return "full" + } + return "partial" + }) + rc.Start() + defer rc.Stop() + if got := rc.Load(); got != "partial" { + t.Fatalf("initial snapshot = %v, want partial", got) + } + + loaded := make(chan struct{}) + stop := make(chan struct{}) + defer close(stop) + go recomputeWhenLoaded(loaded, stop, []*analyticsRecomputer{rc}) + + loadedFlag.Store(true) + close(loaded) + waitForTest(t, "snapshot recomputed after load", func() bool { return rc.Load() == "full" }) + if runs := rc.ComputeRuns(); runs != 2 { + t.Fatalf("ComputeRuns = %d, want 2 (initial + post-load)", runs) + } +} + +// Recomputers run one after the other, in slice order, so a recomputer +// that reads another one's snapshot (roles reads nodes-clock-skew) sees +// the post-load version. +func TestRecomputeWhenLoaded_RunsInOrder(t *testing.T) { + var loadedFlag atomic.Bool + var skewSnapshot atomic.Value + skew := newAnalyticsRecomputer("skew", time.Hour, func() interface{} { + v := "partial" + if loadedFlag.Load() { + v = "full" + } + time.Sleep(20 * time.Millisecond) + skewSnapshot.Store(v) + return v + }) + roles := newAnalyticsRecomputer("roles", time.Hour, func() interface{} { + return skewSnapshot.Load() + }) + skew.Start() + roles.Start() + defer skew.Stop() + defer roles.Stop() + + loaded := make(chan struct{}) + stop := make(chan struct{}) + defer close(stop) + done := make(chan struct{}) + go func() { + recomputeWhenLoaded(loaded, stop, []*analyticsRecomputer{skew, roles}) + close(done) + }() + loadedFlag.Store(true) + close(loaded) + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("recomputeWhenLoaded did not finish") + } + if got := roles.Load(); got != "full" { + t.Fatalf("roles snapshot = %v, want full (it ran before the skew recompute finished)", got) + } +} + +// The post-load order used by StartAnalyticsRecomputers: gated +// recomputers first (their 503 ends soonest), roles after +// nodes-clock-skew (roles reads that snapshot). +func TestAnalyticsRecomputers_PostLoadOrder(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + stop := store.StartAnalyticsRecomputers(time.Hour) + defer stop() + + store.analyticsRecomputerMu.RLock() + list := store.analyticsRecomputersLocked() + store.analyticsRecomputerMu.RUnlock() + pos := map[string]int{} + for i, rc := range list { + pos[rc.name] = i + } + if len(list) != 9 || len(pos) != 9 { + t.Fatalf("want 9 distinct recomputers, got %d (%d distinct)", len(list), len(pos)) + } + for _, name := range []string{"rf", "topology", "channels"} { + if pos[name] > 2 { + t.Errorf("%s at position %d, want among the first three", name, pos[name]) + } + } + if pos["nodes-clock-skew"] > pos["roles"] { + t.Errorf("roles (%d) runs before nodes-clock-skew (%d)", pos["roles"], pos["nodes-clock-skew"]) + } +} + +// A pass that STARTED before the load finished must not end the warm-up, +// even when the load finishes while that pass is still running. +func TestWarmup_PassStartedBeforeLoadDoesNotOpenGate(t *testing.T) { + loaded := make(chan struct{}) + releaseSecond := make(chan struct{}) + var calls atomic.Int32 + rc := newAnalyticsRecomputer("t", time.Hour, func() interface{} { + switch calls.Add(1) { + case 1: + close(loaded) // load finishes while the first pass is running + return "partial" + default: + <-releaseSecond + return "full" + } + }) + rc.setWarmupReadyGate_1659(func() bool { return isClosedForTest(loaded) }) + rc.Start() + defer rc.Stop() + + if !rc.IsWarmingUp_1659() { + t.Fatal("gate opened on a pass that started before the store was loaded") + } + + stop := make(chan struct{}) + defer close(stop) + go recomputeWhenLoaded(loaded, stop, []*analyticsRecomputer{rc}) + waitForTest(t, "post-load pass started", func() bool { return calls.Load() == 2 }) + if !rc.IsWarmingUp_1659() { + t.Fatal("gate opened while the only post-load pass is still running") + } + close(releaseSecond) + waitForTest(t, "gate opens after the post-load pass", func() bool { return !rc.IsWarmingUp_1659() }) + if got := rc.Load(); got != "full" { + t.Fatalf("gate open but snapshot = %v, want full", got) + } +} + +// Force-open timeout followed by the load finishing: the first-chunk +// snapshot that was served after the timeout is replaced right away. +func TestWarmup_ForceOpenedSnapshotReplacedOnLoad(t *testing.T) { + prev := warmupForceTimeout + warmupForceTimeout = 30 * time.Millisecond + defer func() { warmupForceTimeout = prev }() + + loaded := make(chan struct{}) + rc := newAnalyticsRecomputer("t", time.Hour, func() interface{} { + if isClosedForTest(loaded) { + return "full" + } + return "partial" + }) + rc.setWarmupReadyGate_1659(func() bool { return isClosedForTest(loaded) }) + rc.Start() + defer rc.Stop() + stop := make(chan struct{}) + defer close(stop) + go recomputeWhenLoaded(loaded, stop, []*analyticsRecomputer{rc}) + + waitForTest(t, "force timeout opens the gate", func() bool { return !rc.IsWarmingUp_1659() }) + if got := rc.Load(); got != "partial" { + t.Fatalf("fixture precondition: forced-open snapshot = %v, want partial", got) + } + if !rc.FirstPassDoneAt_1659().IsZero() { + t.Fatal("a pre-load pass was recorded as the first full pass") + } + + close(loaded) + waitForTest(t, "snapshot replaced after load", func() bool { return rc.Load() == "full" }) + waitForTest(t, "first full pass recorded", func() bool { return !rc.FirstPassDoneAt_1659().IsZero() }) +} + +// The periodic ticker restarts from the post-load compute, so the next +// periodic pass is a full interval after it instead of a redundant pass +// on the original phase a moment later. +func TestRecomputeNow_RestartsTicker(t *testing.T) { + const interval = 400 * time.Millisecond + rc := newAnalyticsRecomputer("t", interval, func() interface{} { return 1 }) + rc.Start() // run 1 at t0; unreset ticker would fire at t0+400ms + defer rc.Stop() + + time.Sleep(250 * time.Millisecond) + rc.RecomputeNow() // run 2 at ~250ms; next tick now ~650ms + time.Sleep(300 * time.Millisecond) // ~550ms: past the old phase, before the new one + if runs := rc.ComputeRuns(); runs != 2 { + t.Fatalf("ComputeRuns = %d at ~550ms, want 2 (ticker not restarted by RecomputeNow)", runs) + } +} + +// RecomputeNow must not hang when the recomputer is stopped. +func TestRecomputeNow_ReturnsWhenStopped(t *testing.T) { + rc := newAnalyticsRecomputer("t", time.Hour, func() interface{} { return 1 }) + rc.Start() + rc.Stop() + done := make(chan struct{}) + go func() { + rc.RecomputeNow() + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("RecomputeNow blocked on a stopped recomputer") + } +} + +// After a lazy distance-index build, the distance recomputer snapshot +// must already reflect the new index when the handler stops answering 202. +func TestDistanceIndexBuild_RefreshesRecomputerBeforeReportingBuilt(t *testing.T) { + db := setupRichTestDB(t) + defer db.Close() + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatalf("Load(): %v", err) + } + stop := store.StartAnalyticsRecomputers(time.Hour) + defer stop() + before := store.recompDistance.ComputeRuns() + + store.TriggerDistanceIndexBuild() + waitForTest(t, "distance index built", store.DistanceIndexBuilt) + if runs := store.recompDistance.ComputeRuns(); runs <= before { + t.Fatalf("distance index reported built while recomputer still serves the pre-build snapshot (runs %d -> %d)", before, runs) + } +} + +// End to end: recomputers started at the first chunk, background fill +// still pending. Gated endpoints answer 503 until the load is done and +// then serve data from the whole store without waiting for the interval. +// Ungated recomputers keep answering (no new 503) and are recomputed +// right after the load. +func TestAnalyticsRecomputers_FullDataRightAfterStartupLoad(t *testing.T) { + const totalRows = 100 + dbPath := filepath.Join(t.TempDir(), "test.db") + createTestDBSpreadOverDays(t, dbPath, totalRows, 14, time.Now().UTC().Unix()) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + defer db.conn.Close() + store := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 14 * 24, HotStartupHours: 24}) + + bgEntered := make(chan struct{}) + releaseBg := make(chan struct{}) + store.bgLoaderEntryHook = func() { + close(bgEntered) + <-releaseBg + } + loadErr := make(chan error, 1) + go func() { loadErr <- store.RunStartupLoad(500) }() + <-bgEntered + + stop := store.StartAnalyticsRecomputers(time.Hour) + defer stop() + + srv := NewServer(db, &Config{Port: 3000}, NewHub()) + srv.store = store + router := mux.NewRouter() + srv.RegisterRoutes(router) + get := func(path string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", path, nil)) + return w + } + + for _, p := range []string{"/api/analytics/rf", "/api/analytics/topology", "/api/analytics/channels"} { + if w := get(p); w.Code != http.StatusServiceUnavailable { + t.Fatalf("%s before full load: got %d, want 503", p, w.Code) + } + } + for _, p := range []string{"/api/analytics/hash-sizes", "/api/analytics/hash-collisions", "/api/analytics/roles"} { + if w := get(p); w.Code != http.StatusOK { + t.Fatalf("%s before full load: got %d, want 200 (no new warm-up 503)", p, w.Code) + } + } + + store.analyticsRecomputerMu.RLock() + all := map[string]*analyticsRecomputer{ + "topology": store.recompTopology, "rf": store.recompRF, "distance": store.recompDistance, + "channels": store.recompChannels, "hash-collisions": store.recompHashCollisions, + "hash-sizes": store.recompHashSizes, "roles": store.recompRoles, + "observers-clock-skew": store.recompObserversClockSkew, "nodes-clock-skew": store.recompNodesClockSkew, + } + store.analyticsRecomputerMu.RUnlock() + runsBefore := map[string]int64{} + for name, rc := range all { + runsBefore[name] = rc.ComputeRuns() + } + + close(releaseBg) + if err := <-loadErr; err != nil { + t.Fatalf("RunStartupLoad: %v", err) + } + store.mu.RLock() + inMemory := len(store.packets) + store.mu.RUnlock() + if inMemory != totalRows { + t.Fatalf("fixture precondition: %d packets in memory after load, want %d", inMemory, totalRows) + } + + waitForTest(t, "rf gate opens after full load", func() bool { return get("/api/analytics/rf").Code == http.StatusOK }) + var rf map[string]interface{} + if err := json.Unmarshal(get("/api/analytics/rf").Body.Bytes(), &rf); err != nil { + t.Fatalf("rf body: %v", err) + } + if got, _ := rf["totalTransmissions"].(float64); int(got) != totalRows { + t.Fatalf("rf totalTransmissions = %v right after full load, want %d (first-chunk snapshot still served)", rf["totalTransmissions"], totalRows) + } + for _, p := range []string{"/api/analytics/topology", "/api/analytics/channels"} { + if w := get(p); w.Code != http.StatusOK { + t.Fatalf("%s after full load: got %d, want 200", p, w.Code) + } + } + for name, rc := range all { + name, rc := name, rc + waitForTest(t, name+" recomputed after full load", func() bool { return rc.ComputeRuns() > runsBefore[name] }) + } +} diff --git a/cmd/server/analytics_recomputer.go b/cmd/server/analytics_recomputer.go index 57ff5f76..f0db263c 100644 --- a/cmd/server/analytics_recomputer.go +++ b/cmd/server/analytics_recomputer.go @@ -10,6 +10,9 @@ package main import ( + "fmt" + "log" + "strings" "sync" "sync/atomic" "time" @@ -34,9 +37,10 @@ type analyticsRecomputer struct { interval time.Duration compute func() interface{} - cache atomic.Value // holds interface{} — the latest snapshot - stop chan struct{} - done chan struct{} + cache atomic.Value // holds interface{} — the latest snapshot + stop chan struct{} + done chan struct{} + recomputeReq chan chan struct{} // RecomputeNow → loop; the loop closes the inner channel when done startOnce sync.Once stopOnce sync.Once @@ -61,11 +65,12 @@ func newAnalyticsRecomputer(name string, interval time.Duration, compute func() interval = 5 * time.Minute } return &analyticsRecomputer{ - name: name, - interval: interval, - compute: compute, - stop: make(chan struct{}), - done: make(chan struct{}), + name: name, + interval: interval, + compute: compute, + stop: make(chan struct{}), + done: make(chan struct{}), + recomputeReq: make(chan chan struct{}), } } @@ -96,6 +101,10 @@ func (r *analyticsRecomputer) loop() { select { case <-t.C: r.runOnce() + case ack := <-r.recomputeReq: + r.runOnce() + t.Reset(r.interval) + close(ack) case <-r.stop: return } @@ -114,6 +123,10 @@ func (r *analyticsRecomputer) runOnce() { // reach markFirstPassDone otherwise). _ = recover() }() + // Sample the #1659 readiness gate BEFORE computing: a pass that + // started on a partially loaded store must not end the warm-up, + // even if the load finishes while it runs. + ready := r.warmupReadyGateOpen_1659() t0 := time.Now() result := r.compute() r.lastComputeNs.Store(int64(time.Since(t0))) @@ -128,9 +141,53 @@ func (r *analyticsRecomputer) runOnce() { // PR #1688 r1: called on EVERY successful pass (even nil // result) so a compute that returns nil but doesn't panic // still lifts the gate — banner-stuck-forever fix (munger #2). - // The markFirstPassDone helper is idempotent and additionally - // consults the chunked-loader readiness gate (munger #5). - r.markFirstPassDone_1659() + if ready { + r.markFirstPassDone_1659() + } +} + +// RecomputeNow has the loop goroutine run a compute that starts after +// this call, and waits for it to finish. The periodic ticker restarts +// from that compute, so the next periodic pass is a full interval +// later. Returns early if the recomputer is stopped; blocks until Start +// if it has not started yet. +func (r *analyticsRecomputer) RecomputeNow() { + ack := make(chan struct{}) + select { + case r.recomputeReq <- ack: + case <-r.stop: + return + } + select { + case <-ack: + case <-r.stop: + } +} + +// recomputeWhenLoaded waits for loaded to close, then recomputes each +// recomputer once, one at a time and in slice order. Sequential so the +// post-load passes do not all hold the store read lock at once, and so +// a recomputer that reads another one's snapshot can be placed after +// it. Returns when done or when stop closes. +func recomputeWhenLoaded(loaded, stop <-chan struct{}, rcs []*analyticsRecomputer) { + select { + case <-loaded: + case <-stop: + return + } + t0 := time.Now() + parts := make([]string, 0, len(rcs)) + for _, rc := range rcs { + select { + case <-stop: + return + default: + } + rc.RecomputeNow() + parts = append(parts, fmt.Sprintf("%s=%s", rc.name, rc.LastComputeDuration().Round(time.Millisecond))) + } + log.Printf("[analytics-recompute] startup load done: recomputed %d snapshots in %s (%s)", + len(rcs), time.Since(t0).Round(time.Millisecond), strings.Join(parts, " ")) } // Load returns the most recently computed snapshot, or nil if Start @@ -190,6 +247,20 @@ func pickInterval(override, def time.Duration) time.Duration { return def } +// analyticsRecomputersLocked lists the analytics recomputers in the +// order they start and recompute after the startup load: the three +// warm-up-gated ones first, and roles after nodes-clock-skew because +// computeAnalyticsRoles reads that recomputer's snapshot +// (GetFleetClockSkew). Caller holds analyticsRecomputerMu. +func (s *PacketStore) analyticsRecomputersLocked() []*analyticsRecomputer { + return []*analyticsRecomputer{ + s.recompRF, s.recompTopology, s.recompChannels, + s.recompDistance, s.recompHashCollisions, s.recompHashSizes, + s.recompObserversClockSkew, s.recompNodesClockSkew, + s.recompRoles, + } +} + // StartAnalyticsRecomputers wires each analytics endpoint to a // background recompute goroutine. Each runs an initial compute // synchronously (so the first read after startup is a cache hit, never @@ -260,34 +331,50 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o "nodes-clock-skew", pickInterval(ov.NodesClockSkew, defaultInterval), func() interface{} { return s.computeFleetClockSkew() }, ) - all := []*analyticsRecomputer{ - s.recompTopology, s.recompRF, s.recompDistance, - s.recompChannels, s.recompHashCollisions, s.recompHashSizes, - s.recompRoles, - s.recompObserversClockSkew, s.recompNodesClockSkew, - } + all := s.analyticsRecomputersLocked() s.analyticsRecomputerMu.Unlock() - // Issue #1659 (PR #1688 r1, munger #5): wire the chunked-loader - // readiness gate on the three warmup-gated recomputers (RF, - // Topology, Channels). markFirstPassDone_1659 will refuse to - // flip first-pass-done until s.LoadComplete() reports true — - // i.e. the cold-load has populated all observations. Otherwise - // the FIRST recomputer pass runs against the post-restart in-RAM - // slice and the gate opens on partial data (the original #1659 - // bug class). - loadCompleteGate := s.LoadComplete - s.recompRF.setWarmupReadyGate_1659(loadCompleteGate) - s.recompTopology.setWarmupReadyGate_1659(loadCompleteGate) - s.recompChannels.setWarmupReadyGate_1659(loadCompleteGate) + // Issue #1659 (PR #1688 r1, munger #5): wire the loader readiness + // gate on the three warmup-gated recomputers (RF, Topology, + // Channels). Only a pass that STARTS after the whole startup load + // (hot window AND background fill) ends the warm-up. LoadComplete() + // is not enough: it flips at the end of the hot window, so the gate + // used to open on a snapshot that missed the background fill. + loaded := s.StartupLoadDone() + loadedGate := func() bool { + select { + case <-loaded: + return true + default: + return false + } + } + s.recompRF.setWarmupReadyGate_1659(loadedGate) + s.recompTopology.setWarmupReadyGate_1659(loadedGate) + s.recompChannels.setWarmupReadyGate_1659(loadedGate) for _, rc := range all { rc.Start() } + // main.go starts the recomputers at the first load chunk, so the + // initial computes above only saw part of the data. Recompute as + // soon as the load is done instead of a full interval later. + stopPostLoad := make(chan struct{}) + postLoadDone := make(chan struct{}) + go func() { + defer close(postLoadDone) + recomputeWhenLoaded(loaded, stopPostLoad, all) + }() + + var stopOnce sync.Once return func() { - for _, rc := range all { - rc.Stop() - } + stopOnce.Do(func() { + close(stopPostLoad) + for _, rc := range all { + rc.Stop() + } + <-postLoadDone + }) } } diff --git a/cmd/server/analytics_warmup_1659.go b/cmd/server/analytics_warmup_1659.go index 37585c12..8812d11f 100644 --- a/cmd/server/analytics_warmup_1659.go +++ b/cmd/server/analytics_warmup_1659.go @@ -21,12 +21,14 @@ // // 1. (#1688 munger #5) Only mark first-pass-done when BOTH: // a. a recomputer pass has completed, AND -// b. the chunked loader has finished (s.LoadComplete()). +// b. that pass started after the startup load finished, hot +// window and background fill (store.StartupLoadDone()). // The gate's `readyGate` callback is wired by -// StartAnalyticsRecomputers to `store.LoadComplete`. Passes that -// complete while loadComplete is still false leave the gate in -// the warming-up state; the NEXT pass after loadComplete flips -// true is the one that opens the gate. +// StartAnalyticsRecomputers to that signal. Passes that start +// before it leave the gate in the warming-up state; the loop +// recomputes as soon as the signal fires, and that pass opens +// the gate. (The gate used to be store.LoadComplete, which flips +// at the end of the hot window, before the background fill.) // // 2. (#1688 munger #2 + kent-beck #2) The gate MUST lift in bounded // time. If compute() panics on every pass, hangs indefinitely, @@ -40,7 +42,8 @@ // default) elapsed since the recomputer was constructed // forces IsWarmingUp_1659() to false — degraded mode // (serve whatever cache exists, possibly empty) is -// strictly better than a permanent 503. +// strictly better than a permanent 503. The partial snapshot +// served that way is replaced by the post-load recompute. // // Concurrency (#1688 munger #3): // @@ -104,24 +107,23 @@ func (r *analyticsRecomputer) loadWarmupReadyGate_1659() func() bool { return *p } +// warmupReadyGateOpen_1659 reports whether the readyGate (when set) +// is open. runOnce samples it BEFORE computing and only calls +// markFirstPassDone_1659 when it was open: first-pass-done requires a +// pass that started after the store finished loading (munger #5). A +// check after the compute would accept a pass that began on partial +// data and merely finished after the load. +func (r *analyticsRecomputer) warmupReadyGateOpen_1659() bool { + gate := r.loadWarmupReadyGate_1659() + return gate == nil || gate() +} + // markFirstPassDone_1659 is called from analyticsRecomputer.runOnce() -// after every compute attempt (success OR nil result; panics are -// caught upstream and never reach here). +// after a compute attempt (success OR nil result; panics are caught +// upstream and never reach here) that started with the readyGate open. // -// The gate flip is conditional on the readyGate (when set) reporting -// true — this implements the munger #5 fix: first-pass-done must -// require BOTH a recomputer pass complete AND the chunked loader to -// have finished populating the in-RAM observation set. -// -// Idempotent: only the FIRST successful flip wins; subsequent calls -// observe a non-zero firstPassDoneNs and return immediately. +// Idempotent: only the FIRST successful flip wins. func (r *analyticsRecomputer) markFirstPassDone_1659() { - if r.firstPassDoneNs.Load() != 0 { - return - } - if gate := r.loadWarmupReadyGate_1659(); gate != nil && !gate() { - return - } r.firstPassDoneNs.CompareAndSwap(0, time.Now().UnixNano()) } diff --git a/cmd/server/analytics_warmup_1659_test.go b/cmd/server/analytics_warmup_1659_test.go index e78c71d9..3a72130a 100644 --- a/cmd/server/analytics_warmup_1659_test.go +++ b/cmd/server/analytics_warmup_1659_test.go @@ -82,11 +82,12 @@ func TestAnalyticsRF_AfterFirstPassReturns200(t *testing.T) { db := setupTestDB(t) defer db.Close() store := NewPacketStore(db, nil) - // #1688 r1: the warmup gate now ALSO requires LoadComplete() to be - // true before first-pass-done flips (munger #5). Tests that don't - // exercise the chunked loader must flip it manually to model a - // production server that has finished cold-loading. - store.loadComplete.Store(true) + // #1688 r1: the warmup gate ALSO requires the startup load (hot + // window and background fill) to be done before first-pass-done + // flips (munger #5). Tests that don't exercise the loader must + // signal it manually to model a production server that has + // finished cold-loading. + store.signalStartupLoadDone() stop := store.StartAnalyticsRecomputers(50 * time.Millisecond) defer stop() diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 258061fd..84d96afc 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -96,9 +96,35 @@ func (s *PacketStore) OnChunkLoaded(fn func(rowsThisChunk, totalRows int)) { func (s *PacketStore) chunkedLoadInit() { s.chunkInitOnce.Do(func() { s.firstChunkReady = make(chan struct{}) + s.startupLoadDone = make(chan struct{}) }) } +// StartupLoadDone returns a channel closed once RunStartupLoad has +// returned: LoadChunked AND the background fill loader are finished, +// whether they succeeded or not. Nothing more is loaded from SQLite +// after it closes. LoadComplete() is not a substitute: it flips at the +// end of the hot window, before the background fill starts. +func (s *PacketStore) StartupLoadDone() <-chan struct{} { + s.chunkedLoadInit() + return s.startupLoadDone +} + +func (s *PacketStore) signalStartupLoadDone() { + s.chunkedLoadInit() + if !s.startupLoadSignaled.CompareAndSwap(false, true) { + return + } + // The analytics recomputes this signal triggers read these TTL + // caches. Drop what was computed from the partial store so they see + // the loaded data now rather than after the TTL. + s.hashSizeInfoMu.Lock() + s.hashSizeInfoCache = nil + s.hashSizeInfoMu.Unlock() + s.clockSkew.Invalidate() + close(s.startupLoadDone) +} + func (s *PacketStore) signalFirstChunk() { if s.firstChunkSignaled.CompareAndSwap(false, true) { close(s.firstChunkReady) @@ -153,6 +179,7 @@ func (s *PacketStore) fireChunkCallbacks(rowsThisChunk, totalRows int) { // LoadChunked. // // Steady-state contracts post-return: +// - StartupLoadDone() is closed, on every path below. // - LoadChunked error: backgroundLoadFailed=true, backgroundLoadDone // is also set true (terminal observable state — see dij #1). // backgroundLoadErr non-empty. Returns the error. @@ -172,6 +199,7 @@ func (s *PacketStore) fireChunkCallbacks(rowsThisChunk, totalRows int) { // parallelism while ensuring oldestLoaded has a valid floor when the // bg loader starts. func (s *PacketStore) RunStartupLoad(chunkSize int) error { + defer s.signalStartupLoadDone() // Clear any stale error from a previous invocation (single-call // invariant — see godoc above). Production never re-enters but // test fixtures may construct fresh stores that share no state; diff --git a/cmd/server/clock_skew.go b/cmd/server/clock_skew.go index 51bdf9cd..b545d400 100644 --- a/cmd/server/clock_skew.go +++ b/cmd/server/clock_skew.go @@ -220,6 +220,14 @@ func NewClockSkewEngine() *ClockSkewEngine { } } +// Invalidate makes the next Recompute run even if the last one is +// younger than computeInterval. +func (e *ClockSkewEngine) Invalidate() { + e.mu.Lock() + e.lastComputed = time.Time{} + e.mu.Unlock() +} + // Recompute recalculates all clock skew data from the packet store. // Called periodically or on demand. Holds store RLock externally. // Uses read-copy-update: heavy computation runs outside the write lock, diff --git a/cmd/server/store.go b/cmd/server/store.go index 959299f7..6d2f697f 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -471,6 +471,11 @@ type PacketStore struct { chunkCBMu sync.Mutex chunkCallbacks []func(rowsThisChunk, totalRows int) + // startupLoadDone is closed when RunStartupLoad returns (hot window + // plus background fill); see StartupLoadDone. + startupLoadDone chan struct{} + startupLoadSignaled atomic.Bool + // Eviction config and stats retentionHours float64 // 0 = unlimited maxMemoryMB int // 0 = unlimited (packet store memory budget) @@ -4460,6 +4465,17 @@ func (s *PacketStore) TriggerDistanceIndexBuild() { obsAtBuild := s.totalObs s.mu.Unlock() + // The distance recomputer's snapshot was computed from the index + // as it was before this build. Refresh it before reporting the + // index as built, so the handler does not go from 202 to serving + // that older snapshot for up to one recompute interval. + s.analyticsRecomputerMu.RLock() + rc := s.recompDistance + s.analyticsRecomputerMu.RUnlock() + if rc != nil { + rc.RecomputeNow() + } + s.distLazyMu.Lock() s.distLazyBuilding = false s.distLazyBuilt = true