From 715d5d6637fbd0a1337543eee1e97dcf782e5404 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:21:28 +0200 Subject: [PATCH 1/3] test(store): pin the two trackedBytes leaks left after #198 (#202) Hash migration: a test merges two duplicate transmissions, evicts both and asserts trackedBytes returns exactly to the live ballast, which cannot hide a double credit behind the clamp at zero. It passes on master: the in-memory merge branch in migrateContentHashesAsync is dead code (it tests u.newHash, set on a loop copy), so the duplicates keep their own observations and each is credited once. The test guards the accounting if that branch is ever made live. path_json fallback: tests for the five ways a fallback relay reaches the store (Load, LoadChunked, background fill, IngestNewFromDB, IngestNewObservations). They fail on master: byNode/nodeHashes entries the fallback adds are not charged to trackedBytes, and eviction never removes them, so evicted transmissions stay pinned. Relates to #202. Co-Authored-By: Claude Sonnet 5.5 --- cmd/server/tracked_bytes_fallback_202_test.go | 216 ++++++++++++++++++ cmd/server/tracked_bytes_leaks_202_test.go | 132 +++++++++++ 2 files changed, 348 insertions(+) create mode 100644 cmd/server/tracked_bytes_fallback_202_test.go create mode 100644 cmd/server/tracked_bytes_leaks_202_test.go diff --git a/cmd/server/tracked_bytes_fallback_202_test.go b/cmd/server/tracked_bytes_fallback_202_test.go new file mode 100644 index 000000000..3d6a10223 --- /dev/null +++ b/cmd/server/tracked_bytes_fallback_202_test.go @@ -0,0 +1,216 @@ +package main + +import ( + "database/sql" + "fmt" + "path/filepath" + "testing" + "time" +) + +// Issue #202, leak 2: the relays indexObservationRelayHops resolves from +// path_json when an observation has no persisted resolved_path. + +// leak202SeedNodes gives the store a node cache with one repeater per +// acct113PK key (acct113PK(k) starts with the byte k), so every one-byte hop +// resolves to a unique node. withNodes=false installs an empty cache, so the +// same rows resolve nothing: a twin store that differs from the first only by +// the fallback's relays. +func leak202SeedNodes(s *PacketStore, withNodes bool) { + var nodes []nodeInfo + if withNodes { + for k := 0; k < 256; k++ { + nodes = append(nodes, nodeInfo{PublicKey: acct113PK(k), Name: fmt.Sprintf("n%d", k), Role: "repeater"}) + } + } + s.cacheMu.Lock() + s.nodeCache = nodes + s.nodePM = buildPrefixMap(nodes) + s.nodeCacheTime = time.Now() + s.cacheMu.Unlock() +} + +// leak202Fixture is a DB whose observations all have a NULL resolved_path, so +// every relay comes from the path_json fallback. +func leak202Fixture(t *testing.T, n, oldUntil int) string { + t.Helper() + dbPath := filepath.Join(t.TempDir(), "leak202.db") + acct113CreateDBOpts(t, dbPath, n, oldUntil, 0) + return dbPath +} + +func leak202Open(t *testing.T, dbPath string, withNodes bool, cfg *PacketStoreConfig) *PacketStore { + t.Helper() + db, err := OpenDB(dbPath) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.conn.Close() }) + store := NewPacketStore(db, cfg) + leak202SeedNodes(store, withNodes) + return store +} + +func leak202Ready(t *testing.T, s *PacketStore) { + t.Helper() + if !s.WaitIndexesReady(30 * time.Second) { + t.Fatal("background index builds did not finish") + } +} + +// leak202ByNodeEntries counts the (node, tx) pairs byNode holds. +func leak202ByNodeEntries(s *PacketStore) int { + s.mu.RLock() + defer s.mu.RUnlock() + n := 0 + for _, l := range s.byNode { + n += len(l) + } + return n +} + +// leak202Insert adds rows after the store was loaded, as the ingestor would. +func leak202Insert(t *testing.T, dbPath string, fn func(conn *sql.DB, maxObs int)) { + t.Helper() + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + defer conn.Close() + var maxObs int + if err := conn.QueryRow(`SELECT MAX(id) FROM observations`).Scan(&maxObs); err != nil { + t.Fatal(err) + } + fn(conn, maxObs) +} + +// Every way a fallback relay reaches the store. Each builds the same rows +// twice, once with nodes the hops resolve to and once without. +var leak202Scenarios = []struct { + name string + build func(t *testing.T, withNodes bool) *PacketStore +}{ + {"Load", func(t *testing.T, withNodes bool) *PacketStore { + s := leak202Open(t, leak202Fixture(t, 200, 0), withNodes, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + return s + }}, + {"LoadChunked", func(t *testing.T, withNodes bool) *PacketStore { + s := leak202Open(t, leak202Fixture(t, 200, 0), withNodes, &PacketStoreConfig{RetentionHours: 168}) + if err := s.RunStartupLoad(100); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + return s + }}, + {"BackgroundFill", func(t *testing.T, withNodes bool) *PacketStore { + s := leak202Open(t, leak202Fixture(t, 200, 100), withNodes, &PacketStoreConfig{RetentionHours: 168, HotStartupHours: 2}) + if err := s.RunStartupLoad(100); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + if len(s.packets) != 200 { + t.Fatalf("setup: %d packets after the startup load, want 200 (hot window plus background fill)", len(s.packets)) + } + return s + }}, + {"IngestNewFromDB", func(t *testing.T, withNodes bool) *PacketStore { + dbPath := leak202Fixture(t, 20, 0) + s := leak202Open(t, dbPath, withNodes, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + leak202Insert(t, dbPath, func(conn *sql.DB, maxObs int) { + now := time.Now().UTC().Format(time.RFC3339) + for k := 1; k <= 5; k++ { + 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"}')`, 20+k, fmt.Sprintf("live%06d", 20+k), now); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs0', 'Alpha', 'RX', ?, ?)`, maxObs+k, 20+k, fmt.Sprintf(`["%02x","%02x"]`, 0xa0+k, 0xb0+k), now); err != nil { + t.Fatal(err) + } + } + }) + s.IngestNewFromDB(20, 100) + if len(s.packets) != 25 { + t.Fatalf("setup: %d packets after the live ingest, want 25", len(s.packets)) + } + return s + }}, + {"IngestNewObservations", func(t *testing.T, withNodes bool) *PacketStore { + dbPath := leak202Fixture(t, 20, 0) + s := leak202Open(t, dbPath, withNodes, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + var sinceObs int + leak202Insert(t, dbPath, func(conn *sql.DB, maxObs int) { + sinceObs = maxObs + now := time.Now().UTC().Format(time.RFC3339) + for k := 1; k <= 5; k++ { // late observations of existing transmissions, through new relays + if _, err := conn.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, path_json, timestamp) + VALUES (?, ?, 'obs3', 'Delta', 'RX', ?, ?)`, maxObs+k, k, fmt.Sprintf(`["%02x","%02x","%02x"]`, 0xc0+k, 0xd0+k, 0xe0+k), now); err != nil { + t.Fatal(err) + } + } + }) + s.IngestNewObservations(sinceObs, 100) + return s + }}, +} + +// The fallback puts a transmission into byNode under each relay it resolves, +// so it is memory the store holds and must be charged. The twin without +// nodes holds none of it. +func TestFallbackRelays_AreCharged_202(t *testing.T) { + for _, sc := range leak202Scenarios { + t.Run(sc.name, func(t *testing.T) { + with, without := sc.build(t, true), sc.build(t, false) + if got, base := leak202ByNodeEntries(with), leak202ByNodeEntries(without); got <= base { + t.Fatalf("setup: the fallback indexed nothing (byNode entries %d with nodes, %d without)", got, base) + } + if with.trackedBytes <= without.trackedBytes { + t.Fatalf("trackedBytes = %d with the fallback's byNode entries, %d without them: they are not charged", + with.trackedBytes, without.trackedBytes) + } + }) + } +} + +// Eviction removes what the fallback indexed (byNode, nodeHashes) and credits +// what it charged: an emptied store holds nothing and is charged nothing. +func TestFallbackRelays_EvictionRemovesAndCredits_202(t *testing.T) { + for _, sc := range leak202Scenarios { + t.Run(sc.name, func(t *testing.T) { + s := sc.build(t, true) + s.mu.Lock() + for _, tx := range s.packets { // live tx carry current timestamps + tx.FirstSeen = time.Now().UTC().Add(-100 * time.Hour).Format(time.RFC3339) + } + s.retentionHours = 24 + s.mu.Unlock() + if n := s.RunEviction(); n == 0 { + t.Fatal("setup: eviction removed nothing") + } + s.mu.RLock() + defer s.mu.RUnlock() + if len(s.packets) != 0 { + t.Fatalf("setup: %d transmissions left after eviction", len(s.packets)) + } + if s.trackedBytes != 0 { + t.Errorf("trackedBytes = %d after every transmission was evicted, want 0", s.trackedBytes) + } + if len(s.byNode) != 0 || len(s.nodeHashes) != 0 { + t.Errorf("an emptied store still indexes %d nodes in byNode and %d in nodeHashes: evicted transmissions are pinned", + len(s.byNode), len(s.nodeHashes)) + } + }) + } +} diff --git a/cmd/server/tracked_bytes_leaks_202_test.go b/cmd/server/tracked_bytes_leaks_202_test.go new file mode 100644 index 000000000..498f06726 --- /dev/null +++ b/cmd/server/tracked_bytes_leaks_202_test.go @@ -0,0 +1,132 @@ +package main + +import ( + "testing" + "time" +) + +// Issue #202: two accounting leaks left after #198. +// +// Every test asserts against a live "ballast" transmission that is never +// evicted, so a double credit shows up as a number below the ballast's +// charge instead of being hidden by the clamp at zero in the eviction pass. + +// leak202Ballast inserts one recent, unrelated transmission with one +// observation. It only has to stay live. +func leak202Ballast(t *testing.T, db *DB) { + t.Helper() + now := time.Now().UTC().Format(time.RFC3339) + if _, err := db.conn.Exec(`INSERT INTO transmissions + (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (900, '1500aabbccdd', 'ballast0000000001', ?, 1, 5, '{"type":"GRP_TXT","text":"ballast"}')`, now); err != nil { + t.Fatal(err) + } + if _, err := db.conn.Exec(`INSERT INTO observations + (id, transmission_id, observer_id, observer_name, path_json, timestamp) + VALUES (900, 900, 'observer', 'Observer', '["aa"]', ?)`, now); err != nil { + t.Fatal(err) + } +} + +// leak202LiveCharge is the charge of everything still in the store. +func leak202LiveCharge(s *PacketStore) int64 { + s.mu.RLock() + defer s.mu.RUnlock() + sum, _ := acct113Sum(s) + return sum +} + +// leak202TxCharge is everything trackedBytes holds for one transmission. +func leak202TxCharge(t *testing.T, s *PacketStore, id int) int64 { + t.Helper() + s.mu.RLock() + defer s.mu.RUnlock() + tx := s.byTxID[id] + if tx == nil { + t.Fatalf("tx %d not in the store", id) + } + sum := s.txChargedBytes(tx) + for _, o := range tx.Observations { + sum += estimateStoreObsBytes(o) + } + return sum +} + +// A hash migration that merges two duplicate transmissions must credit each +// observation exactly once when the store later evicts them. +func TestHashMigrationMerge_EvictionCreditsEveryObservationOnce_202(t *testing.T) { + db := setupTestDBv2(t) + store := NewPacketStore(db, nil) + + rawHex := "0A00D69FD7A5A7475DB07337749AE61FA53A4788E976" + rows := []struct { + id int + oldHash string + firstSeen string + paths []string // one observation per entry + }{ + {1, "old-hash-one", "2026-01-01T00:00:00Z", []string{`["AA"]`, `["AA","BB"]`}}, + {2, "old-hash-two", "2026-01-01T00:00:01Z", []string{`["CC"]`, `["CC","DD","EE"]`, `["FF"]`}}, + } + obsID := 0 + for _, r := range rows { + if _, err := db.conn.Exec(`INSERT INTO transmissions + (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, ?, ?, ?, 1, 5, '{}')`, r.id, rawHex, r.oldHash, r.firstSeen); err != nil { + t.Fatal(err) + } + for _, p := range r.paths { + obsID++ + if _, err := db.conn.Exec(`INSERT INTO observations + (id, transmission_id, observer_id, observer_name, path_json, timestamp) + VALUES (?, ?, 'observer', 'Observer', ?, ?)`, obsID, r.id, p, r.firstSeen); err != nil { + t.Fatal(err) + } + } + } + leak202Ballast(t, db) + + if err := store.Load(); err != nil { + t.Fatal(err) + } + if !store.WaitIndexesReady(30 * time.Second) { + t.Fatal("background index builds did not finish") + } + ballast := leak202TxCharge(t, store, 900) + + migrateContentHashesAsync(store, 1, 0) + if !store.hashMigrationComplete.Load() { + t.Fatal("hash migration did not complete") + } + var dbTx int + if err := db.conn.QueryRow(`SELECT COUNT(*) FROM transmissions WHERE raw_hex = ?`, rawHex).Scan(&dbTx); err != nil { + t.Fatal(err) + } + if dbTx != 1 { + t.Fatalf("setup: %d duplicate rows left in the DB, want the merge to leave 1", dbTx) + } + + // Whatever the merge did in memory, trackedBytes must still equal what the + // store holds, before anything is evicted. + if got, want := store.trackedBytes, leak202LiveCharge(store); got != want { + t.Fatalf("after the merge trackedBytes = %d, live charges sum to %d (drift %d)", got, want, got-want) + } + + store.mu.Lock() + store.retentionHours = 24 + store.mu.Unlock() + if n := store.RunEviction(); n == 0 { + t.Fatal("setup: eviction removed nothing") + } + + store.mu.RLock() + got, live := store.trackedBytes, len(store.packets) + store.mu.RUnlock() + if live != 1 { + t.Fatalf("setup: %d transmissions live after eviction, want only the ballast", live) + } + if got != ballast { + t.Fatalf("after evicting both duplicates trackedBytes = %d, want exactly the ballast's %d (drift %d)", + got, ballast, got-ballast) + } +} From 7540a02d118de300e4a9a2b5e968bdb95bcc0f2c Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:26:11 +0200 Subject: [PATCH 2/3] fix(store): charge the path_json fallback and remove it at eviction (#202) indexObservationRelayHops fed fallback relays (observations without a persisted resolved_path) into byNode and nodeHashes without charging trackedBytes, and eviction never removed those entries, so evicted transmissions stayed pinned in byNode. addFallbackRelay now charges each new (relay, tx) pair, records it in a per-tx fallbackByNode map and txChargedBytes adds the record's allowance, so eviction removes the entries and credits exactly what was charged. No StoreTx field is added (still 320 bytes) and nothing new runs per packet beyond one map lookup per added relay. hash_migrate.go: the in-memory merge branch, which would have copied the duplicate's observations to the survivor while the duplicate stayed in s.packets (a double credit at eviction), never ran: it tested u.newHash, which was cleared on a loop copy. Remove the dead branch and the dead marker; behaviour is unchanged and the new test keeps the accounting pinned. acct113Sum, the shared accounting invariant, gains the fallback term. Relates to #202. Co-Authored-By: Claude Sonnet 5.5 --- cmd/server/hash_migrate.go | 25 ++- cmd/server/store.go | 84 +++++++++- .../tracked_bytes_accounting_113_test.go | 3 +- .../tracked_bytes_fallback_exact_202_test.go | 158 ++++++++++++++++++ 4 files changed, 248 insertions(+), 22 deletions(-) create mode 100644 cmd/server/tracked_bytes_fallback_exact_202_test.go diff --git a/cmd/server/hash_migrate.go b/cmd/server/hash_migrate.go index c76ca9357..f51cbcb41 100644 --- a/cmd/server/hash_migrate.go +++ b/cmd/server/hash_migrate.go @@ -93,7 +93,6 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration dbTx.Exec("UPDATE observations SET transmission_id = ? WHERE transmission_id = ?", survID, u.tx.ID) dbTx.Exec("DELETE FROM transmissions WHERE id = ?", u.tx.ID) collisionSurvivors[u.newHash] = append(collisionSurvivors[u.newHash], survID) - u.newHash = "" // mark for in-memory removal only } } } @@ -106,22 +105,18 @@ func migrateContentHashesAsync(store *PacketStore, batchSize int, yieldDuration // Update in-memory index under write lock. store.mu.Lock() + // A merged duplicate stays in memory with its own observations (see + // collisionSurvivors above): the DB rows were re-parented, the + // in-memory ones are not, so trackedBytes keeps charging each + // observation to the one tx that holds it and eviction credits it + // once (#202). Copying them to the survivor here would credit them + // twice, as each of the two txs is evicted. A fix that removes the + // duplicate must move them (see + // TestHashMigrationMerge_EvictionCreditsEveryObservationOnce_202). for _, u := range updates { delete(store.byHash, u.oldHash) - if u.newHash == "" { - // Merged duplicate — remove from packets slice and indexes. - delete(store.byTxID, u.tx.ID) - // Move observations to survivor if present. - if surv := store.byHash[ComputeContentHash(u.tx.RawHex)]; surv != nil { - for _, obs := range u.tx.Observations { - surv.Observations = append(surv.Observations, obs) - surv.ObservationCount++ - } - } - } else { - u.tx.Hash = u.newHash - store.byHash[u.newHash] = u.tx - } + u.tx.Hash = u.newHash + store.byHash[u.newHash] = u.tx } for newHash, survivorIDs := range collisionSurvivors { evidence := collisionEvidenceByHash[newHash] diff --git a/cmd/server/store.go b/cmd/server/store.go index 566294d95..a2ba63611 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -430,6 +430,15 @@ type PacketStore struct { // retainResolvedPathHops drops anything no longer in s.packets. pathHopResolved map[*StoreTx][]uint64 + // fallbackByNode records, per transmission, the relay pubkeys the + // path_json fallback (indexObservationRelayHops) put it under in byNode + // and nodeHashes, and nothing else. Without a persisted resolved_path + // there is no other record of them, so eviction reads them here to remove + // the entries and to credit what addFallbackRelay charged (#202). At most + // one entry per (relay, live transmission): evictStaleInternal deletes + // evicted transmissions' records. + fallbackByNode map[*StoreTx][]string + // Precomputed distinct advert pubkey count (refcounted for eviction correctness). // Updated incrementally during Load/Ingest/Evict — avoids JSON parsing in GetPerfStoreStats. advertPubkeys map[string]int // pubkey → number of advert packets referencing it @@ -773,6 +782,7 @@ func NewPacketStore(db *DB, cfg *PacketStoreConfig, cacheTTLs ...map[string]inte spTxIndex: make(map[string][]*StoreTx, 4096), advertPubkeys: make(map[string]int), pathHopResolved: make(map[*StoreTx][]uint64), + fallbackByNode: make(map[*StoreTx][]string), clockSkew: NewClockSkewEngine(), useResolvedPathIndex: true, areaNodeCache: make(map[string]map[string]bool), @@ -1541,7 +1551,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // nodes keep their analytics history across a restart without // polluting the resolved_path/path-hop indexes (#1352). for _, pk := range localByNodePKsByTx[tx.ID] { - s.addToByNode(tx, pk) + s.addFallbackRelay(tx, pk) } } s.mu.Unlock() @@ -1869,10 +1879,22 @@ func resolvedRelayBytes(n int) int64 { return perResolvedRecordBytes + int64(n)*perResolvedRelayBytes } +// fallbackRelayBytes is the memory a tx holds for n relays the path_json +// fallback indexed it under (#202): per relay a byNode slot, a nodeHashes +// entry and the pubkey in the record, plus the fallbackByNode map entry once. +// Like resolvedRelayBytes it is a function of n alone, so eviction re-derives +// the charge from len(s.fallbackByNode[tx]). +func fallbackRelayBytes(n int) int64 { + if n <= 0 { + return 0 + } + return perFallbackRecordBytes + int64(n)*perFallbackRelayBytes +} + // 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])) + return int64(tx.chargedBytes) + resolvedRelayBytes(len(s.pathHopResolved[tx])) + fallbackRelayBytes(len(s.fallbackByNode[tx])) } func pathLen(pathJSON string) int { @@ -1999,7 +2021,7 @@ func (s *PacketStore) indexObservationRelayHops(tx *StoreTx, persisted persisted return } for _, pk := range extractResolvedPubkeys(resolvePathForObsColdLoad(pathJSON, observerID, tx, relayPM, skipped)) { - s.addToByNode(tx, pk) + s.addFallbackRelay(tx, pk) } } @@ -2032,16 +2054,39 @@ func (s *PacketStore) indexByNode(tx *StoreTx) bool { // addToByNode adds tx to byNode[pubkey] with dedup via nodeHashes. // Returns true if this is a genuinely new node (pubkey not seen before). func (s *PacketStore) addToByNode(tx *StoreTx, pubkey string) bool { - isNew := s.nodeHashes[pubkey] == nil + isNew, _ := s.indexByNodeKey(tx, pubkey) + return isNew +} + +// indexByNodeKey is addToByNode that also reports whether it added an entry +// (false: tx was already indexed under pubkey). +func (s *PacketStore) indexByNodeKey(tx *StoreTx, pubkey string) (isNew, added bool) { + isNew = s.nodeHashes[pubkey] == nil if isNew { s.nodeHashes[pubkey] = make(map[string]bool) } if s.nodeHashes[pubkey][tx.Hash] { - return false + return false, false } s.nodeHashes[pubkey][tx.Hash] = true s.byNode[pubkey] = append(s.byNode[pubkey], tx) - return isNew + return isNew, true +} + +// addFallbackRelay indexes tx under a relay the path_json fallback resolved, +// in byNode only (#1352), and charges and records what that added (#202). A +// relay tx is already indexed under (a decoded pubkey, or a resolved relay, +// which has its own allowance) adds nothing and costs nothing. Must hold s.mu. +func (s *PacketStore) addFallbackRelay(tx *StoreTx, pubkey string) { + if _, added := s.indexByNodeKey(tx, pubkey); !added { + return + } + if s.fallbackByNode == nil { + s.fallbackByNode = make(map[*StoreTx][]string) + } + before := len(s.fallbackByNode[tx]) + s.fallbackByNode[tx] = append(s.fallbackByNode[tx], pubkey) + s.trackedBytes += fallbackRelayBytes(before+1) - fallbackRelayBytes(before) } // trackAdvertPubkey increments the advertPubkeys refcount for ADVERT packets. @@ -5197,6 +5242,16 @@ const ( 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 + // path_json fallback relays (#202), charged per distinct relay a tx is + // indexed under by addFallbackRelay. Not measured on its own: derived from + // the resolved allowance above by dropping what the fallback does not + // create (the byPathHop slot, the pathHopResolved hash and the + // resolved-pubkey index entry) and keeping the byNode slot, the nodeHashes + // entry and, in place of the hash, the 16-byte pubkey string header in the + // record; the record is that of pathHopResolved. + perFallbackRelayBytes = 80 // byNode slot + nodeHashes entry + pubkey header in fallbackByNode (append growth) + perFallbackRecordBytes = 140 // fallbackByNode 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 @@ -5486,6 +5541,23 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { affectedNodes[pk] = struct{}{} evictedFromNode[pk] = true } + // ...and the relays the path_json fallback indexed it under, which no + // resolved_path names (#202). The record is deleted here; its charge + // was taken in txChargedBytes above. + for _, pk := range s.fallbackByNode[tx] { + if evictedFromNode[pk] { + continue + } + if hashes, ok := s.nodeHashes[pk]; ok { + delete(hashes, tx.Hash) + if len(hashes) == 0 { + delete(s.nodeHashes, pk) + } + } + affectedNodes[pk] = struct{}{} + evictedFromNode[pk] = true + } + delete(s.fallbackByNode, tx) // Remove from resolved pubkey index s.removeFromResolvedPubkeyIndex(tx.ID) diff --git a/cmd/server/tracked_bytes_accounting_113_test.go b/cmd/server/tracked_bytes_accounting_113_test.go index fb416a8e3..f70d591c2 100644 --- a/cmd/server/tracked_bytes_accounting_113_test.go +++ b/cmd/server/tracked_bytes_accounting_113_test.go @@ -19,6 +19,7 @@ import ( // // trackedBytes == Σ over live tx of ( tx.chargedBytes // + resolvedRelayBytes(len(pathHopResolved[tx])) +// + fallbackRelayBytes(len(fallbackByNode[tx])) // + Σ estimateStoreObsBytes(obs) ) // // and tx.chargedBytes == estimateStoreTxBytes(tx) for every live tx, i.e. no @@ -31,7 +32,7 @@ func acct113Sum(s *PacketStore) (sum int64, stale int) { if int64(tx.chargedBytes) != estimateStoreTxBytes(tx) { stale++ } - sum += int64(tx.chargedBytes) + resolvedRelayBytes(len(s.pathHopResolved[tx])) + sum += int64(tx.chargedBytes) + resolvedRelayBytes(len(s.pathHopResolved[tx])) + fallbackRelayBytes(len(s.fallbackByNode[tx])) for _, o := range tx.Observations { sum += estimateStoreObsBytes(o) } diff --git a/cmd/server/tracked_bytes_fallback_exact_202_test.go b/cmd/server/tracked_bytes_fallback_exact_202_test.go new file mode 100644 index 000000000..302be43c6 --- /dev/null +++ b/cmd/server/tracked_bytes_fallback_exact_202_test.go @@ -0,0 +1,158 @@ +package main + +import ( + "fmt" + "testing" + "time" +) + +// Issue #202, leak 2, continued: the amount charged and credited is exact, not +// just non-zero (tracked_bytes_fallback_202_test.go). + +// The fallback's charge is exactly the record eviction later reads, per +// transmission, and every recorded relay is indexed under the transmission. +func TestFallbackRelays_ChargeIsExactlyTheRecord_202(t *testing.T) { + for _, sc := range leak202Scenarios { + t.Run(sc.name, func(t *testing.T) { + with, without := sc.build(t, true), sc.build(t, false) + with.mu.RLock() + defer with.mu.RUnlock() + + var want int64 + records := 0 + for tx, pks := range with.fallbackByNode { + records++ + want += fallbackRelayBytes(len(pks)) + seen := map[string]bool{} + for _, pk := range pks { + if seen[pk] { + t.Fatalf("tx %d is recorded twice under %s", tx.ID, pk) + } + seen[pk] = true + if !with.nodeHashes[pk][tx.Hash] || countTxInLocked(with.byNode[pk], tx) != 1 { + t.Fatalf("tx %d is recorded under %s but byNode/nodeHashes do not hold it once", tx.ID, pk) + } + } + } + if records == 0 { + t.Fatal("setup: no fallback records") + } + if got := with.trackedBytes - without.trackedBytes; got != want { + t.Errorf("the fallback added %d bytes to trackedBytes, its records account for %d", got, want) + } + sum, stale := acct113Sum(with) + if stale > 0 || with.trackedBytes != sum { + t.Errorf("trackedBytes = %d, live charges sum to %d (%d stale)", with.trackedBytes, sum, stale) + } + }) + } +} + +// Evicting one half keeps the other half exactly charged, with the fallback's +// records of the evicted half gone; evicting the rest empties the store. +func TestFallbackRelays_PartialEvictionStaysExact_202(t *testing.T) { + s := leak202Scenarios[2].build(t, true) // BackgroundFill: 100 old, 100 recent + s.mu.Lock() + s.retentionHours = 24 + s.mu.Unlock() + acct113Check(t, s, "before eviction") + if n := s.RunEviction(); n != 100 { + t.Fatalf("evicted %d transmissions, want the 100 old ones", n) + } + acct113Check(t, s, "after evicting the old half") + s.mu.RLock() + live := map[*StoreTx]bool{} + for _, tx := range s.packets { + live[tx] = true + } + for tx := range s.fallbackByNode { + if !live[tx] { + t.Fatalf("a fallback record of evicted tx %d is still held", tx.ID) + } + } + if len(s.fallbackByNode) == 0 || s.trackedBytes <= 0 { + t.Fatalf("setup: %d records and trackedBytes %d with half of the store live", len(s.fallbackByNode), s.trackedBytes) + } + s.mu.RUnlock() + + s.mu.Lock() + for _, tx := range s.packets { + tx.FirstSeen = time.Now().UTC().Add(-100 * time.Hour).Format(time.RFC3339) + } + s.mu.Unlock() + if n := s.RunEviction(); n != 100 { + t.Fatalf("second pass evicted %d, want 100", n) + } + if s.trackedBytes != 0 || len(s.fallbackByNode) != 0 || len(s.byNode) != 0 || len(s.nodeHashes) != 0 { + t.Fatalf("emptied store: trackedBytes=%d records=%d byNode=%d nodeHashes=%d, want all 0", + s.trackedBytes, len(s.fallbackByNode), len(s.byNode), len(s.nodeHashes)) + } +} + +// One call to indexObservationRelayHops charges each new (relay, tx) pair once. +// Repeating it, or a relay the tx is already indexed under, charges nothing. +func TestIndexObservationRelayHops_FallbackChargesOncePerRelay_202(t *testing.T) { + s := leak202Scenarios[0].build(t, false) // Load: no fallback relays yet + s.mu.Lock() + defer s.mu.Unlock() + tx := s.packets[0] + // Resolve against the full node set, so the paths below find their nodes. + var nodes []nodeInfo + for k := 0; k < 256; k++ { + nodes = append(nodes, nodeInfo{PublicKey: acct113PK(k), Role: "repeater"}) + } + pm := buildPrefixMap(nodes) + + var fresh []int // relays tx is not indexed under yet + for k := 0; k < 256 && len(fresh) < 3; k++ { + if !s.nodeHashes[acct113PK(k)][tx.Hash] { + fresh = append(fresh, k) + } + } + path := func(ks ...int) string { + j := "[" + for i, k := range ks { + if i > 0 { + j += "," + } + j += fmt.Sprintf(`"%02x"`, k) + } + return j + "]" + } + hopsSeen := map[string]bool{} + index := func(pathJSON string) int64 { + before := s.trackedBytes + s.indexObservationRelayHops(tx, persistedRelayPath{}, pathJSON, "", pm, hopsSeen, nil) + return s.trackedBytes - before + } + + if d, want := index(path(fresh[0], fresh[1])), fallbackRelayBytes(2); d != want { + t.Fatalf("two new relays charged %d, want %d (a record and two relays)", d, want) + } + if d := index(path(fresh[0], fresh[1])); d != 0 { + t.Errorf("the same relays again charged %d, want 0", d) + } + if d, want := index(path(fresh[1], fresh[2])), fallbackRelayBytes(3)-fallbackRelayBytes(2); d != want { + t.Errorf("one more relay charged %d, want %d (one relay, no second record)", d, want) + } + if n := len(s.fallbackByNode[tx]); n != 3 { + t.Errorf("record holds %d relays, want 3", n) + } + + // A relay the tx is already indexed under (a decoded pubkey, say) is not + // the fallback's: no charge, no record. + other := s.packets[1] + var held int + for k := 0; k < 256; k++ { + if !s.nodeHashes[acct113PK(k)][other.Hash] { + held = k + break + } + } + s.addToByNode(other, acct113PK(held)) + before := s.trackedBytes + s.indexObservationRelayHops(other, persistedRelayPath{}, path(held), "", pm, hopsSeen, nil) + if d := s.trackedBytes - before; d != 0 || len(s.fallbackByNode[other]) != 0 { + t.Errorf("a relay already indexed charged %d and recorded %d, want 0 and 0", d, len(s.fallbackByNode[other])) + } +} From 4073d408962ecf65b2f87f116a6ceff800659b01 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:33:02 +0200 Subject: [PATCH 3/3] test(store): env-gated timing and heap check for the #202 paths With CORESCOPE_PERF_202= the tests time the path_json fallback Load and eviction and the hash-migration merge, and compare the fallback's charge with its marginal heap. They skip otherwise and use only functions that exist on master, so the same file builds against it for before/after runs. Relates to #202. Co-Authored-By: Claude Sonnet 5.5 --- cmd/server/tracked_bytes_perf_202_test.go | 129 ++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 cmd/server/tracked_bytes_perf_202_test.go diff --git a/cmd/server/tracked_bytes_perf_202_test.go b/cmd/server/tracked_bytes_perf_202_test.go new file mode 100644 index 000000000..96c7ea8a8 --- /dev/null +++ b/cmd/server/tracked_bytes_perf_202_test.go @@ -0,0 +1,129 @@ +package main + +import ( + "fmt" + "os" + "sort" + "strconv" + "testing" + "time" +) + +// Issue #202 perf check. Not a pass/fail test: with CORESCOPE_PERF_202= +// it times the two paths this change touches and prints one RESULT line per +// measurement, so the same file can be compiled against master +// (go test -c, then run the two binaries alternately): +// +// CORESCOPE_PERF_202=7 ./server.test -test.run TestPerf_202 -test.v +// +// It uses only functions that exist on master. +func perf202Runs(t *testing.T) int { + n, _ := strconv.Atoi(os.Getenv("CORESCOPE_PERF_202")) + if n <= 0 { + t.Skip("set CORESCOPE_PERF_202= to measure") + } + return n +} + +func perf202Median(d []time.Duration) time.Duration { + sort.Slice(d, func(i, j int) bool { return d[i] < d[j] }) + return d[len(d)/2] +} + +// The path_json fallback: Load of 5000 transmissions whose hops all resolve +// through it, and then the eviction of all of them. +func TestPerf_FallbackLoadAndEvict_202(t *testing.T) { + runs := perf202Runs(t) + dbPath := leak202Fixture(t, 5000, 5000) + var load, evict []time.Duration + var tracked int64 + for i := 0; i < runs; i++ { + s := leak202Open(t, dbPath, true, &PacketStoreConfig{}) + t0 := time.Now() + if err := s.Load(); err != nil { + t.Fatal(err) + } + load = append(load, time.Since(t0)) + leak202Ready(t, s) + tracked = s.trackedBytes + s.mu.Lock() + s.retentionHours = 24 + s.mu.Unlock() + t0 = time.Now() + if n := s.RunEviction(); n != 5000 { + t.Fatalf("evicted %d, want 5000", n) + } + evict = append(evict, time.Since(t0)) + } + fmt.Printf("RESULT fallback-load-5000 median_ns=%d trackedBytes=%d\n", perf202Median(load).Nanoseconds(), tracked) + fmt.Printf("RESULT fallback-evict-5000 median_ns=%d\n", perf202Median(evict).Nanoseconds()) +} + +// The fallback's charge against the heap it really holds: the marginal heap +// of a store whose hops resolve through the fallback, over the same store +// whose hops resolve nothing, and the marginal trackedBytes. On master the +// second number is 0. +func TestPerf_FallbackChargeVsHeap_202(t *testing.T) { + runs := perf202Runs(t) + dbPath := leak202Fixture(t, 5000, 0) + build := func(withNodes bool) func() *PacketStore { + return func() *PacketStore { + s := leak202Open(t, dbPath, withNodes, &PacketStoreConfig{}) + if err := s.Load(); err != nil { + t.Fatal(err) + } + leak202Ready(t, s) + return s + } + } + for i := 0; i < runs; i++ { + with, hWith := heapOf(build(true)) + without, hWithout := heapOf(build(false)) + fmt.Printf("RESULT fallback-charge-vs-heap-5000 run=%d heap_delta=%d charge_delta=%d\n", + i, hWith-hWithout, with.trackedBytes-without.trackedBytes) + } +} + +// Hash migration with 1000 duplicate pairs (2000 transmissions, 2 observations +// each) that all collide and merge. +func TestPerf_HashMigrateMerge_202(t *testing.T) { + runs := perf202Runs(t) + var migrate []time.Duration + for i := 0; i < runs; i++ { + db := setupTestDBv2(t) + tx, err := db.conn.Begin() + if err != nil { + t.Fatal(err) + } + obsID := 0 + for id := 1; id <= 2000; id++ { + raw := fmt.Sprintf("0A00D69FD7A5A7475DB07337749AE61FA53A4788%04X", (id+1)/2) + if _, err := tx.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES (?, ?, ?, '2026-01-01T00:00:00Z', 1, 5, '{}')`, id, raw, fmt.Sprintf("old-%d", id)); err != nil { + t.Fatal(err) + } + for k := 0; k < 2; k++ { + obsID++ + if _, err := tx.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, path_json, timestamp) + VALUES (?, ?, 'observer', 'Observer', '["AA","BB"]', ?)`, obsID, id, obsID); err != nil { + t.Fatal(err) + } + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + store := NewPacketStore(db, nil) + if err := store.Load(); err != nil { + t.Fatal(err) + } + t0 := time.Now() + migrateContentHashesAsync(store, 500, 0) + migrate = append(migrate, time.Since(t0)) + if len(store.packets) != 2000 { + t.Fatalf("%d packets after the migration, want 2000", len(store.packets)) + } + db.conn.Close() + } + fmt.Printf("RESULT hash-migrate-merge-2000 median_ns=%d\n", perf202Median(migrate).Nanoseconds()) +}