fix(store): charge the path_json fallback and pin the hash-migration accounting (#202)

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]

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)

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)
 		}