perf(store): on-demand subpath counts instead of permanent indexes

cmd/server/analytics_cache_wiring_test.go

@@ -353,12 +353,19 @@ func TestHandleAnalyticsSubpaths_MaxLenFirmwareCeiling(t *testing.T) {
 	for i := range hops {
 		hops[i] = fmt.Sprintf("%02x", i)
 	}
+	// A time window counts every subpath length of each transmission (the
+	// unfiltered list stops at subpathMaxHops, #413), so the 63-hop path is
+	// asked for through ?window=.
+	pathJSON, _ := json.Marshal(hops)
 	srv.store.mu.Lock()
-	srv.store.spIndex[strings.Join(hops, ",")] = 1
+	srv.store.packets = append(srv.store.packets, &StoreTx{
+		ID: 1 << 30, Hash: "ceiling63", PathJSON: string(pathJSON),
+		FirstSeen: time.Now().UTC().Format(time.RFC3339),
+	})
 	srv.store.mu.Unlock()
 
 	get := func(maxLen string) []interface{} {
-		req := httptest.NewRequest(http.MethodGet, "/api/analytics/subpaths?minLen=63&maxLen="+maxLen, nil)
+		req := httptest.NewRequest(http.MethodGet, "/api/analytics/subpaths?minLen=63&window=24h&maxLen="+maxLen, nil)
 		w := httptest.NewRecorder()
 		router.ServeHTTP(w, req)
 		if w.Code != 200 {

cmd/server/channel_analytics_test.go

@@ -27,8 +27,6 @@ func newChannelTestStore(packets []*StoreTx) *PacketStore {
 		chanCache:     make(map[string]*cachedResult),
 		distCache:     make(map[string]*cachedResult),
 		subpathCache:  make(map[string]*cachedResult),
-		spIndex:       make(map[string]int),
-		spTxIndex:     make(map[string][]*StoreTx),
 		advertPubkeys: make(map[string]int),
 		clockSkew:     NewClockSkewEngine(),
 	}

cmd/server/chunked_load.go

@@ -528,7 +528,6 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 	sort.SliceStable(s.packets, func(i, j int) bool {
 		return s.packets[i].FirstSeen < s.packets[j].FirstSeen
 	})
-	s.buildSubpathIndex()
 	s.buildPathHopIndex()
 	s.buildDistanceIndex(nodeSnapshot)
 	if s.hotStartupHours > 0 {

cmd/server/chunked_load_fixes_test.go

@@ -24,17 +24,14 @@ func TestLoadChunked_MarksIndexesReady(t *testing.T) {
 	store := openChunkedTestStore(t, 100)
 	defer store.db.conn.Close()
 
-	if store.SubpathIndexReady() || store.PathHopIndexReady() {
+	if store.PathHopIndexReady() {
 		t.Fatal("indexes must start NOT ready")
 	}
 
 	if err := store.LoadChunked(50); err != nil {
 		t.Fatalf("LoadChunked: %v", err)
 	}
 
-	if !store.SubpathIndexReady() {
-		t.Fatal("SubpathIndexReady() must be true after LoadChunked builds the index")
-	}
 	if !store.PathHopIndexReady() {
 		t.Fatal("PathHopIndexReady() must be true after LoadChunked builds the index")
 	}

cmd/server/coverage_test.go

@@ -8,6 +8,7 @@ import (
 	"math"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
@@ -2290,48 +2291,36 @@ func TestStoreGetAnalyticsSubpaths(t *testing.T) {
 	})
 }
 
-func TestSubpathPrecomputedIndex(t *testing.T) {
+func TestSubpathCountsFromPackets(t *testing.T) {
 	db := setupRichTestDB(t)
 	defer db.Close()
 	store := NewPacketStore(db, nil)
 	store.Load()
-	// #1008: indexes built in background goroutine; wait before reading.
-	if !store.WaitIndexesReady(5 * time.Second) {
-		t.Fatal("indexes never became ready")
-	}
-
-	// After Load(), the precomputed index must be populated.
-	if len(store.spIndex) == 0 {
-		t.Fatal("expected spIndex to be populated after Load()")
-	}
-	if store.spTotalPaths == 0 {
-		t.Fatal("expected spTotalPaths > 0 after Load()")
-	}
 
 	// The rich test DB has paths ["aa","bb"], ["aabb","ccdd"], and
-	// ["eeff","0011","2233"].  That yields 5 unique raw subpaths.
+	// ["eeff","0011","2233"].  That yields 5 unique raw subpaths, counted
+	// on demand from the packets (#413).
 	expectedRaw := map[string]int{
 		"aa,bb":          1,
 		"aabb,ccdd":      1,
 		"eeff,0011":      1,
 		"0011,2233":      1,
 		"eeff,0011,2233": 1,
 	}
-	for key, want := range expectedRaw {
-		got, ok := store.spIndex[key]
-		if !ok {
-			t.Errorf("expected spIndex[%q] to exist", key)
-		} else if got != want {
-			t.Errorf("spIndex[%q] = %d, want %d", key, got, want)
-		}
+	fast := store.GetAnalyticsSubpaths("", 2, 8, 100)
+	gotRaw := map[string]int{}
+	for _, sp := range fast["subpaths"].([]map[string]interface{}) {
+		gotRaw[strings.Join(sp["rawHops"].([]string), ",")] += sp["count"].(int)
 	}
-	if store.spTotalPaths != 3 {
-		t.Errorf("spTotalPaths = %d, want 3", store.spTotalPaths)
+	if !reflect.DeepEqual(gotRaw, expectedRaw) {
+		t.Errorf("raw subpath counts = %v, want %v", gotRaw, expectedRaw)
+	}
+	if tp := fast["totalPaths"]; tp != 3 {
+		t.Errorf("totalPaths = %v, want 3", tp)
 	}
 
 	// Fast-path (no region) and slow-path (with region) must return the
 	// same shape.
-	fast := store.GetAnalyticsSubpaths("", 2, 8, 100)
 	slow := store.GetAnalyticsSubpaths("SJC", 2, 4, 50)
 	for _, r := range []map[string]interface{}{fast, slow} {
 		if _, ok := r["subpaths"]; !ok {
@@ -2341,41 +2330,15 @@ func TestSubpathPrecomputedIndex(t *testing.T) {
 			t.Error("missing totalPaths in result")
 		}
 	}
-
-	// Verify fast path totalPaths matches index.
-	if tp, ok := fast["totalPaths"].(int); ok && tp != store.spTotalPaths {
-		t.Errorf("fast totalPaths=%d, spTotalPaths=%d", tp, store.spTotalPaths)
-	}
 }
 
-func TestSubpathTxIndexPopulated(t *testing.T) {
+func TestSubpathDetailMatchCount(t *testing.T) {
 	db := setupRichTestDB(t)
 	defer db.Close()
 	store := NewPacketStore(db, nil)
 	store.Load()
-	// #1008: indexes built in background goroutine; wait before reading.
-	if !store.WaitIndexesReady(5 * time.Second) {
-		t.Fatal("indexes never became ready")
-	}
-
-	// spTxIndex must be populated alongside spIndex
-	if len(store.spTxIndex) == 0 {
-		t.Fatal("expected spTxIndex to be populated after Load()")
-	}
-
-	// Every key in spIndex must also exist in spTxIndex with matching count
-	for key, count := range store.spIndex {
-		txs, ok := store.spTxIndex[key]
-		if !ok {
-			t.Errorf("spTxIndex missing key %q that exists in spIndex", key)
-			continue
-		}
-		if len(txs) != count {
-			t.Errorf("spTxIndex[%q] has %d txs, spIndex count is %d", key, len(txs), count)
-		}
-	}
 
-	// GetSubpathDetail should return correct match count via indexed lookup
+	// GetSubpathDetail counts the transmissions holding the subpath
 	detail := store.GetSubpathDetail([]string{"eeff", "0011"})
 	if matches := detail.TotalMatches; matches != 1 {
 		t.Errorf("totalMatches = %d, want 1", matches)

cmd/server/eviction_test.go

@@ -19,7 +19,6 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore
 		byNode:        make(map[string][]*StoreTx),
 		nodeHashes:    make(map[string]map[string]bool),
 		byPayloadType: make(map[int][]*StoreTx),
-		spIndex:       make(map[string]int),
 		distHops:      make([]distHopRecord, 0),
 		distPaths:     make([]distPathRecord, 0),
 		rfCache:       make(map[string]*cachedResult),
@@ -83,9 +82,6 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore
 		store.distHops = append(store.distHops, distHopRecord{tx: tx, Hash: hash})
 		store.distPaths = append(store.distPaths, distPathRecord{tx: tx, Hash: hash})
 
-		// Subpath index
-		addTxToSubpathIndex(store.spIndex, tx)
-
 		// Track bytes for self-accounting
 		store.trackedBytes += rechargeTx(tx)
 		for _, obs := range tx.Observations {
@@ -261,7 +257,6 @@ func TestEvictStale_CleansResolvedPathNodeIndexes(t *testing.T) {
 		byNode:        make(map[string][]*StoreTx),
 		nodeHashes:    make(map[string]map[string]bool),
 		byPayloadType: make(map[int][]*StoreTx),
-		spIndex:       make(map[string]int),
 		distHops:      make([]distHopRecord, 0),
 		distPaths:     make([]distPathRecord, 0),
 		rfCache:       make(map[string]*cachedResult),
@@ -565,14 +560,11 @@ func TestEstimateStoreTxBytes(t *testing.T) {
 		PathJSON:    `["aa","bb"]`,
 	}
 	est := estimateStoreTxBytes(tx)
-	// Manual calculation: base + string lengths + index entries + path hops + subpaths
+	// Manual calculation: base + string lengths + index entries + path hops
 	hops := int64(len(txGetParsedPath(tx)))
 	manualCalc := int64(storeTxBaseBytes) + int64(len(tx.RawHex)+len(tx.Hash)+len(tx.DecodedJSON)+len(tx.PathJSON)) + int64(numIndexesPerTx*indexEntryBytes)
 	manualCalc += int64(decodedCacheFactor * len(tx.DecodedJSON))
 	manualCalc += hops * perPathHopBytes
-	if hops > 1 {
-		manualCalc += (hops * (hops - 1) / 2) * perSubpathEntryBytes
-	}
 	if est != manualCalc {
 		t.Fatalf("estimateStoreTxBytes = %d, want %d (manual calc)", est, manualCalc)
 	}

cmd/server/hash_migrate.go

@@ -125,9 +125,9 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration
 			continue
 		}
 
-		// A merge edits the subpath and path-hop indexes: let their
-		// background build finish first, or it could index a duplicate that
-		// was just removed.
+		// A merge edits the path-hop index: let its background build
+		// finish first, or it could index a duplicate that was just
+		// removed.
 		if !indexesReady {
 			store.WaitIndexesReady(30 * time.Second)
 			indexesReady = true
@@ -446,9 +446,6 @@ func (s *PacketStore) finishHashMerge(m *hashMerge, metadata pathNodeSnapshot) i
 		s.trackedBytes -= s.txChargedBytes(l)
 		s.untrackAdvertPubkey(l)
 		s.removeFromResolvedPubkeyIndex(l.ID)
-		if removeTxFromSubpathIndexFull(s.spIndex, s.spTxIndex, l) {
-			s.spTotalPaths--
-		}
 		delete(s.pathHopResolved, l)
 		delete(s.fallbackByNode, l)
 		if l.PayloadType != nil {

cmd/server/hash_migrate_log_race_301_test.go

@@ -8,28 +8,28 @@ import (
 	"time"
 )
 
-// hm301InstallSubpathGate holds the subpath index build until the returned
+// hm301InstallPathHopGate holds the path-hop index build until the returned
 // release runs. release is idempotent, so the caller can defer it right after
 // the gate goes in and still release it on purpose later: without that
 // deferred net a t.Fatal between the two (store.Load returning an error, say)
 // would leave the build goroutine blocked on the channel for the rest of the
 // run (#310, nit 3 of the #307 review).
-func hm301InstallSubpathGate(store *PacketStore) (release func()) {
+func hm301InstallPathHopGate(store *PacketStore) (release func()) {
 	gate := make(chan struct{})
-	store.subpathBuildGate = func() { <-gate }
+	store.pathHopBuildGate = func() { <-gate }
 	return sync.OnceFunc(func() { close(gate) })
 }
 
 // The deferred net and the deliberate release are the same func, so releasing
 // twice must be a no-op rather than a "close of closed channel" panic.
-func TestHashMigrateSubpathGateReleaseIsIdempotent_310(t *testing.T) {
+func TestHashMigrateBuildGateReleaseIsIdempotent_310(t *testing.T) {
 	store := &PacketStore{}
-	release := hm301InstallSubpathGate(store)
+	release := hm301InstallPathHopGate(store)
 
 	blocked := make(chan struct{})
 	go func() {
 		defer close(blocked)
-		store.subpathBuildGate()
+		store.pathHopBuildGate()
 	}()
 
 	release()
@@ -42,12 +42,12 @@ func TestHashMigrateSubpathGateReleaseIsIdempotent_310(t *testing.T) {
 	}
 }
 
-// Issue #301: Load() starts the two background index builds, and each builder
-// logs one more line ("index build complete") AFTER it has flipped its ready
-// flag. WaitIndexesReady returning therefore does not mean the builders are
-// done logging, and a builder of an earlier store can log at any later time.
+// Issue #301: Load() starts the background index build, and the builder logs
+// one more line ("index build complete") AFTER it has flipped its ready flag.
+// WaitIndexesReady returning therefore does not mean the builder is done
+// logging, and a builder of an earlier store can log at any later time.
 // The log capture of the hash-migrate harness must take those writes without a
-// data race: here the subpath builder is released inside the capture, so its
+// data race: here the path-hop builder is released inside the capture, so its
 // last line lands in the captured output while nothing orders it before the
 // harness's reads. Red under -race with an unsynchronized buffer.
 func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) {
@@ -59,13 +59,13 @@ func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) {
 	}
 	t.Cleanup(func() { db.conn.Close() })
 	store := NewPacketStore(db, &PacketStoreConfig{})
-	release := hm301InstallSubpathGate(store)
+	release := hm301InstallPathHopGate(store)
 	defer release() // a t.Fatal below must not strand the build goroutine
 	if err := store.Load(); err != nil {
 		t.Fatal(err)
 	}
 
-	const want = "index build complete: subpath"
+	const want = "index build complete: pathHop"
 	out := captureLogRead(func(read func() string) {
 		release()
 		if !store.WaitIndexesReady(30 * time.Second) {
@@ -82,13 +82,13 @@ func TestHashMigrate_LogCaptureToleratesIndexBuilderLogs_301(t *testing.T) {
 		deadline := time.Now().Add(30 * time.Second)
 		for !strings.Contains(read(), want) {
 			if time.Now().After(deadline) {
-				t.Error("the subpath builder's last line did not arrive")
+				t.Error("the path-hop builder's last line did not arrive")
 				return
 			}
 			runtime.Gosched()
 		}
 	})
 	if !strings.Contains(out, want) {
-		t.Fatalf("the subpath builder's last line was not captured:\n%s", out)
+		t.Fatalf("the path-hop builder's last line was not captured:\n%s", out)
 	}
 }

cmd/server/hash_migrate_readonly_215_test.go

@@ -163,8 +163,6 @@ func hm215OpenIdx(t testing.TB, d *hm215DB, resolvedIndex bool) *PacketStore {
 }
 
 // hm215Snapshot is the part of the store a leftover would show up in.
-// spTotalPaths is left out on purpose: eviction never decrements it (master
-// behaviour, unrelated to this change), so it cannot return to a baseline.
 type hm215Snapshot struct {
 	Packets       int
 	TrackedBytes  int64
@@ -181,8 +179,6 @@ type hm215Snapshot struct {
 	TotalObs      int
 	AdvertKeys    map[string]int
 	ByPathHop     map[string]int
-	SpIndex       map[string]int
-	SpTxIndex     map[string]int
 }
 
 func hm215Snap(s *PacketStore) hm215Snapshot {
@@ -196,7 +192,7 @@ func hm215Snap(s *PacketStore) hm215Snapshot {
 		PathHopRecs: len(s.pathHopResolved), FallbackRecs: len(s.fallbackByNode),
 		ResolvedIdx: len(s.resolvedPubkeyIndex), TotalObs: s.totalObs,
 		AdvertKeys: map[string]int{},
-		ByPathHop:  map[string]int{}, SpIndex: map[string]int{}, SpTxIndex: map[string]int{},
+		ByPathHop:  map[string]int{},
 	}
 	for pk, list := range s.byNode {
 		for _, tx := range list {
@@ -222,12 +218,6 @@ func hm215Snap(s *PacketStore) hm215Snapshot {
 	for k, list := range s.byPathHop {
 		sn.ByPathHop[k] = len(list)
 	}
-	for k, n := range s.spIndex {
-		sn.SpIndex[k] = n
-	}
-	for k, list := range s.spTxIndex {
-		sn.SpTxIndex[k] = len(list)
-	}
 	return sn
 }
 
@@ -461,17 +451,6 @@ func TestHashMigrate_MergesDuplicatesInMemory_215(t *testing.T) {
 					}
 				}
 			}
-			// The subpath total follows the merge: it counts transmissions with a
-			// path of 2+ hops, and a merged-away duplicate no longer counts.
-			multiHop := 0
-			for _, tx := range s.packets {
-				if len(txGetParsedPath(tx)) >= 2 {
-					multiHop++
-				}
-			}
-			if s.spTotalPaths != multiHop {
-				t.Errorf("spTotalPaths = %d, want %d (transmissions with 2+ hop paths)", s.spTotalPaths, multiHop)
-			}
 			// Every observation of the DB survives, except the 4 identical
 			// (obs0, ["aa"]) ones that merge into 1 per group.
 			if want := 2 /*ballast*/ + 2*(1+3+3) + 2; s.totalObs != want {