diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index a37242c9..1eebfe83 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -492,7 +492,8 @@ func (s *PacketStore) LoadChunked(chunkSize int) error { return s.packets[i].FirstSeen < s.packets[j].FirstSeen }) s.buildPathHopIndex() - s.buildDistanceIndex() + // Distance stays lazy, as in Load(): its first request builds the index + // and publishes the ready gate. An eager build here would be repeated. if s.hotStartupHours > 0 { s.oldestLoaded = hotCutoffStr } else if len(s.packets) > 0 { diff --git a/cmd/server/distance_lazy_index_test.go b/cmd/server/distance_lazy_index_test.go index f4c2b043..d680a0a4 100644 --- a/cmd/server/distance_lazy_index_test.go +++ b/cmd/server/distance_lazy_index_test.go @@ -1,6 +1,8 @@ package main import ( + "encoding/json" + "net/http" "net/http/httptest" "sync" "sync/atomic" @@ -40,6 +42,236 @@ func TestDistanceIndexNotBuiltOnLoad(t *testing.T) { } } +// #2227: production uses RunStartupLoad/LoadChunked, not the legacy Load. +// Keep startup's required indexes, but do no distance-index work until asked. +func TestLoadChunkedDistanceIndexRemainsLazy(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + // The existing fixture has two located nodes; this path makes an eager + // distance build observable instead of passing with empty distance data. + mustExecDB(t, db, `UPDATE observations SET path_json = '["aa","11"]' WHERE id = 1`) + store := NewPacketStore(db, nil) + if err := store.RunStartupLoad(2); err != nil { + t.Fatal(err) + } + if len(store.distHops) != 0 || len(store.distPaths) != 0 || store.DistanceIndexBuilt() { + t.Fatalf("chunked startup built the lazy distance index: %d hops, %d paths", len(store.distHops), len(store.distPaths)) + } + if !store.LoadComplete() || !store.SubpathIndexReady() || !store.PathHopIndexReady() || len(store.byPathHop) == 0 { + t.Fatal("chunked startup must retain path-hop entries and publish its required indexes") + } + stop := store.StartAnalyticsRecomputers(time.Hour) + defer stop() + // Drain startup's refresh so it cannot hide a missing lazy-build refresh. + waitForTest(t, "post-load distance snapshot", func() bool { return store.recompDistance.ComputeRuns() >= 2 }) + runsBeforeBuild := store.recompDistance.ComputeRuns() + + entered, release := make(chan struct{}, 1), make(chan struct{}) + var releaseOnce sync.Once + var builds atomic.Int32 + store.distanceBuildHook = func() { builds.Add(1); entered <- struct{}{}; <-release } + defer func() { + releaseOnce.Do(func() { close(release) }) + waitForTest(t, "lazy distance build", store.DistanceIndexBuilt) + }() + srv := NewServer(db, &Config{}, nil) + srv.store = store + request := func() *httptest.ResponseRecorder { + w := httptest.NewRecorder() + srv.handleAnalyticsDistance(w, httptest.NewRequest(http.MethodGet, "/api/analytics/distance", nil)) + return w + } + if w := request(); w.Code != http.StatusAccepted || w.Header().Get("Retry-After") == "" { + t.Fatalf("first distance response = %d, want 202 with Retry-After", w.Code) + } + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("first distance request did not start the lazy build") + } + if w := request(); w.Code != http.StatusAccepted { + t.Fatalf("distance response during build = %d, want 202", w.Code) + } + releaseOnce.Do(func() { close(release) }) + waitForTest(t, "lazy distance build", store.DistanceIndexBuilt) + w := request() + if w.Code != http.StatusOK { + t.Fatalf("distance response after build = %d, want 200", w.Code) + } + var body struct { + TopHops []DistanceHop `json:"topHops"` + TopPaths []DistancePath `json:"topPaths"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if len(body.TopHops) == 0 || len(body.TopPaths) == 0 { + t.Fatalf("ready response served an empty distance snapshot: %s", w.Body.String()) + } + if store.recompDistance.ComputeRuns() <= runsBeforeBuild { + t.Fatal("lazy build did not refresh the real distance recomputer before returning 200") + } + // A completed build must not be repeated by requests or an explicit trigger. + store.TriggerDistanceIndexBuild() + if w := request(); w.Code != http.StatusOK || !store.DistanceIndexBuilt() || builds.Load() != 1 { + t.Fatalf("ready distance index rebuilt: status=%d, builds=%d", w.Code, builds.Load()) + } +} + +func TestDistanceRequestDuringChunkedLoadWaitsForLaterPackets(t *testing.T) { + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + seedTestData(t, db) + mustExecDB(t, db, `UPDATE observations SET path_json = '["aa","11"]' WHERE transmission_id = 1`) + mustExecDB(t, db, `UPDATE observations SET path_json = '["11","aa"]' WHERE transmission_id = 3`) + store := NewPacketStore(db, nil) + firstChunk, releaseLoad, loadDone := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + store.OnChunkLoaded(func(_, total int) { + if total == 1 { + close(firstChunk) + <-releaseLoad + } + }) + var loadErr error + go func() { + defer close(loadDone) + loadErr = store.RunStartupLoad(1) + }() + t.Cleanup(func() { + releaseOnce.Do(func() { close(releaseLoad) }) + select { + case <-loadDone: + case <-time.After(3 * time.Second): + t.Error("chunked load did not finish") + } + }) + select { + case <-firstChunk: + case <-time.After(3 * time.Second): + t.Fatal("first chunk did not load") + } + if store.LoadComplete() || store.LoadProgress() != 1 { + t.Fatal("fixture must hold the loader after its first packet") + } + stop := store.StartAnalyticsRecomputers(time.Hour) + t.Cleanup(stop) + buildEntered := make(chan struct{}, 1) + var builds atomic.Int32 + store.distanceBuildHook = func() { builds.Add(1); buildEntered <- struct{}{} } + srv := NewServer(db, &Config{}, nil) + srv.store = store + request := func() *httptest.ResponseRecorder { + w := httptest.NewRecorder() + srv.handleAnalyticsDistance(w, httptest.NewRequest(http.MethodGet, "/api/analytics/distance", nil)) + return w + } + if w := request(); w.Code != http.StatusAccepted || w.Header().Get("Retry-After") == "" { + t.Fatalf("distance response while loading = %d, want 202 with Retry-After", w.Code) + } + select { + case <-buildEntered: + // Finish an erroneous early build while later chunks remain blocked, + // so it cannot race the loader and accidentally include the full data. + waitForTest(t, "premature distance build", store.DistanceIndexBuilt) + t.Error("distance build started before the initial chunks finished") + case <-time.After(100 * time.Millisecond): + } + if w := request(); w.Code != http.StatusAccepted { + t.Errorf("distance poll while loading = %d, want 202", w.Code) + } + releaseOnce.Do(func() { close(releaseLoad) }) + select { + case <-loadDone: + case <-time.After(3 * time.Second): + t.Fatal("chunked load did not finish") + } + if loadErr != nil { + t.Fatal(loadErr) + } + waitForTest(t, "post-load distance snapshot", func() bool { return store.recompDistance.ComputeRuns() >= 2 }) + runsBeforeBuild := store.recompDistance.ComputeRuns() + if w := request(); w.Code != http.StatusAccepted { + t.Errorf("first distance request after loading = %d, want 202", w.Code) + } + waitForTest(t, "full distance build", store.DistanceIndexBuilt) + w := request() + if w.Code != http.StatusOK { + t.Fatalf("distance response after build = %d, want 200", w.Code) + } + var body struct { + TopPaths []DistancePath `json:"topPaths"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + paths := make(map[string]bool, len(body.TopPaths)) + for _, path := range body.TopPaths { + if path.TotalDist > 0 { + paths[path.Hash] = true + } + } + if !paths["abc123def4567890"] || !paths["def456abc1230099"] { + t.Errorf("distance paths must include both first and later chunks: %v", paths) + } + if builds.Load() != 1 || store.recompDistance.ComputeRuns() <= runsBeforeBuild { + t.Errorf("want one full build followed by a fresh recompute; builds=%d, runs=%d -> %d", + builds.Load(), runsBeforeBuild, store.recompDistance.ComputeRuns()) + } +} + +func TestDistanceAfterCompletedHotWindowRemainsAvailable(t *testing.T) { + for _, tc := range []struct { + name string + done, failed bool + }{ + {"background pending", false, false}, + {"background failed", false, true}, + {"background failed and terminal", true, true}, + } { + t.Run(tc.name, func(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + mustExecDB(t, db, `UPDATE observations SET path_json = '["aa","11"]' WHERE id = 1`) + store := NewPacketStore(db, &PacketStoreConfig{HotStartupHours: 2, RetentionHours: 168}) + if err := store.LoadChunked(1); err != nil { + t.Fatal(err) + } + if !store.LoadComplete() { + t.Fatal("fixture must have a complete initial hot window") + } + // These background-fill states must not invalidate the loaded hot window. + store.backgroundLoadFailed.Store(tc.failed) + store.backgroundLoadDone.Store(tc.done) + srv := &Server{store: store} + request := func() *httptest.ResponseRecorder { + w := httptest.NewRecorder() + srv.handleAnalyticsDistance(w, httptest.NewRequest(http.MethodGet, "/api/analytics/distance", nil)) + return w + } + if w := request(); w.Code != http.StatusAccepted { + t.Fatalf("first distance request for a complete hot window = %d, want 202", w.Code) + } + waitForTest(t, "hot-window distance build", store.DistanceIndexBuilt) + w := request() + if w.Code != http.StatusOK { + t.Fatalf("distance response for a complete hot window = %d, want 200", w.Code) + } + var body struct { + TopPaths []DistancePath `json:"topPaths"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if len(body.TopPaths) == 0 { + t.Fatal("distance response must retain the loaded hot-window data") + } + }) + } +} + // TestDistanceFirstRequestReturns202: first /api/analytics/distance call // must trigger async build and return 202 + Retry-After. The handler must // NOT block for the full build. diff --git a/cmd/server/routes.go b/cmd/server/routes.go index c17a871f..41483253 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -2594,6 +2594,12 @@ func (s *Server) handleAnalyticsDistance(w http.ResponseWriter, r *http.Request) region := r.URL.Query().Get("region") area := r.URL.Query().Get("area") if s.store != nil { + // A failed initial load cannot finish the lazy build. A completed hot + // window remains usable even if its later background fill fails. + if s.store.backgroundLoadDone.Load() && s.store.backgroundLoadFailed.Load() && !s.store.LoadComplete() { + writeError(w, http.StatusInternalServerError, "Distance analytics unavailable: initial packet load failed") + return + } // Lazy build (#1011): distance index is not built at startup. // First request triggers an async build and gets 202 + // Retry-After; concurrent requests during the build window diff --git a/cmd/server/runstartup_load_test.go b/cmd/server/runstartup_load_test.go index fa7f22f7..9b665fb5 100644 --- a/cmd/server/runstartup_load_test.go +++ b/cmd/server/runstartup_load_test.go @@ -18,7 +18,10 @@ package main // before loadBackgroundChunks executes (so oldestLoaded is set) import ( + "encoding/json" "fmt" + "net/http" + "net/http/httptest" "path/filepath" "strings" "sync" @@ -262,6 +265,32 @@ func TestRunStartupLoad_LoadChunkedError_MidLoad(t *testing.T) { if !strings.Contains(store.BackgroundLoadError(), "LoadChunked failed") { t.Fatalf("BackgroundLoadError must mention LoadChunked failure; got %q", store.BackgroundLoadError()) } + if !store.firstChunkSignaled.Load() || store.LoadComplete() || store.LoadProgress() != 50 { + t.Fatal("fixture must fail after serving readiness for the first chunk") + } + srv := &Server{store: store} + for i := 0; i < 2; i++ { + w := httptest.NewRecorder() + srv.handleAnalyticsDistance(w, httptest.NewRequest(http.MethodGet, "/api/analytics/distance", nil)) + if w.Code != http.StatusInternalServerError { + t.Fatalf("distance request after terminal initial-load failure = %d, want 500; body=%s", w.Code, w.Body.String()) + } + if w.Header().Get("Retry-After") != "" { + t.Fatal("terminal initial-load failure must not ask the client to keep retrying") + } + var body struct { + Error string `json:"error"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body.Error != "Distance analytics unavailable: initial packet load failed" { + t.Fatalf("unexpected terminal distance error: %q", body.Error) + } + if strings.Contains(w.Body.String(), loadErr.Error()) || strings.Contains(w.Body.String(), store.BackgroundLoadError()) { + t.Fatal("distance error must not expose internal database failure details") + } + } } // TestLoadBackgroundChunks_PanicsOnOldestLoadedEmpty_Invariant covers the diff --git a/cmd/server/store.go b/cmd/server/store.go index 81ed86e7..fca450a0 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -4948,6 +4948,12 @@ func (s *PacketStore) DistanceIndexBuilding() bool { // the last build OR at most once per 5 minutes — whichever is more // restrictive. The first-ever build always runs. func (s *PacketStore) TriggerDistanceIndexBuild() { + // HTTP starts after the first chunk, but later initial chunks do not + // update this index. Keep requests on the 202 retry path until those + // chunks finish, without changing legacy Load() or manual stores. + if s.firstChunkSignaled.Load() && !s.LoadComplete() { + return + } s.distLazyMu.Lock() if s.distLazyBuilding { s.distLazyMu.Unlock()