diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index 3d8343db7..c0d5bd557 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -487,6 +487,7 @@ func (s *PacketStore) LoadChunked(chunkSize int) error { s.mu.Lock() for _, tx := range s.packets { pickBestObservation(tx) + s.trackedBytes += rechargeTx(tx) s.indexByNode(tx) } // Restore the "s.packets sorted oldest-first by FirstSeen" invariant @@ -621,7 +622,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold s.byPayloadType[pt] = append(s.byPayloadType[pt], tx) } s.trackAdvertPubkey(tx) - s.trackedBytes += estimateStoreTxBytes(tx) + s.trackedBytes += rechargeTx(tx) } if obsID.Valid { diff --git a/cmd/server/eviction_test.go b/cmd/server/eviction_test.go index fa79e6b57..3d8cdacc2 100644 --- a/cmd/server/eviction_test.go +++ b/cmd/server/eviction_test.go @@ -87,7 +87,7 @@ func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore addTxToSubpathIndex(store.spIndex, tx) // Track bytes for self-accounting - store.trackedBytes += estimateStoreTxBytes(tx) + store.trackedBytes += rechargeTx(tx) for _, obs := range tx.Observations { store.trackedBytes += estimateStoreObsBytes(obs) } @@ -568,6 +568,7 @@ func TestEstimateStoreTxBytes(t *testing.T) { // Manual calculation: base + string lengths + index entries + perTxMaps + path hops + subpaths 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 += perTxMapsBytes manualCalc += hops * perPathHopBytes if hops > 1 { @@ -587,8 +588,9 @@ func TestEstimateStoreObsBytes(t *testing.T) { PathJSON: `["aa"]`, } est := estimateStoreObsBytes(obs) - // storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96) = 300 - expected := int64(192 + 6 + 6 + 2*48) + // storeObsBaseBytes(192) + len(ObserverID=6) + len(PathJSON=6) + 2*48(96) + // + the dedup key: obsKeyEntryBytes + len(ObserverID) + len(PathJSON) + expected := int64(192+6+6+2*48) + int64(obsKeyEntryBytes+6+6) if est != expected { t.Fatalf("estimateStoreObsBytes = %d, want %d", est, expected) } diff --git a/cmd/server/main.go b/cmd/server/main.go index 12a2d2682..0c9cc2694 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -337,6 +337,7 @@ func main() { store.mu.Lock() for j := i; j < end && j < len(store.packets); j++ { pickBestObservation(store.packets[j]) + store.trackedBytes += rechargeTx(store.packets[j]) } store.mu.Unlock() if end < totalPackets { diff --git a/cmd/server/store.go b/cmd/server/store.go index f05b6c828..566294d95 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -54,7 +54,11 @@ type StoreTx struct { PathJSON string Direction string LatestSeen string // max observation timestamp (or FirstSeen if no observations) - UniqueObserverCount int // cached count of distinct observer IDs + UniqueObserverCount int32 // cached count of distinct observer IDs + // chargedBytes is what trackedBytes was last charged for this tx + // (observations and resolved relay hops excluded). It takes the 4 bytes + // UniqueObserverCount gave up, so StoreTx stays 320 bytes. + chargedBytes uint32 // Cached parsed fields (set once, read many) parsedPath []string // cached parsePathJSON result pathParsed bool // whether parsedPath has been set @@ -1022,7 +1026,7 @@ func (s *PacketStore) Load() error { s.byPayloadType[pt] = append(s.byPayloadType[pt], tx) } s.trackAdvertPubkey(tx) - s.trackedBytes += estimateStoreTxBytes(tx) + s.trackedBytes += rechargeTx(tx) } if obsID.Valid { @@ -1086,6 +1090,7 @@ func (s *PacketStore) Load() error { // now that pickBestObservation has propagated the best path. for _, tx := range s.packets { pickBestObservation(tx) + s.trackedBytes += rechargeTx(tx) s.indexByNode(tx) } @@ -1338,7 +1343,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { if txID > localMaxTxID { localMaxTxID = txID } - localTrackedBytes += estimateStoreTxBytes(tx) + localTrackedBytes += rechargeTx(tx) } if obsID.Valid { @@ -1420,6 +1425,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // Pick best observation for each local packet before merging. for _, tx := range localPackets { pickBestObservation(tx) + localTrackedBytes += rechargeTx(tx) } if len(localPackets) == 0 { @@ -1830,6 +1836,45 @@ func pickBestObservation(tx *StoreTx) { tx.pathParsed = false // invalidate cached parsed path } +// rechargeTx re-estimates tx and returns the change since it was last +// charged, for the caller to add to trackedBytes (#113). A tx is first +// charged when it is created, before its observations are merged, so its +// path costs (byPathHop, spTxIndex) are unknown then: call it again once +// pickBestObservation has set the path. Eviction subtracts what was charged +// (chargedBytes, see txChargedBytes) and never re-estimates, so the total +// stays exact however often the path changed in between. +// +// It reads and writes only tx, so it is safe on a tx that is not yet shared +// (loadChunk charges its local batch outside s.mu); otherwise call it with +// s.mu held. +func rechargeTx(tx *StoreTx) int64 { + est := estimateStoreTxBytes(tx) + if est > math.MaxUint32 { // far beyond any real tx; keeps the uint32 field honest + est = math.MaxUint32 + } + d := est - int64(tx.chargedBytes) + tx.chargedBytes = uint32(est) + return d +} + +// resolvedRelayBytes is the memory a tx holds for n distinct resolved relay +// keys (#164): per key a byPathHop slot, a hash in pathHopResolved and the +// other per-key index entries indexResolvedPathHops adds, plus the +// pathHopResolved map entry once. It is a function of n alone, so the charge +// can be re-derived from len(s.pathHopResolved[tx]) at eviction. +func resolvedRelayBytes(n int) int64 { + if n <= 0 { + return 0 + } + return perResolvedRecordBytes + int64(n)*perResolvedRelayBytes +} + +// txChargedBytes is everything trackedBytes currently holds for tx, observations +// excluded. Must be called with s.mu held. +func (s *PacketStore) txChargedBytes(tx *StoreTx) int64 { + return int64(tx.chargedBytes) + resolvedRelayBytes(len(s.pathHopResolved[tx])) +} + func pathLen(pathJSON string) int { if pathJSON == "" { return 0 @@ -3081,7 +3126,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac s.byPayloadType[pt] = append(s.byPayloadType[pt], tx) } s.trackAdvertPubkey(tx) - s.trackedBytes += estimateStoreTxBytes(tx) + s.trackedBytes += rechargeTx(tx) if _, exists := broadcastTxs[r.txID]; !exists { broadcastTxs[r.txID] = tx @@ -3158,6 +3203,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac // Pick best observation for new transmissions for _, tx := range broadcastTxs { pickBestObservation(tx) + s.trackedBytes += rechargeTx(tx) } // Incrementally update precomputed subpath index with new transmissions @@ -3605,6 +3651,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] } for _, tx := range updatedTxs { pickBestObservation(tx) + s.trackedBytes += rechargeTx(tx) } pathHopMutated := false for txID, tx := range updatedTxs { @@ -4518,6 +4565,9 @@ func (s *PacketStore) buildPathHopIndex() { func (s *PacketStore) retainResolvedPathHops(prev map[string][]*StoreTx) int { if len(prev) == 0 { // No resolved entries anywhere, so nothing is recorded as indexed. + for _, indexed := range s.pathHopResolved { + s.trackedBytes -= resolvedRelayBytes(len(indexed)) + } clear(s.pathHopResolved) return 0 } @@ -4527,8 +4577,9 @@ func (s *PacketStore) retainResolvedPathHops(prev map[string][]*StoreTx) int { } // Keep the per-transmission record in step with what is carried over // below: entries of transmissions that are gone are not (#158). - for tx := range s.pathHopResolved { + for tx, indexed := range s.pathHopResolved { if _, ok := live[tx]; !ok { + s.trackedBytes -= resolvedRelayBytes(len(indexed)) delete(s.pathHopResolved, tx) } } @@ -4724,6 +4775,10 @@ func (s *PacketStore) addResolvedPubkeysToPathHopIndex(tx *StoreTx, pubkeys []st return false } s.pathHopResolved[tx] = indexed + // Charge the new keys (#164): one byPathHop slot and one hash each, plus + // the map entry the first time. Eviction derives the same total back from + // len(indexed), see txChargedBytes. + s.trackedBytes += resolvedRelayBytes(len(indexed)) - resolvedRelayBytes(before) // Mutating byPathHop invalidates the batch relay-stats cache (#1164). s.invalidateRelayStatsCache() return true @@ -5118,6 +5173,33 @@ const ( // Per subpath entry in spTxIndex: string key + slice append + pointer perSubpathEntryBytes = 40 + + // ParsedDecoded caches json.Unmarshal of DecodedJSON as a + // map[string]interface{}, about 4x the JSON length. Analytics touch + // every tx, so it is charged up front rather than when the cache fills + // (charging in ParsedDecoded() would write trackedBytes from read paths + // that do not hold s.mu). + decodedCacheFactor = 4 + + // Per obs: the tx.obsKeys dedup entry. The "observerID|pathJSON" key is + // built by concatenation, so its bytes (added per obs in + // estimateStoreObsBytes) are a separate allocation from the obs fields; + // this is the string header, the bucket slot and the bool. + obsKeyEntryBytes = 16 + indexEntryBytes + 1 + + // Resolved relay hops (#164), charged per distinct key a tx is indexed + // under after insertion (see resolvedRelayBytes). Calibrated on this + // branch with TestTrackedBytesVsHeap_Measure_113: the marginal heap of a + // store with 1, 1.9 and 4 resolved keys per tx is 195, 264 and 449 B per + // tx at 100k transmissions and 275, 355 and 592 B at 500k (map growth + // steps), i.e. 110-170 B per pathHopResolved record plus 85-105 B per + // key. These are the midpoints. + perResolvedRelayBytes = 96 // byPathHop slot + pathHopResolved hash (both with append growth) + byNode slot, nodeHashes entry and resolved-pubkey index entry + perResolvedRecordBytes = 140 // pathHopResolved map entry: key, slice header, bucket share, first backing array + + // A typical tx is resolved to this many distinct relays. Used by + // estimateStoreTxBytesTypical only. + typicalResolvedRelays = 3 ) // estimateStoreTxBytes returns the estimated memory cost of a StoreTx (excluding observations). @@ -5126,6 +5208,7 @@ func estimateStoreTxBytes(tx *StoreTx) int64 { base := int64(storeTxBaseBytes) base += int64(len(tx.RawHex) + len(tx.Hash) + len(tx.DecodedJSON) + len(tx.PathJSON)) base += int64(numIndexesPerTx * indexEntryBytes) + base += int64(decodedCacheFactor * len(tx.DecodedJSON)) // Per-tx maps: obsKeys + observerSet base += perTxMapsBytes @@ -5150,13 +5233,16 @@ func estimateStoreTxBytesTypical(numObs int) int64 { // Typical tx: ~64 byte hash, ~200 byte decoded JSON, ~40 byte path, 3 hops base := int64(storeTxBaseBytes) + 64 + 200 + 40 base += int64(numIndexesPerTx * indexEntryBytes) + base += decodedCacheFactor * 200 base += perTxMapsBytes hops := int64(3) base += hops * perPathHopBytes base += (hops * (hops - 1) / 2) * perSubpathEntryBytes + base += resolvedRelayBytes(typicalResolvedRelays) // Add observation costs - obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 // observer ID + name + path + obsBase := int64(storeObsBaseBytes) + 30 + 30 + 60 + 25 // observer ID + name + path + timestamp obsBase += int64(numIndexesPerObs * indexEntryBytes) + obsBase += obsKeyEntryBytes + 30 + 60 // dedup key: observer ID + path // No per-obs ResolvedPath overhead (#800) base += int64(numObs) * obsBase return base @@ -5166,8 +5252,10 @@ func estimateStoreTxBytesTypical(numObs int) int64 { // ResolvedPath membership index overhead is tracked separately. func estimateStoreObsBytes(obs *StoreObs) int64 { base := int64(storeObsBaseBytes) - base += int64(len(obs.PathJSON) + len(obs.ObserverID)) + base += int64(len(obs.PathJSON) + len(obs.ObserverID) + len(obs.ObserverName) + + len(obs.ObserverIATA) + len(obs.Direction) + len(obs.RawHex) + len(obs.Timestamp)) base += int64(numIndexesPerObs * indexEntryBytes) + base += int64(obsKeyEntryBytes + len(obs.ObserverID) + len(obs.PathJSON)) // ResolvedPath field removed (#800) — no per-obs RP overhead return base } @@ -5224,7 +5312,7 @@ func (s *PacketStore) evictionCandidateTxIDs() []int { memCutoff := cutoffIdx for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark { tx := s.packets[memCutoff] - bytesToEvict += estimateStoreTxBytes(tx) + bytesToEvict += s.txChargedBytes(tx) for _, obs := range tx.Observations { bytesToEvict += estimateStoreObsBytes(obs) } @@ -5293,7 +5381,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { memCutoff := cutoffIdx for memCutoff < len(s.packets) && (s.trackedBytes-bytesToEvict) > lowWatermark { tx := s.packets[memCutoff] - bytesToEvict += estimateStoreTxBytes(tx) + bytesToEvict += s.txChargedBytes(tx) for _, obs := range tx.Observations { bytesToEvict += estimateStoreObsBytes(obs) } @@ -5342,7 +5430,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { delete(s.byHash, tx.Hash) delete(s.byTxID, tx.ID) evictedTxIDs[tx.ID] = struct{}{} - evictedBytes += estimateStoreTxBytes(tx) + evictedBytes += s.txChargedBytes(tx) for _, obs := range tx.Observations { delete(s.byObsID, obs.ID) diff --git a/cmd/server/tracked_bytes_accounting_113_test.go b/cmd/server/tracked_bytes_accounting_113_test.go new file mode 100644 index 000000000..fb416a8e3 --- /dev/null +++ b/cmd/server/tracked_bytes_accounting_113_test.go @@ -0,0 +1,589 @@ +package main + +import ( + "database/sql" + "encoding/json" + "fmt" + "math/rand" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +// Issues #113 and #164: trackedBytes (which drives packetStore.maxMemoryMB +// eviction) must equal what the store was charged, exactly. +// +// The invariant every test below checks: +// +// trackedBytes == Σ over live tx of ( tx.chargedBytes +// + resolvedRelayBytes(len(pathHopResolved[tx])) +// + Σ estimateStoreObsBytes(obs) ) +// +// and tx.chargedBytes == estimateStoreTxBytes(tx) for every live tx, i.e. no +// charge is stale. Master charged a tx before pickBestObservation gave it a +// path and never charged resolved relay hops, then re-estimated at eviction. + +// acct113Sum is what trackedBytes must equal: the charges of the live store. +func acct113Sum(s *PacketStore) (sum int64, stale int) { + for _, tx := range s.packets { + if int64(tx.chargedBytes) != estimateStoreTxBytes(tx) { + stale++ + } + sum += int64(tx.chargedBytes) + resolvedRelayBytes(len(s.pathHopResolved[tx])) + for _, o := range tx.Observations { + sum += estimateStoreObsBytes(o) + } + } + return sum, stale +} + +// acct113Check asserts the accounting invariant for the whole store. +func acct113Check(t *testing.T, s *PacketStore, where string) { + t.Helper() + s.mu.RLock() + defer s.mu.RUnlock() + sum, stale := acct113Sum(s) + if stale > 0 { + t.Errorf("%s: %d of %d tx carry a stale charge (chargedBytes != estimateStoreTxBytes)", where, stale, len(s.packets)) + } + if s.trackedBytes != sum { + t.Errorf("%s: trackedBytes = %d, charges of the live store sum to %d (drift %d)", where, s.trackedBytes, sum, s.trackedBytes-sum) + } +} + +func TestTrackedBytes_LoadChargesPathAndResolvedRelays_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 400, 0) + store := acct113Load(t, dbPath) + if len(store.pathHopResolved) == 0 { + t.Fatal("setup: no tx has resolved relay hops; the test would not cover #164") + } + multiHop := 0 + for _, tx := range store.packets { + if len(txGetParsedPath(tx)) > 1 { + multiHop++ + } + } + if multiHop == 0 { + t.Fatal("setup: no tx has a multi-hop path") + } + acct113Check(t, store, "after Load") +} + +func TestTrackedBytes_EvictionSubtractsExactlyWhatWasCharged_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 400, 200) + store := acct113Load(t, dbPath) + store.retentionHours = 24 + acct113Check(t, store, "after Load") + + if n := store.RunEviction(); n != 200 { + t.Fatalf("evicted %d tx, want the 200 older than the retention window", n) + } + // 200 tx stay: the remaining total must still be exactly their charges. + acct113Check(t, store, "after evicting the old half") + if store.trackedBytes <= 0 { + t.Fatal("trackedBytes hit zero with half of the store still live") + } + + store.retentionHours = 0.0001 // everything is now older than the window + if n := store.RunEviction(); n != 200 { + t.Fatalf("second pass evicted %d tx, want 200", n) + } + if store.trackedBytes != 0 || len(store.packets) != 0 || len(store.pathHopResolved) != 0 { + t.Fatalf("empty store: trackedBytes=%d packets=%d pathHopResolved=%d, want 0/0/0", + store.trackedBytes, len(store.packets), len(store.pathHopResolved)) + } +} + +// A tx whose estimate changed after it was charged (a path assigned later, +// or any other charged allocation) must still be removed at the amount it +// was charged: re-estimating at eviction is what made master drift. +func TestTrackedBytes_EvictionIgnoresAFreshEstimate_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 100, 50) + store := acct113Load(t, dbPath) + store.retentionHours = 24 + + victim := store.packets[0] // oldest: will be evicted + victim.PathJSON = `["aa","bb","cc","dd","ee","ff","11","22","33","44","55","66","77","88","99","00"]` + victim.pathParsed = false + if estimateStoreTxBytes(victim) == int64(victim.chargedBytes) { + t.Fatal("setup: the fresh estimate did not change") + } + + store.RunEviction() + // The live remainder: 50 tx that were never touched. + store.mu.RLock() + live, _ := acct113Sum(store) + got := store.trackedBytes + store.mu.RUnlock() + if got != live { + t.Fatalf("after eviction trackedBytes=%d, live charges=%d: eviction subtracted a fresh estimate, not the charge", got, live) + } +} + +func TestTrackedBytes_RechargeCoversALaterPath_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 40, 0) + store := acct113Load(t, dbPath) + + tx := store.packets[0] + long := &StoreObs{ID: 9_000_001, TransmissionID: tx.ID, ObserverID: "obs-late", ObserverName: "Late Observer", + PathJSON: `["01","02","03","04","05","06","07","08"]`, Timestamp: tx.FirstSeen} + store.mu.Lock() + tx.Observations = append(tx.Observations, long) + tx.ObservationCount++ + store.trackedBytes += estimateStoreObsBytes(long) + pickBestObservation(tx) + d := rechargeTx(tx) + store.trackedBytes += d + store.mu.Unlock() + + if d <= 0 { + t.Fatalf("an 8-hop path replaced a shorter one but the recharge delta is %d", d) + } + acct113Check(t, store, "after the longer path was assigned") + store.mu.Lock() + if d2 := rechargeTx(tx); d2 != 0 { + t.Errorf("a second recharge with nothing changed returned %d, want 0", d2) + } + store.mu.Unlock() +} + +func TestTrackedBytes_ResolvedRelaysAreChargedIncrementally_164(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 60, 0) + store := acct113Load(t, dbPath) + + // Start from a tx with no record of resolved relays. + tx := store.packets[0] + store.mu.Lock() + store.trackedBytes -= resolvedRelayBytes(len(store.pathHopResolved[tx])) + delete(store.pathHopResolved, tx) + store.mu.Unlock() + + hops := map[string]bool{} + step := func(keys ...string) int64 { + store.mu.Lock() + defer store.mu.Unlock() + before := store.trackedBytes + store.addResolvedPubkeysToPathHopIndex(tx, keys, hops) + return store.trackedBytes - before + } + k1, k2, k3 := acct113PK(201), acct113PK(202), acct113PK(203) + + d1 := step(k1, k2) + if d1 <= 0 { + t.Fatalf("two new resolved relays added %d bytes to trackedBytes", d1) + } + if want := resolvedRelayBytes(2); d1 != want { + t.Errorf("first two relays charged %d, want resolvedRelayBytes(2) = %d (slot + hash per key, entry overhead once)", d1, want) + } + if d := step(k1, k2); d != 0 { + t.Errorf("the same relays seen again (a later observation) charged %d, want 0", d) + } + d3 := step(k3) + if want := resolvedRelayBytes(3) - resolvedRelayBytes(2); d3 != want || d3 < 16 { + t.Errorf("a third relay charged %d, want %d (at least a byPathHop slot and a hash: 16)", d3, want) + } + acct113Check(t, store, "after adding resolved relays") + + // A rebuild keeps the records, so it must not move the total. + store.mu.Lock() + before := store.trackedBytes + store.buildPathHopIndex() + after := store.trackedBytes + store.mu.Unlock() + if after != before { + t.Errorf("buildPathHopIndex moved trackedBytes %d -> %d although every record was carried over", before, after) + } + acct113Check(t, store, "after a rebuild that retains the records") + + // A rebuild with no previous resolved entries clears the records: their + // charge must go with them. + store.mu.Lock() + store.byPathHop = map[string][]*StoreTx{} + store.buildPathHopIndex() + store.mu.Unlock() + if len(store.pathHopResolved) != 0 { + t.Fatalf("setup: the rebuild kept %d records", len(store.pathHopResolved)) + } + acct113Check(t, store, "after a rebuild that clears the records") +} + +// acct113StartupStore loads a store the way production starts: RunStartupLoad +// (LoadChunked, then, with hotStartupHours > 0, the loadChunk background fill) +// and not Load(). The accounting must hold on that path as well. +func acct113StartupStore(t *testing.T, dbPath string, hotStartupHours float64) *PacketStore { + t.Helper() + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + store := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 168, HotStartupHours: hotStartupHours}) + if err := store.RunStartupLoad(100); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(30 * time.Second) { + t.Fatal("background index builds did not finish") + } + return store +} + +func TestTrackedBytes_LoadChunkedChargesPathAndResolvedRelays_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 400, 0) + store := acct113StartupStore(t, dbPath, 0) + if len(store.packets) != 400 || len(store.pathHopResolved) == 0 { + t.Fatalf("setup: %d packets, %d resolved records; want 400 and some records", len(store.packets), len(store.pathHopResolved)) + } + acct113Check(t, store, "after LoadChunked") +} + +// A hot window through LoadChunked, then the older rows through loadChunk (the +// background fill): both halves must be charged, and the merge must not +// double-charge or lose a resolved record. +func TestTrackedBytes_BackgroundFillChargesPathAndResolvedRelays_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 400, 200) // 200 rows 3 days old, 200 rows 1 hour old + store := acct113StartupStore(t, dbPath, 2) + if len(store.packets) != 400 { + t.Fatalf("setup: %d packets in memory after the startup load, want 400 (hot window plus background fill)", len(store.packets)) + } + if len(store.pathHopResolved) == 0 { + t.Fatal("setup: no resolved records") + } + acct113Check(t, store, "after LoadChunked and the loadChunk background fill") + + store.retentionHours = 24 + if n := store.RunEviction(); n != 200 { + t.Fatalf("evicted %d tx, want the 200 old ones", n) + } + acct113Check(t, store, "after evicting the background-filled half") +} + +// The estimate terms are pinned one by one, not only through the tolerance of +// the heap ratio: every observation string, the dedup key, and the allowance +// per resolved record and per key. +func TestEstimateStoreObsBytes_CountsEveryTerm_113(t *testing.T) { + newObs := func() *StoreObs { + return &StoreObs{ + ObserverID: strings.Repeat("a", 3), ObserverName: strings.Repeat("b", 5), ObserverIATA: strings.Repeat("c", 7), + Direction: strings.Repeat("d", 11), RawHex: strings.Repeat("e", 13), Timestamp: strings.Repeat("f", 17), + PathJSON: strings.Repeat("g", 19), + } + } + obs := newObs() + want := int64(storeObsBaseBytes+numIndexesPerObs*indexEntryBytes) + + 3 + 5 + 7 + 11 + 13 + 17 + 19 + // every string field, once + int64(obsKeyEntryBytes) + 3 + 19 // the dedup key: observer ID + path + if got := estimateStoreObsBytes(obs); got != want { + t.Errorf("estimateStoreObsBytes = %d, want %d", got, want) + } + // Each string must move the estimate by exactly its length. + for name, set := range map[string]func(o *StoreObs){ + "ObserverName": func(o *StoreObs) { o.ObserverName += "x" }, + "ObserverIATA": func(o *StoreObs) { o.ObserverIATA += "x" }, + "Direction": func(o *StoreObs) { o.Direction += "x" }, + "RawHex": func(o *StoreObs) { o.RawHex += "x" }, + "Timestamp": func(o *StoreObs) { o.Timestamp += "x" }, + } { + longer := newObs() + set(longer) + if d := estimateStoreObsBytes(longer) - estimateStoreObsBytes(obs); d != 1 { + t.Errorf("one more byte of %s moved the estimate by %d, want 1", name, d) + } + } + // ObserverID and PathJSON are charged twice: as the field and in the dedup key. + for name, set := range map[string]func(o *StoreObs){ + "ObserverID": func(o *StoreObs) { o.ObserverID += "x" }, + "PathJSON": func(o *StoreObs) { o.PathJSON += "x" }, + } { + longer := newObs() + set(longer) + if d := estimateStoreObsBytes(longer) - estimateStoreObsBytes(obs); d != 2 { + t.Errorf("one more byte of %s moved the estimate by %d, want 2 (field and dedup key)", name, d) + } + } +} + +func TestResolvedRelayBytes_AllowanceHasARecordAndAKeyPart_164(t *testing.T) { + if resolvedRelayBytes(0) != 0 || resolvedRelayBytes(-1) != 0 { + t.Fatal("a tx without a record is charged nothing") + } + perKey := resolvedRelayBytes(2) - resolvedRelayBytes(1) + record := resolvedRelayBytes(1) - perKey + for n := 1; n <= 6; n++ { + if want := record + int64(n)*perKey; resolvedRelayBytes(n) != want { + t.Errorf("resolvedRelayBytes(%d) = %d, want one record (%d) plus %d keys of %d = %d", n, resolvedRelayBytes(n), record, n, perKey, want) + } + } + // Physical floors: a key costs at least a byPathHop slot and a hash (16), + // a record at least its map key and slice header (8 + 24). + if perKey < 16 { + t.Errorf("a resolved key is charged %d bytes, less than a byPathHop slot plus a hash (16)", perKey) + } + if record < 32 { + t.Errorf("a pathHopResolved record is charged %d bytes, less than its map key plus slice header (32)", record) + } +} + +// A record whose tx is not (or no longer) in s.packets is dropped by a +// rebuild; its charge must go with it. loadChunk indexes a batch's resolved +// relays before the batch is published into s.packets, so the window exists. +func TestTrackedBytes_RebuildCreditsRecordsOfTxNotInTheStore_164(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 40, 0) + store := acct113Load(t, dbPath) + + ghost := &StoreTx{ID: 99999, Hash: "ghost", PathJSON: `["aa"]`, obsKeys: map[string]bool{}, observerSet: map[string]bool{}} + hops := map[string]bool{} + store.mu.Lock() + before := store.trackedBytes + store.addResolvedPubkeysToPathHopIndex(ghost, []string{acct113PK(210), acct113PK(211)}, hops) + charged := store.trackedBytes - before + store.mu.Unlock() + if charged != resolvedRelayBytes(2) { + t.Fatalf("the ghost's two relays were charged %d, want %d", charged, resolvedRelayBytes(2)) + } + + store.mu.Lock() + store.buildPathHopIndex() + store.mu.Unlock() + if _, ok := store.pathHopResolved[ghost]; ok { + t.Fatal("setup: the rebuild kept the record of a tx that is not in the store") + } + if store.trackedBytes != before { + t.Errorf("after the rebuild trackedBytes=%d, want %d: the dropped record's charge stayed", store.trackedBytes, before) + } + acct113Check(t, store, "after dropping a record of a tx outside the store") +} + +// The candidates evictionCandidateTxIDs hands out (the batch it prefetches +// resolved pubkeys for) must be exactly the tx the pass then evicts, for a +// memory-triggered pass as for a retention pass. +func TestEvictionCandidatesMatchWhatIsEvicted_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 3000, 0) + store := acct113Load(t, dbPath) + store.maxMemoryMB = int(store.trackedBytes / 1048576) + + store.mu.Lock() + candidates := map[int]bool{} + for _, id := range store.evictionCandidateTxIDs() { + candidates[id] = true + } + live := map[int]bool{} + for _, tx := range store.packets { + live[tx.ID] = true + } + store.mu.Unlock() + if len(candidates) == 0 { + t.Fatal("setup: no eviction candidates") + } + store.RunEviction() + evicted := map[int]bool{} + store.mu.RLock() + for _, tx := range store.packets { + delete(live, tx.ID) + } + store.mu.RUnlock() + for id := range live { + evicted[id] = true + } + if len(evicted) != len(candidates) { + t.Fatalf("%d candidates, %d evicted", len(candidates), len(evicted)) + } + for id := range evicted { + if !candidates[id] { + t.Fatalf("tx %d was evicted but was not a candidate", id) + } + } +} + +// Insert, extra observations, path changes, resolved relays and eviction in a +// deterministic mix: the invariant holds after every step, and an emptied +// store is exactly zero. +func TestTrackedBytes_MixedSequenceDoesNotDrift_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 200, 0) + store := acct113Load(t, dbPath) + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + rng := rand.New(rand.NewSource(113)) + maxTx, maxObs := 200, 0 + if err := conn.QueryRow(`SELECT COALESCE(MAX(id),0) FROM observations`).Scan(&maxObs); err != nil { + t.Fatal(err) + } + hops := map[string]bool{} + now := time.Now().UTC() + for step := 0; step < 60; step++ { + switch rng.Intn(4) { + case 0: // new transmissions + firstNew, firstNewObs := maxTx, maxObs + for k := 0; k < 3; k++ { + maxTx++ + ts := now.Add(time.Duration(step*10+k) * time.Second).Format(time.RFC3339) + if _, err := conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, 'aabb', ?, ?, 1, 5, 1, '{"type":"GRP_TXT","text":"live"}')`, maxTx, fmt.Sprintf("live%06d", maxTx), ts); err != nil { + t.Fatal(err) + } + maxObs++ + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs0', 'Alpha', 'RX', ?, ?)`, maxObs, maxTx, `["aa","bb"]`, ts); err != nil { + t.Fatal(err) + } + } + _ = firstNewObs + store.IngestNewFromDB(firstNew, 100) + case 1: // a longer path heard on an existing transmission + id := 1 + rng.Intn(200) + maxObs++ + p := make([]string, 2+rng.Intn(6)) + for i := range p { + p[i] = acct113Hop(id, i, step) + } + pj, _ := json.Marshal(p) + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs3', 'Delta', 'RX', ?, ?)`, maxObs, id, string(pj), now.Format(time.RFC3339)); err != nil { + t.Fatal(err) + } + store.IngestNewObservations(maxObs-1, 10) + case 2: // relays resolved after insertion + store.mu.Lock() + if len(store.packets) > 0 { + tx := store.packets[rng.Intn(len(store.packets))] + store.addResolvedPubkeysToPathHopIndex(tx, []string{acct113PK(rng.Intn(256)), acct113PK(rng.Intn(256))}, hops) + } + store.mu.Unlock() + case 3: // age the head of the store out + store.mu.Lock() + for i := 0; i < len(store.packets) && i < 5; i++ { + store.packets[i].FirstSeen = now.Add(-100 * time.Hour).Format(time.RFC3339) + } + store.retentionHours = 24 + store.mu.Unlock() + store.RunEviction() + } + acct113Check(t, store, fmt.Sprintf("after step %d", step)) + if t.Failed() { + return + } + } + store.mu.Lock() + for _, tx := range store.packets { // live tx carry timestamps from the future + tx.FirstSeen = now.Add(-100 * time.Hour).Format(time.RFC3339) + } + store.retentionHours = 24 + store.mu.Unlock() + store.RunEviction() + if store.trackedBytes != 0 || len(store.packets) != 0 { + t.Fatalf("emptied store: trackedBytes=%d packets=%d, want 0/0", store.trackedBytes, len(store.packets)) + } +} + +// A memory-triggered pass must start above the high watermark and finish at +// or below the low one, without taking much more than needed. +func TestMemoryEviction_CrossesHighAndReachesLow_113(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 3000, 0) + store := acct113Load(t, dbPath) + acct113Check(t, store, "after Load") + + tracked := store.trackedBytes + // Just below the tracked size: one pass needs about 15-24% of the store, + // under the 25% safety cap. + store.maxMemoryMB = int(tracked / 1048576) + if store.maxMemoryMB < 4 { + t.Fatalf("setup: only %d bytes tracked; the store is too small for the watermark arithmetic", tracked) + } + high := int64(store.maxMemoryMB) * 1048576 + low := int64(float64(high) * 0.85) + if tracked <= high { + t.Fatalf("setup: trackedBytes %d does not exceed the high watermark %d", tracked, high) + } + before := len(store.packets) + store.RunEviction() + if store.trackedBytes > low { + t.Errorf("after the pass trackedBytes=%d is above the low watermark %d", store.trackedBytes, low) + } + if slack := int64(float64(tracked)/float64(before)) * 8; store.trackedBytes < low-slack { + t.Errorf("trackedBytes=%d is far below the low watermark %d: the pass evicted too much", store.trackedBytes, low) + } + if len(store.packets) >= before { + t.Fatal("nothing was evicted") + } + acct113Check(t, store, "after the memory pass") +} + +// The cold-load budget is the estimate of a typical tx; it must stay in line +// with what the store really charges for a tx of that shape: 64-byte hash, +// 200-byte decoded JSON, a 3-hop path, three observations with a 30-byte +// observer ID and name, a 60-byte path and a 25-byte timestamp, and the +// typical number of resolved relays. +func TestEstimateStoreTxBytesTypical_TracksTheRealCharge_113(t *testing.T) { + const numObs = 3 + rep := func(c string, n int) string { return strings.Repeat(c, n) } + tx := &StoreTx{ + ID: 1, Hash: rep("h", 64), DecodedJSON: rep("d", 200), PathJSON: `["aa","bb","cc"]` + rep(" ", 24), + obsKeys: map[string]bool{}, observerSet: map[string]bool{}, + } + tx.PathJSON = tx.PathJSON[:40] + total := rechargeTx(tx) + for i := 0; i < numObs; i++ { + total += estimateStoreObsBytes(&StoreObs{ + ObserverID: rep("o", 30), ObserverName: rep("n", 30), PathJSON: rep("p", 60), Timestamp: rep("t", 25), + }) + } + total += resolvedRelayBytes(typicalResolvedRelays) + typical := float64(estimateStoreTxBytesTypical(numObs)) + if typical < float64(total)*0.9 || typical > float64(total)*1.1 { + t.Errorf("estimateStoreTxBytesTypical(%d) = %.0f, a tx of that shape is charged %d: more than 10%% apart", numObs, typical, total) + } + if got := len(txGetParsedPath(tx)); got != 3 { + t.Fatalf("setup: the typical tx has %d hops, want 3", got) + } +} + +// trackedBytes against the real heap after a GC, on a store loaded through +// Load() with the decode cache filled the way analytics fill it. The store's +// own maps and slices are what trackedBytes is meant to bound; the ratio +// must be close to 1 (master: 0.6-0.7 on the same data). +func TestTrackedBytesTracksTheHeap_113(t *testing.T) { + if testing.Short() { + t.Skip("allocates a 20k-transmission store") + } + dbPath := filepath.Join(t.TempDir(), "acct.db") + acct113CreateDB(t, dbPath, 20000, 0) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + store := NewPacketStore(db, &PacketStoreConfig{}) + var m0, m1 runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&m0) + if err := store.Load(); err != nil { + t.Fatal(err) + } + for _, tx := range store.packets { // what analytics do: the decode cache is charged up front + tx.ParsedDecoded() + } + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m1) + heap := int64(m1.HeapAlloc) - int64(m0.HeapAlloc) + ratio := float64(store.trackedBytes) / float64(heap) + t.Logf("trackedBytes = %.1f MB, heap growth = %.1f MB, ratio = %.2f", float64(store.trackedBytes)/1048576, float64(heap)/1048576, ratio) + if ratio < 0.85 || ratio > 1.15 { + t.Errorf("trackedBytes is %.2f of the real heap growth, want 0.85-1.15", ratio) + } + runtime.KeepAlive(store) +} diff --git a/cmd/server/tracked_bytes_bench_113_test.go b/cmd/server/tracked_bytes_bench_113_test.go new file mode 100644 index 000000000..9e4a808d7 --- /dev/null +++ b/cmd/server/tracked_bytes_bench_113_test.go @@ -0,0 +1,159 @@ +package main + +import ( + "database/sql" + "fmt" + "path/filepath" + "testing" + "time" +) + +// Benchmarks for the accounting added in #113 / #164: the cold-load path +// (every tx and observation is charged and recharged), live ingest of new +// transmissions and of new observations on known ones, and an eviction pass. +// They use production entry points only, so they also run on master for the +// "before" numbers (see the PR for the interleaved comparison). The three +// DB-backed ingest benchmarks spend most of their time in SQLite, so the two +// pure-Go ones at the end isolate the accounting itself. + +const bench113Tx = 20000 + +func bench113DB(b *testing.B) string { + b.Helper() + dbPath := filepath.Join(b.TempDir(), "bench.db") + acct113CreateDB(b, dbPath, bench113Tx, bench113Tx/2) + return dbPath +} + +func BenchmarkTrackedBytes_Load_113(b *testing.B) { + dbPath := bench113DB(b) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + b.StopTimer() + db, err := OpenDB(dbPath) + if err != nil { + b.Fatal(err) + } + s := NewPacketStore(db, &PacketStoreConfig{}) + b.StartTimer() + if err := s.Load(); err != nil { + b.Fatal(err) + } + b.StopTimer() + db.conn.Close() + } +} + +// New transmissions, one observation each, 200 per iteration. +func BenchmarkTrackedBytes_IngestNewTx_113(b *testing.B) { + dbPath := bench113DB(b) + store := acct113Load(b, dbPath) + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + b.Fatal(err) + } + defer conn.Close() + maxTx := bench113Tx + maxObs := 0 + if err := conn.QueryRow(`SELECT COALESCE(MAX(id),0) FROM observations`).Scan(&maxObs); err != nil { + b.Fatal(err) + } + ts := time.Now().UTC().Add(time.Hour).Format(time.RFC3339) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + b.StopTimer() + first := maxTx + for k := 0; k < 200; k++ { + maxTx++ + maxObs++ + conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, 'aabb', ?, ?, 1, 5, 1, '{"type":"GRP_TXT","text":"live"}')`, maxTx, fmt.Sprintf("bn%08d", maxTx), ts) + conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs0', 'Alpha', 'RX', '["aa","bb","cc"]', ?)`, maxObs, maxTx, ts) + } + b.StartTimer() + store.IngestNewFromDB(first, 1000) + } +} + +// New observations with a longer path on known transmissions, 200 per iteration. +func BenchmarkTrackedBytes_IngestNewObs_113(b *testing.B) { + dbPath := bench113DB(b) + store := acct113Load(b, dbPath) + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + b.Fatal(err) + } + defer conn.Close() + maxObs := 0 + if err := conn.QueryRow(`SELECT COALESCE(MAX(id),0) FROM observations`).Scan(&maxObs); err != nil { + b.Fatal(err) + } + ts := time.Now().UTC().Format(time.RFC3339) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + b.StopTimer() + first := maxObs + for k := 0; k < 200; k++ { + maxObs++ + conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs3', 'Delta', 'RX', '["01","02","03","04","05","06"]', ?)`, maxObs, 1+(maxObs*7)%bench113Tx, ts) + } + b.StartTimer() + store.IngestNewObservations(first, 1000) + } +} + +// One retention pass that evicts half of a 20k store. +func BenchmarkTrackedBytes_Evict_113(b *testing.B) { + dbPath := bench113DB(b) + b.ReportAllocs() + for i := 0; i < b.N; i++ { + b.StopTimer() + s := acct113Load(b, dbPath) + s.retentionHours = 24 + b.StartTimer() + if n := s.RunEviction(); n != bench113Tx/2 { + b.Fatalf("evicted %d, want %d", n, bench113Tx/2) + } + } +} + +// Adding relays to a tx that already has records: the steady state (the same +// relays again, nothing appended) and the growth case (a fresh tx each time, +// three new keys). Runs on master too; there it does no accounting. +func BenchmarkAddResolvedPubkeys_Steady_113(b *testing.B) { + s := &PacketStore{byPathHop: map[string][]*StoreTx{}, pathHopResolved: map[*StoreTx][]uint64{}} + tx := &StoreTx{ID: 1, PathJSON: `["aa","bb","cc"]`} + keys := []string{acct113PK(1), acct113PK(2), acct113PK(3)} + hops := map[string]bool{} + s.addResolvedPubkeysToPathHopIndex(tx, keys, hops) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.addResolvedPubkeysToPathHopIndex(tx, keys, hops) + } +} + +func BenchmarkAddResolvedPubkeys_Grow_113(b *testing.B) { + s := &PacketStore{byPathHop: map[string][]*StoreTx{}, pathHopResolved: map[*StoreTx][]uint64{}} + keys := []string{acct113PK(1), acct113PK(2), acct113PK(3)} + hops := map[string]bool{} + txs := make([]*StoreTx, 1024) + for i := range txs { + txs[i] = &StoreTx{ID: i, PathJSON: `["aa","bb","cc"]`} + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + tx := txs[i%len(txs)] + if i%len(txs) == 0 { + clear(s.pathHopResolved) + clear(s.byPathHop) + } + s.addResolvedPubkeysToPathHopIndex(tx, keys, hops) + } +} diff --git a/cmd/server/tracked_bytes_measure_113_test.go b/cmd/server/tracked_bytes_measure_113_test.go new file mode 100644 index 000000000..0aaeb3d4d --- /dev/null +++ b/cmd/server/tracked_bytes_measure_113_test.go @@ -0,0 +1,214 @@ +package main + +import ( + "database/sql" + "encoding/json" + "fmt" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "testing" + "time" +) + +// Synthetic store for the trackedBytes accounting tests (#113, #164) and the +// tracked-vs-heap measurement below. It is built from standard Load() +// inputs only, so it also runs against master for the "before" numbers. + +// acct113PK is a 64-hex relay pubkey from a fixed pool. +func acct113PK(k int) string { + return fmt.Sprintf("%02x", k%256) + strings.Repeat("c4", 31) +} + +func acct113Hop(i, h, j int) string { return fmt.Sprintf("%02x", (i*7+h*13+j)%256) } + +// acct113CreateDB writes n transmissions with 1..4 observations each. The +// observations of one transmission carry paths of different lengths (the +// longest wins pickBestObservation, after the tx was created) and a +// resolved_path of full pubkeys. oldUntil of them are 3 days old, the rest +// one hour. +func acct113CreateDB(tb testing.TB, dbPath string, n, oldUntil int) { + acct113CreateDBOpts(tb, dbPath, n, oldUntil, 99) +} + +// acct113CreateDBOpts is acct113CreateDB with the resolved_path of every +// observation limited to its first maxResolved hops (0: left NULL), so the +// measurement can vary the resolved relays per transmission. +func acct113CreateDBOpts(tb testing.TB, dbPath string, n, oldUntil, maxResolved int) { + tb.Helper() + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + tb.Fatal(err) + } + defer conn.Close() + for _, q := range []string{ + `CREATE TABLE transmissions (id INTEGER PRIMARY KEY, raw_hex TEXT, hash TEXT, first_seen TEXT, + route_type INTEGER, payload_type INTEGER, payload_version INTEGER, decoded_json TEXT)`, + `CREATE TABLE observations (id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_id TEXT, + observer_name TEXT, direction TEXT, snr REAL, rssi REAL, score INTEGER, path_json TEXT, + timestamp TEXT, raw_hex TEXT, resolved_path TEXT)`, + `CREATE TABLE observers (rowid INTEGER PRIMARY KEY, id TEXT, name TEXT, iata TEXT)`, + `CREATE TABLE nodes (pubkey TEXT PRIMARY KEY, name TEXT, role TEXT, lat REAL, lon REAL, + last_seen TEXT, first_seen TEXT, frequency REAL)`, + `CREATE TABLE schema_version (version INTEGER)`, + `INSERT INTO schema_version (version) VALUES (1)`, + `CREATE INDEX idx_tx_first_seen ON transmissions(first_seen)`, + } { + if _, err := conn.Exec(q); err != nil { + tb.Fatalf("setup: %v\n%s", err, q) + } + } + names := []string{"Alpha", "Bravo", "Charlie", "Delta"} + for j, name := range names { + if _, err := conn.Exec(`INSERT INTO observers (rowid, id, name, iata) VALUES (?, ?, ?, 'TST')`, j+1, fmt.Sprintf("obs%d", j), name); err != nil { + tb.Fatal(err) + } + } + if _, err := conn.Exec("BEGIN"); err != nil { + tb.Fatal(err) + } + now := time.Now().UTC() + obsID := 0 + for i := 1; i <= n; i++ { + ts := now.Add(-time.Hour + time.Duration(i)*time.Second) + if i <= oldUntil { + ts = now.Add(-72*time.Hour + time.Duration(i)*time.Second) + } + decoded := fmt.Sprintf(`{"type":"GRP_TXT","channel":"#test","text":"message number %d with some body text","sender":"node%d"}`, i, i%50) + if _, err := conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, ?, 1, 5, 1, ?)`, i, fmt.Sprintf("1500%036d", i), fmt.Sprintf("h%06d", i), ts.Format(time.RFC3339), decoded); err != nil { + tb.Fatal(err) + } + for j := 0; j < 1+i%4; j++ { + obsID++ + hops := 1 + (i+j)%5 + path := make([]string, hops) + resolved := make([]string, hops) + for h := range path { + path[h] = acct113Hop(i, h, j) + resolved[h] = acct113PK(i + h*7) + } + pj, _ := json.Marshal(path) + rj, _ := json.Marshal(resolved) + var rp interface{} + if maxResolved > 0 { + if len(resolved) > maxResolved { + rj, _ = json.Marshal(resolved[:maxResolved]) + } + rp = string(rj) + } + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, resolved_path) + VALUES (?, ?, ?, ?, 'RX', -5.0, -90.0, 5, ?, ?, ?)`, + obsID, i, fmt.Sprintf("obs%d", j), names[j], string(pj), ts.Format(time.RFC3339), rp); err != nil { + tb.Fatal(err) + } + } + } + if _, err := conn.Exec("COMMIT"); err != nil { + tb.Fatal(err) + } +} + +func acct113Load(tb testing.TB, dbPath string) *PacketStore { + tb.Helper() + db, err := OpenDB(dbPath) + if err != nil { + tb.Fatal(err) + } + store := NewPacketStore(db, &PacketStoreConfig{}) + if err := store.Load(); err != nil { + tb.Fatal(err) + } + // Load starts the path-hop and subpath index builds in the background; + // they read every tx, so let them finish before a test touches one. + if !store.WaitIndexesReady(30 * time.Second) { + tb.Fatal("background index builds did not finish") + } + return store +} + +// heapOf returns the heap growth of building a store with build(), after a +// forced GC on both sides. +func heapOf(build func() *PacketStore) (*PacketStore, int64) { + var m0, m1 runtime.MemStats + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m0) + s := build() + runtime.GC() + runtime.GC() + runtime.ReadMemStats(&m1) + return s, int64(m1.HeapAlloc) - int64(m0.HeapAlloc) +} + +// TestTrackedBytesVsHeap_Measure_113 prints trackedBytes against the real heap +// for synthetic stores. It runs only with CORESCOPE_TRACKED_BYTES_MEASURE set +// to a comma-separated list of transmission counts, e.g. "100000,500000": +// +// CORESCOPE_TRACKED_BYTES_MEASURE=100000,500000 go test -run Measure_113 -v -timeout 30m +// +// For each size it loads the store with no resolved relays and with 1, 2 and +// all resolved hops per observation, so the marginal heap per resolved relay +// key and per pathHopResolved record can be read off. It compiles against +// master too, for the "before" numbers. +func TestTrackedBytesVsHeap_Measure_113(t *testing.T) { + sizes := os.Getenv("CORESCOPE_TRACKED_BYTES_MEASURE") + if sizes == "" { + t.Skip("set CORESCOPE_TRACKED_BYTES_MEASURE=100000,500000 to run") + } + type result struct { + tracked, heap int64 + keys, records int + obs int + } + for _, field := range strings.Split(sizes, ",") { + n, err := strconv.Atoi(strings.TrimSpace(field)) + if err != nil { + t.Fatal(err) + } + run := func(maxResolved int, warm bool) result { + dbPath := filepath.Join(t.TempDir(), "m.db") + acct113CreateDBOpts(t, dbPath, n, 0, maxResolved) + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + store, heap := heapOf(func() *PacketStore { + s := NewPacketStore(db, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + if warm { // what analytics do: parse every transmission's JSON + for _, tx := range s.packets { + tx.ParsedDecoded() + } + } + return s + }) + r := result{tracked: store.trackedBytes, heap: heap, records: len(store.pathHopResolved), obs: store.totalObs} + for _, h := range store.pathHopResolved { + r.keys += len(h) + } + runtime.KeepAlive(store) + db.conn.Close() + return r + } + mb := func(b int64) float64 { return float64(b) / 1048576 } + base := run(0, false) + t.Logf("N=%d tx, %d obs", n, base.obs) + t.Logf(" no resolved relays, decode cache cold: tracked %.1f MB heap %.1f MB tracked/heap %.2f", mb(base.tracked), mb(base.heap), float64(base.tracked)/float64(base.heap)) + warm := run(0, true) + t.Logf(" no resolved relays, decode cache warm: tracked %.1f MB heap %.1f MB tracked/heap %.2f", mb(warm.tracked), mb(warm.heap), float64(warm.tracked)/float64(warm.heap)) + base = warm // the resolved-relay rows below are measured warm + for _, m := range []int{1, 2, 99} { + r := run(m, true) + t.Logf(" resolved hops <= %d per obs: %d keys in %d records (%.1f keys/record): tracked %.1f MB heap %.1f MB tracked/heap %.2f", + m, r.keys, r.records, float64(r.keys)/float64(r.records), mb(r.tracked), mb(r.heap), float64(r.tracked)/float64(r.heap)) + t.Logf(" marginal per record: heap %.1f B, tracked %.1f B (per key: heap %.1f B, tracked %.1f B)", + float64(r.heap-base.heap)/float64(r.records), float64(r.tracked-base.tracked)/float64(r.records), + float64(r.heap-base.heap)/float64(r.keys), float64(r.tracked-base.tracked)/float64(r.keys)) + } + } +} diff --git a/cmd/server/tracked_bytes_recharge_bench_113_test.go b/cmd/server/tracked_bytes_recharge_bench_113_test.go new file mode 100644 index 000000000..629ae7563 --- /dev/null +++ b/cmd/server/tracked_bytes_recharge_bench_113_test.go @@ -0,0 +1,37 @@ +package main + +import "testing" + +// rechargeTx next to the estimate it wraps, on a tx with a 6-hop path and +// decoded JSON (the hot call: once per pickBestObservation). +func BenchmarkRechargeTx_113(b *testing.B) { + tx := &StoreTx{ID: 1, RawHex: "aabbccddeeff", Hash: "0123456789abcdef", DecodedJSON: `{"type":"GRP_TXT","channel":"#test","text":"hello world","sender":"node1"}`, + PathJSON: `["01","02","03","04","05","06"]`} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if i%2 == 0 { + tx.PathJSON = `["01","02","03","04","05","06"]` + } else { + tx.PathJSON = `["01","02","03","04","05","07"]` + } + tx.pathParsed = false + rechargeTx(tx) + } +} + +func BenchmarkEstimateStoreTxBytesOnly_113(b *testing.B) { + tx := &StoreTx{ID: 1, RawHex: "aabbccddeeff", Hash: "0123456789abcdef", DecodedJSON: `{"type":"GRP_TXT","channel":"#test","text":"hello world","sender":"node1"}`, + PathJSON: `["01","02","03","04","05","06"]`} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if i%2 == 0 { + tx.PathJSON = `["01","02","03","04","05","06"]` + } else { + tx.PathJSON = `["01","02","03","04","05","07"]` + } + tx.pathParsed = false + estimateStoreTxBytes(tx) + } +} diff --git a/docs/deployment.md b/docs/deployment.md index 2f93e355d..471023cfa 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -608,6 +608,8 @@ The in-memory packet store grows with retained packets. Configure retention limi } ``` +`packetStore.maxMemoryMB` bounds the store **and the caches that belong to it** — the decoded-packet cache, the path indexes, the resolved relay entries and the per-packet index entries, not just the stored rows. It is enforced in two places: the startup load stops at the budget, and the store evicts oldest-first when it exceeds it, down to 85% of it. Leaving it unset means no limit. Actual usage is on `/api/perf` as `packetStore.trackedMB`, next to `maxMB`. + ### Database locked errors SQLite doesn't support concurrent writers well. Ensure only one CoreScope instance accesses the database file. If running multiple containers, each needs its own database. diff --git a/docs/go-migration.md b/docs/go-migration.md index ec87dff78..c262d5f8b 100644 --- a/docs/go-migration.md +++ b/docs/go-migration.md @@ -78,9 +78,17 @@ Node.js uses `mqtt://` and `mqtts://` scheme prefixes. The Go MQTT library (paho Both engines support `retention.nodeDays` (default: 7). Stale nodes are moved to the `inactive_nodes` table on the same schedule. No config change needed. -### `packetStore.maxMemoryMB` (Go ignores this — it's Node-only) +### `packetStore.maxMemoryMB` (both engines read it) -The Node.js server has a configurable in-memory packet store limit (`packetStore.maxMemoryMB`). The Go server has its own in-memory store that loads all packets from SQLite on startup — it does not read this config value. This is safe to leave in your config; Go simply ignores it. +Both engines take a memory budget for the in-memory packet store from `packetStore.maxMemoryMB`. In the Go server it does three things: + +- **Startup:** the cold load stops once the budget is reached, so the server starts with the newest packets that fit rather than everything in SQLite. +- **While running:** the store evicts oldest-first when it passes the budget, down to 85% of it. +- **Go heap:** when neither `GOMEMLIMIT` nor `runtime.maxMemoryMB` is set, `GOMEMLIMIT` is derived from this value with 1.5x headroom. + +Unset (the default) means no limit: every packet within retention is loaded and none is evicted for memory. An old Node-era value left in a config file is therefore **not** inert — it caps the Go store too. Check it before upgrading, and read `/api/perf` (`packetStore.trackedMB` against `maxMB`) afterwards. + +**What the budget counts.** The store tracks what it holds per transmission and per observation, and evicts exactly what it charged. The estimate covers the stored rows, the index entries, the path indexes (`byPathHop` and the subpath index, both of which depend on the path chosen after the transmission was created), the decoded-packet cache (charged up front for every transmission, whether or not analytics have parsed it yet), the per-observation dedup key and strings, and the resolved relay hops (#164). Since #113 and #164 the estimate matches the real heap of the store within a few percent on synthetic data; before, it covered about two thirds of it, and less once relays were resolved. A budget that used to hold the whole store may therefore now evict, and the cold load admits correspondingly fewer packets at the same `maxMemoryMB`. ### `channelKeys` / `channel-rainbow.json` (compatible) @@ -338,7 +346,7 @@ docker start corescope-prod |---------|--------|------------| | Companion bridge advertisements | `meshcore/advertisement` topic not handled by Go ingestor | Users relying on companion bridge adverts must stay on Node.js or wait for Go support | | Companion bridge `self_info` | `meshcore/self_info` topic not handled | Same as above — minimal impact (only affects local node identity) | -| `packetStore.maxMemoryMB` config | Go doesn't read this setting | Go manages its own memory; no action needed | +| `packetStore.maxMemoryMB` config | Read by both engines, but Go's store accounts memory differently | Review the value before upgrading — see [`packetStore.maxMemoryMB`](#packetstoremaxmemorymb-both-engines-read-it) | | Docker Hub images | Go images not published yet | Build locally with `docker build -f Dockerfile.go` | | `manage.sh --engine` flag | Can't toggle engines via manage.sh | Manual image swap required (see [Switch to Go](#switch-to-go)) | @@ -349,7 +357,7 @@ docker start corescope-prod | `engine` field in `/api/health` | Not present or `"node"` | Always `"go"` | | MQTT URL scheme | Uses `mqtt://` / `mqtts://` natively | Auto-converts to `tcp://` / `ssl://` (transparent) | | Process model | Single Node.js process (server + ingestor) | Two binaries: `corescope-ingestor` + `corescope-server` (managed by supervisord) | -| Memory management | Configurable via `packetStore.maxMemoryMB` | Loads all packets; no configurable limit | +| Memory management | Configurable via `packetStore.maxMemoryMB` | Same setting: caps the cold load, evicts oldest-first above it, and sets `GOMEMLIMIT` when that is otherwise unset | | Startup time | Faster (no compilation) | Slightly slower (loads all packets from DB into memory) | ---