fix(analytics): recompute once the store has fully loaded (#2025)

Refs #2023, #1659, #1724

### Problem
`main.go:258` waits only for the first load chunk, then `main.go:402`
starts the analytics recomputers. `Start()` computes immediately on that
chunk (`analytics_recomputer.go:86` on master) and the next compute
waits a full interval (`:93`, 5 min default). The chunk loader walks by
ascending id, so that chunk holds the oldest transmissions.

- RF, topology, channels: the #1659 gate checked `LoadComplete()` after
the compute (`analytics_warmup_1659.go:122`). `LoadComplete` flips at
the end of the hot window (`chunked_load.go:489`), before the background
fill (`store.go:1455`), so the gate could open on a snapshot without the
background fill, and the 60 s force timeout (`:73`, `:162`) opened it on
the first-chunk snapshot. In an end-to-end test on master,
`/api/analytics/rf` returned 200 with 8 of 100 packets before the
background fill ran.
- Distance, hash-collisions, hash-sizes, roles, observers-clock-skew,
nodes-clock-skew: no gate, partial snapshot served from the start.
- Distance additionally served a snapshot from the previous index for up
to one interval after each lazy index build.

On a staging instance, a default analytics request returned 5,911
packets with hours-old last buckets until the next recompute (about
74k).

### Change
- `StartupLoadDone()` (`chunked_load.go:108`): closed when
`RunStartupLoad` returns, on every path (`chunked_load.go:202`). Closing
it drops the hash-size info cache (15 s TTL) and the clock-skew engine
throttle (30 s, `clock_skew.go:225`), both read by the post-load
computes.
- `recomputeWhenLoaded` (`analytics_recomputer.go:172`): on that signal,
recompute each recomputer once, sequentially, via `RecomputeNow`
(`:154`), which runs on the recomputer's own loop and restarts its
ticker (`:106`). Order (`:255`): rf, topology, channels, distance,
hash-collisions, hash-sizes, observers-clock-skew, nodes-clock-skew,
roles (roles reads the nodes-clock-skew snapshot). Logs one line with
per-recomputer durations.
- Warm-up gate: now the same signal (`:343-354`), sampled before the
compute starts (`:129`), so a pass that began on partial data never
opens it. 503 + `Retry-After: 5` and the force timeout are unchanged; a
forced-open snapshot is replaced by the post-load recompute.
- Ungated endpoints: no new 503s (their API has none); snapshot replaced
right after the load.
- Distance: the lazy index build refreshes the distance recomputer
before reporting built (`store.go:4476`).
- Recompute intervals and config unchanged.

### Performance
One extra compute per recomputer per process start, run sequentially so
they do not all hold the store read lock at once. Ticker phases
afterwards are offset by the cumulative post-load compute durations
instead of all starting within the first-chunk compute window (relevant
to #1724; the effect on lock waves is not measured).

### Tests
`analytics_recompute_after_load_test.go`: signal open during background
fill, closed after success and failure; cache drops; immediate and
ordered post-load recompute; gate not opened by a pass started before
the load; forced-open snapshot replaced on load; ticker restart;
distance refresh before 202 ends; end to end with recomputers started
before the background fill (RF 503 until load, then `totalTransmissions`
equals the full store; six ungated endpoints 200 during load; all nine
recomputed after load). 9 of these failed on master with stubs; 8
single-line mutations each caught. `go test ./...` in `cmd/server`
passes.

### Staging validation
Deployed together with the review follow-ups of #2015-#2023 (build
`c646310f`), container restart:

```
16:35:20 [store] first chunk ready (chunkSize=10000)
16:35:25 [store] LoadChunked complete ... starting background fill loader
16:36:58 [store] background load complete: 121120/121282 packets in memory (coverage=99.9%)
16:37:03 [analytics-recompute] startup load done: recomputed 10 snapshots in 5.155s (rf=955ms topology=1.684s channels=43ms distance=49ms hash-collisions=30ms hash-sizes=338ms observers-clock-skew=369ms nodes-clock-skew=692ms roles=2ms retransmissions=994ms)
```

Right after that line, `/api/analytics/rf` reported `totalTransmissions`
121,121 against 120,700 packets in memory, and the retransmissions
default shape from #2023 covered the full 7 days. Before this change
both waited for the next 5 minute tick.

### Merge order with #2023
#2023 adds a tenth recomputer. Whichever of the two merges second has to
add `recompRetransmissions` to `analyticsRecomputersLocked`, wire it to
`loadedGate` instead of `LoadComplete`, and change 9 to 10 in
`TestAnalyticsRecomputers_PostLoadOrder`; the retransmissions gate test
then calls `signalStartupLoadDone()` instead of setting `loadComplete`.
That resolution is what ran on staging above.

### Not verified
- Repeater-enrich recomputer and the region/window TTL caches
(hash-collisions region results have a 1 h TTL) may also keep partial
results after the load; not changed here.
- Recompute order is tested structurally, not with roles/clock-skew
data.
- Whether this reduces the #1724 stalls; not measured.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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

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

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

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

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;

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,

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