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