feat: implement packet store eviction/aging to prevent OOM (#273)

## Summary

The in-memory `PacketStore` had **no eviction or aging** — it grew
unbounded until OOM killed the process. At ~3K packets/hour and ~5KB per
packet (not the 450 bytes previously estimated), an 8GB VM would OOM in
a few days.

## Changes

### Time-based eviction
- Configurable via `config.json`: `"packetStore": { "retentionHours": 24
}`
- Packets older than the retention window are evicted from the head of
the sorted slice

### Memory-based cap
- Configurable via `"packetStore": { "maxMemoryMB": 1024 }`
- Hard ceiling — evicts oldest packets when estimated memory exceeds the
cap

### Index cleanup
When a `StoreTx` is evicted, ALL associated data is removed from:
- `byHash`, `byTxID`, `byObsID`, `byObserver`, `byNode`, `byPayloadType`
- `nodeHashes`, `distHops`, `distPaths`, `spIndex`

### Periodic execution
- Background ticker runs eviction every 60 seconds
- Analytics caches and hash size cache are invalidated after eviction

### Stats fixes
- `estimatedMB` now uses ~5KB/packet + ~500B/observation (was 430B +
200B)
- `evicted` counter reflects actual evictions (was hardcoded to 0)
- Removed fake `maxPackets: 2386092` and `maxMB: 1024` from stats

### Config example
```json
{
  "packetStore": {
    "retentionHours": 24,
    "maxMemoryMB": 1024
  }
}
```

Both values default to 0 (unlimited) for backward compatibility.

## Tests
- 7 new tests in `eviction_test.go` covering time-based, memory-based,
index cleanup, thread safety, config parsing, and no-op when disabled
- All existing tests pass unchanged

Co-authored-by: Kpa-clawbot <kpabap+clawdbot@gmail.com>
This commit is contained in:
Kpa-clawbot
2026-03-30 03:42:11 +00:00
committed by GitHub
co-authored by Kpa-clawbot
parent a555b68915
commit 77d8f35a04
8 changed files with 568 additions and 80 deletions
+8
View File
@@ -45,6 +45,14 @@ type Config struct {
CacheTTL map[string]interface{} `json:"cacheTTL"`
Retention *RetentionConfig `json:"retention,omitempty"`
PacketStore *PacketStoreConfig `json:"packetStore,omitempty"`
}
// PacketStoreConfig controls in-memory packet store limits.
type PacketStoreConfig struct {
RetentionHours float64 `json:"retentionHours"` // max age of packets in hours (0 = unlimited)
MaxMemoryMB int `json:"maxMemoryMB"` // hard memory ceiling in MB (0 = unlimited)
}
type RetentionConfig struct {
+62 -62
View File
@@ -259,7 +259,7 @@ func TestStoreQueryMultiNodePackets(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
t.Run("empty pubkeys", func(t *testing.T) {
@@ -313,7 +313,7 @@ func TestIngestNewFromDB(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
initialMax := store.MaxTransmissionID()
@@ -384,7 +384,7 @@ func TestIngestNewFromDBv2(t *testing.T) {
db := setupTestDBv2(t)
defer db.Close()
seedV2Data(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
initialMax := store.MaxTransmissionID()
@@ -412,7 +412,7 @@ func TestMaxTransmissionID(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
maxID := store.MaxTransmissionID()
@@ -421,7 +421,7 @@ func TestMaxTransmissionID(t *testing.T) {
}
t.Run("empty store", func(t *testing.T) {
emptyStore := NewPacketStore(db)
emptyStore := NewPacketStore(db, nil)
if emptyStore.MaxTransmissionID() != 0 {
t.Error("expected 0 for empty store")
}
@@ -599,7 +599,7 @@ func TestTransmissionsForObserverIndex(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Query packets for an observer — hits the byObserver index
@@ -622,7 +622,7 @@ func TestGetChannelMessagesFromStore(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Test channel should exist from seed data
@@ -675,7 +675,7 @@ func TestGetChannelMessagesDedupe(t *testing.T) {
db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp)
VALUES (4, 2, 9.0, -93, '[]', ?)`, epoch)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
msgs, total := store.GetChannelMessages("#test", 100, 0)
@@ -692,7 +692,7 @@ func TestGetChannelsFromStore(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
channels := store.GetChannels("")
@@ -872,7 +872,7 @@ func TestPickBestObservation(t *testing.T) {
func TestIndexByNode(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
t.Run("empty decoded_json", func(t *testing.T) {
tx := &StoreTx{Hash: "h1"}
@@ -973,7 +973,7 @@ func TestPollerStartWithStore(t *testing.T) {
defer db.Close()
seedTestData(t, db)
hub := NewHub()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
poller := NewPoller(db, hub, 50*time.Millisecond)
@@ -1000,7 +1000,7 @@ func TestPerfMiddlewareSlowQuery(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
@@ -1339,7 +1339,7 @@ func TestStoreQueryPacketsEdgeCases(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
t.Run("hash filter", func(t *testing.T) {
@@ -1654,7 +1654,7 @@ func TestStorePerfAndCacheStats(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
stats := store.GetPerfStoreStats()
@@ -1674,7 +1674,7 @@ func TestEnrichObs(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Find an observation from the loaded store
@@ -1928,7 +1928,7 @@ func TestStoreGetTimestamps(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
ts := store.GetTimestamps("2000-01-01")
@@ -1983,7 +1983,7 @@ func setupRichTestDB(t *testing.T) *DB {
func TestStoreGetBulkHealthWithStore(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
results := store.GetBulkHealth(50, "")
@@ -2009,7 +2009,7 @@ func TestStoreGetBulkHealthWithStore(t *testing.T) {
func TestStoreGetAnalyticsHashSizes(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsHashSizes("")
@@ -2031,7 +2031,7 @@ func TestStoreGetAnalyticsHashSizes(t *testing.T) {
func TestStoreGetAnalyticsSubpaths(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsSubpaths("", 2, 8, 100)
@@ -2048,7 +2048,7 @@ func TestStoreGetAnalyticsSubpaths(t *testing.T) {
func TestSubpathPrecomputedIndex(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// After Load(), the precomputed index must be populated.
@@ -2102,7 +2102,7 @@ func TestSubpathPrecomputedIndex(t *testing.T) {
func TestStoreGetAnalyticsRFCacheHit(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// First call — cache miss
@@ -2128,7 +2128,7 @@ func TestStoreGetAnalyticsRFCacheHit(t *testing.T) {
func TestStoreGetAnalyticsTopology(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsTopology("")
@@ -2158,7 +2158,7 @@ func TestStoreGetAnalyticsTopology(t *testing.T) {
func TestStoreGetAnalyticsChannels(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsChannels("")
@@ -2205,7 +2205,7 @@ func TestStoreGetAnalyticsChannelsNumericHash(t *testing.T) {
db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp)
VALUES (6, 1, 12.0, -88, '[]', ?)`, recentEpoch)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsChannels("")
@@ -2250,7 +2250,7 @@ func TestStoreGetAnalyticsChannelsNumericHash(t *testing.T) {
func TestStoreGetAnalyticsDistance(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetAnalyticsDistance("")
@@ -2267,7 +2267,7 @@ func TestStoreGetAnalyticsDistance(t *testing.T) {
func TestStoreGetSubpathDetail(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.GetSubpathDetail([]string{"aabb", "ccdd"})
@@ -2287,7 +2287,7 @@ func TestHandleAnalyticsRFWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2318,7 +2318,7 @@ func TestHandleBulkHealthWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2338,7 +2338,7 @@ func TestHandleAnalyticsSubpathsWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2358,7 +2358,7 @@ func TestHandleAnalyticsSubpathDetailWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2378,7 +2378,7 @@ func TestHandleAnalyticsDistanceWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2398,7 +2398,7 @@ func TestHandleAnalyticsHashSizesWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2418,7 +2418,7 @@ func TestHandleAnalyticsTopologyWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2438,7 +2438,7 @@ func TestHandleAnalyticsChannelsWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2457,7 +2457,7 @@ func TestHandleAnalyticsChannelsWithStore(t *testing.T) {
func TestGetChannelMessagesRichData(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
messages, total := store.GetChannelMessages("#test", 100, 0)
@@ -2502,7 +2502,7 @@ func TestHandleChannelMessagesWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2524,7 +2524,7 @@ func TestHandleChannelsWithStore(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -2556,7 +2556,7 @@ func TestStoreGetStoreStats(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
stats, err := store.GetStoreStats()
@@ -2574,7 +2574,7 @@ func TestStoreQueryGroupedPackets(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.QueryGroupedPackets(PacketQuery{Limit: 50, Order: "DESC"})
@@ -2589,7 +2589,7 @@ func TestStoreGetPacketByHash(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
pkt := store.GetPacketByHash("abc123def4567890")
@@ -2630,7 +2630,7 @@ func TestResolvePayloadTypeNameUnknown(t *testing.T) {
func TestCacheHitTopology(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// First call — cache miss
@@ -2655,7 +2655,7 @@ func TestCacheHitTopology(t *testing.T) {
func TestCacheHitHashSizes(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
r1 := store.GetAnalyticsHashSizes("")
@@ -2678,7 +2678,7 @@ func TestCacheHitHashSizes(t *testing.T) {
func TestCacheHitChannels(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
r1 := store.GetAnalyticsChannels("")
@@ -2701,7 +2701,7 @@ func TestCacheHitChannels(t *testing.T) {
func TestGetChannelMessagesEdgeCases(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Channel not found — empty result
@@ -2735,7 +2735,7 @@ func TestFilterPacketsEmptyRegion(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Region with no observers → empty result
@@ -2749,7 +2749,7 @@ func TestFilterPacketsSinceUntil(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Since far future → empty
@@ -2776,7 +2776,7 @@ func TestFilterPacketsHashOnly(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Single hash fast-path — found
@@ -2796,7 +2796,7 @@ func TestFilterPacketsObserverWithType(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Observer + type filter (takes non-indexed path)
@@ -2809,7 +2809,7 @@ func TestFilterPacketsNodeFilter(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Node filter — exercises DecodedJSON containment check
@@ -2892,7 +2892,7 @@ func TestGetNodeHashSizeInfoEdgeCases(t *testing.T) {
db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp)
VALUES (10, 1, 10.0, -90, '[]', ?)`, recentEpoch)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
info := store.GetNodeHashSizeInfo()
@@ -3054,7 +3054,7 @@ func TestGetChannelMessagesDedupeRepeats(t *testing.T) {
db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp)
VALUES (3, 1, 10.0, -90, '[]', ?)`, recentEpoch)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
msgs, total := store.GetChannelMessages("#general", 10, 0)
@@ -3080,7 +3080,7 @@ func TestTransmissionsForObserverFromSlice(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Test with from=nil (index path) — for non-existent observer
@@ -3100,7 +3100,7 @@ func TestTransmissionsForObserverFromSlice(t *testing.T) {
func TestGetPerfStoreStatsPublicKeyField(t *testing.T) {
db := setupRichTestDB(t)
defer db.Close()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
stats := store.GetPerfStoreStats()
@@ -3142,7 +3142,7 @@ func TestStoreGetTransmissionByID(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
pkt := store.GetTransmissionByID(1)
@@ -3162,7 +3162,7 @@ func TestStoreGetPacketByID(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Get an observation ID from the store
@@ -3194,7 +3194,7 @@ func TestStoreGetObservationsForHash(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
obs := store.GetObservationsForHash("abc123def4567890")
@@ -3366,7 +3366,7 @@ func TestIngestNewFromDBDuplicateObs(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
initialMax := store.MaxTransmissionID()
@@ -3397,7 +3397,7 @@ func TestIngestNewObservations(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
// Get initial observation count for transmission 1 (hash abc123def4567890)
@@ -3475,7 +3475,7 @@ func TestIngestNewObservationsV2(t *testing.T) {
db := setupTestDBv2(t)
defer db.Close()
seedV2Data(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
tx := store.byHash["abc123def4567890"]
@@ -3546,7 +3546,7 @@ func TestHandleNodeAnalyticsNameless(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
srv.store = store
router := mux.NewRouter()
@@ -3588,7 +3588,7 @@ func TestStoreQueryPacketsRegionFilter(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
result := store.QueryPackets(PacketQuery{Region: "SJC", Limit: 50, Order: "DESC"})
@@ -3676,7 +3676,7 @@ func TestGetChannelMessagesAfterIngest(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
store.Load()
initialMax := store.MaxTransmissionID()
+252
View File
@@ -0,0 +1,252 @@
package main
import (
"fmt"
"sync/atomic"
"testing"
"time"
)
// makeTestStore creates a PacketStore with fake packets for eviction testing.
// It does NOT use a DB — indexes are populated manually.
func makeTestStore(count int, startTime time.Time, intervalMin int) *PacketStore {
store := &PacketStore{
packets: make([]*StoreTx, 0, count),
byHash: make(map[string]*StoreTx, count),
byTxID: make(map[int]*StoreTx, count),
byObsID: make(map[int]*StoreObs, count*2),
byObserver: make(map[string][]*StoreObs),
byNode: make(map[string][]*StoreTx),
nodeHashes: make(map[string]map[string]bool),
byPayloadType: make(map[int][]*StoreTx),
spIndex: make(map[string]int),
distHops: make([]distHopRecord, 0),
distPaths: make([]distPathRecord, 0),
rfCache: make(map[string]*cachedResult),
topoCache: make(map[string]*cachedResult),
hashCache: make(map[string]*cachedResult),
chanCache: make(map[string]*cachedResult),
distCache: make(map[string]*cachedResult),
subpathCache: make(map[string]*cachedResult),
rfCacheTTL: 15 * time.Second,
}
obsID := 1000
for i := 0; i < count; i++ {
ts := startTime.Add(time.Duration(i*intervalMin) * time.Minute)
hash := fmt.Sprintf("hash%04d", i)
txID := i + 1
pt := 4 // ADVERT
decodedJSON := fmt.Sprintf(`{"pubKey":"pk%04d"}`, i)
tx := &StoreTx{
ID: txID,
Hash: hash,
FirstSeen: ts.UTC().Format(time.RFC3339),
PayloadType: &pt,
DecodedJSON: decodedJSON,
PathJSON: `["aa","bb","cc"]`,
}
// Add 2 observations per tx
for j := 0; j < 2; j++ {
obsID++
obsIDStr := fmt.Sprintf("obs%d", j)
obs := &StoreObs{
ID: obsID,
TransmissionID: txID,
ObserverID: obsIDStr,
ObserverName: fmt.Sprintf("Observer%d", j),
Timestamp: ts.UTC().Format(time.RFC3339),
}
tx.Observations = append(tx.Observations, obs)
tx.ObservationCount++
store.byObsID[obsID] = obs
store.byObserver[obsIDStr] = append(store.byObserver[obsIDStr], obs)
store.totalObs++
}
store.packets = append(store.packets, tx)
store.byHash[hash] = tx
store.byTxID[txID] = tx
store.byPayloadType[pt] = append(store.byPayloadType[pt], tx)
// Index by node
pk := fmt.Sprintf("pk%04d", i)
if store.nodeHashes[pk] == nil {
store.nodeHashes[pk] = make(map[string]bool)
}
store.nodeHashes[pk][hash] = true
store.byNode[pk] = append(store.byNode[pk], tx)
// Add to distance index
store.distHops = append(store.distHops, distHopRecord{tx: tx, Hash: hash})
store.distPaths = append(store.distPaths, distPathRecord{tx: tx, Hash: hash})
// Subpath index
addTxToSubpathIndex(store.spIndex, tx)
}
return store
}
func TestEvictStale_TimeBasedEviction(t *testing.T) {
now := time.Now().UTC()
// 100 packets: first 50 are 48h old, last 50 are 1h old
store := makeTestStore(100, now.Add(-48*time.Hour), 0)
// Override: set first 50 to 48h ago, last 50 to 1h ago
for i := 0; i < 50; i++ {
store.packets[i].FirstSeen = now.Add(-48 * time.Hour).Format(time.RFC3339)
}
for i := 50; i < 100; i++ {
store.packets[i].FirstSeen = now.Add(-1 * time.Hour).Format(time.RFC3339)
}
store.retentionHours = 24
evicted := store.EvictStale()
if evicted != 50 {
t.Fatalf("expected 50 evicted, got %d", evicted)
}
if len(store.packets) != 50 {
t.Fatalf("expected 50 remaining, got %d", len(store.packets))
}
if len(store.byHash) != 50 {
t.Fatalf("expected 50 in byHash, got %d", len(store.byHash))
}
if len(store.byTxID) != 50 {
t.Fatalf("expected 50 in byTxID, got %d", len(store.byTxID))
}
// 50 remaining * 2 obs each = 100 obs
if store.totalObs != 100 {
t.Fatalf("expected 100 obs remaining, got %d", store.totalObs)
}
if len(store.byObsID) != 100 {
t.Fatalf("expected 100 in byObsID, got %d", len(store.byObsID))
}
if atomic.LoadInt64(&store.evicted) != 50 {
t.Fatalf("expected evicted counter=50, got %d", atomic.LoadInt64(&store.evicted))
}
// Verify evicted hashes are gone
if _, ok := store.byHash["hash0000"]; ok {
t.Fatal("hash0000 should have been evicted")
}
// Verify remaining hashes exist
if _, ok := store.byHash["hash0050"]; !ok {
t.Fatal("hash0050 should still exist")
}
// Verify distance indexes cleaned
if len(store.distHops) != 50 {
t.Fatalf("expected 50 distHops, got %d", len(store.distHops))
}
if len(store.distPaths) != 50 {
t.Fatalf("expected 50 distPaths, got %d", len(store.distPaths))
}
}
func TestEvictStale_NoEvictionWhenDisabled(t *testing.T) {
now := time.Now().UTC()
store := makeTestStore(10, now.Add(-48*time.Hour), 60)
// No retention set (defaults to 0)
evicted := store.EvictStale()
if evicted != 0 {
t.Fatalf("expected 0 evicted, got %d", evicted)
}
if len(store.packets) != 10 {
t.Fatalf("expected 10 remaining, got %d", len(store.packets))
}
}
func TestEvictStale_MemoryBasedEviction(t *testing.T) {
now := time.Now().UTC()
// Create enough packets to exceed a small memory limit
// 1000 packets * 5KB + 2000 obs * 500B ≈ 6MB
store := makeTestStore(1000, now.Add(-1*time.Hour), 0)
// All packets are recent (1h old) so time-based won't trigger
store.retentionHours = 24
store.maxMemoryMB = 3 // ~3MB limit, should evict roughly half
evicted := store.EvictStale()
if evicted == 0 {
t.Fatal("expected some evictions for memory cap")
}
// After eviction, estimated memory should be <= 3MB
estMB := store.estimatedMemoryMB()
if estMB > 3.5 { // small tolerance
t.Fatalf("expected <=3.5MB after eviction, got %.1fMB", estMB)
}
}
func TestEvictStale_CleansNodeIndexes(t *testing.T) {
now := time.Now().UTC()
store := makeTestStore(10, now.Add(-48*time.Hour), 0)
store.retentionHours = 24
// Verify node indexes exist before eviction
if len(store.byNode) != 10 {
t.Fatalf("expected 10 nodes indexed, got %d", len(store.byNode))
}
if len(store.nodeHashes) != 10 {
t.Fatalf("expected 10 nodeHashes, got %d", len(store.nodeHashes))
}
evicted := store.EvictStale()
if evicted != 10 {
t.Fatalf("expected 10 evicted, got %d", evicted)
}
// All should be cleaned
if len(store.byNode) != 0 {
t.Fatalf("expected 0 nodes, got %d", len(store.byNode))
}
if len(store.nodeHashes) != 0 {
t.Fatalf("expected 0 nodeHashes, got %d", len(store.nodeHashes))
}
if len(store.byPayloadType) != 0 {
t.Fatalf("expected 0 payload types, got %d", len(store.byPayloadType))
}
if len(store.byObserver) != 0 {
t.Fatalf("expected 0 observers, got %d", len(store.byObserver))
}
}
func TestEvictStale_RunEvictionThreadSafe(t *testing.T) {
now := time.Now().UTC()
store := makeTestStore(20, now.Add(-48*time.Hour), 0)
store.retentionHours = 24
evicted := store.RunEviction()
if evicted != 20 {
t.Fatalf("expected 20 evicted, got %d", evicted)
}
}
func TestStartEvictionTicker_NoopWhenDisabled(t *testing.T) {
store := &PacketStore{}
stop := store.StartEvictionTicker()
stop() // should not panic
}
func TestNewPacketStoreWithConfig(t *testing.T) {
cfg := &PacketStoreConfig{
RetentionHours: 48,
MaxMemoryMB: 512,
}
store := NewPacketStore(nil, cfg)
if store.retentionHours != 48 {
t.Fatalf("expected retentionHours=48, got %f", store.retentionHours)
}
if store.maxMemoryMB != 512 {
t.Fatalf("expected maxMemoryMB=512, got %d", store.maxMemoryMB)
}
}
func TestNewPacketStoreNilConfig(t *testing.T) {
store := NewPacketStore(nil, nil)
if store.retentionHours != 0 {
t.Fatalf("expected retentionHours=0, got %f", store.retentionHours)
}
}
+5 -1
View File
@@ -128,7 +128,7 @@ func main() {
}
// In-memory packet store
store := NewPacketStore(database)
store := NewPacketStore(database, cfg.PacketStore)
if err := store.Load(); err != nil {
log.Fatalf("[store] failed to load: %v", err)
}
@@ -164,6 +164,10 @@ func main() {
poller.store = store
go poller.Start()
// Start periodic eviction
stopEviction := store.StartEvictionTicker()
defer stopEviction()
// Graceful shutdown
httpServer := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.Port),
+1 -1
View File
@@ -409,7 +409,7 @@ func TestParityWSMultiObserverGolden(t *testing.T) {
defer db.Close()
seedTestData(t, db)
hub := NewHub()
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store load failed: %v", err)
}
+4 -4
View File
@@ -17,7 +17,7 @@ func setupTestServer(t *testing.T) (*Server, *mux.Router) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store.Load failed: %v", err)
}
@@ -1259,7 +1259,7 @@ func TestNodeAnalyticsNoNameNode(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store.Load failed: %v", err)
}
@@ -1295,7 +1295,7 @@ func TestNodeHealthForNoNameNode(t *testing.T) {
cfg := &Config{Port: 3000}
hub := NewHub()
srv := NewServer(db, cfg, hub)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store.Load failed: %v", err)
}
@@ -1890,7 +1890,7 @@ t.Error("hash_sizes_seen should not be set for single size")
func TestGetNodeHashSizeInfoFlipFlop(t *testing.T) {
db := setupTestDB(t)
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store.Load failed: %v", err)
}
+234 -10
View File
@@ -103,6 +103,11 @@ type PacketStore struct {
hashSizeInfoMu sync.Mutex
hashSizeInfoCache map[string]*hashSizeNodeInfo
hashSizeInfoAt time.Time
// Eviction config and stats
retentionHours float64 // 0 = unlimited
maxMemoryMB int // 0 = unlimited
evicted int64 // total packets evicted
}
// Precomputed distance records for fast analytics aggregation.
@@ -143,8 +148,8 @@ type cachedResult struct {
}
// NewPacketStore creates a new empty packet store backed by db.
func NewPacketStore(db *DB) *PacketStore {
return &PacketStore{
func NewPacketStore(db *DB, cfg *PacketStoreConfig) *PacketStore {
ps := &PacketStore{
db: db,
packets: make([]*StoreTx, 0, 65536),
byHash: make(map[string]*StoreTx, 65536),
@@ -163,6 +168,11 @@ func NewPacketStore(db *DB) *PacketStore {
rfCacheTTL: 15 * time.Second,
spIndex: make(map[string]int, 4096),
}
if cfg != nil {
ps.retentionHours = cfg.RetentionHours
ps.maxMemoryMB = cfg.MaxMemoryMB
}
return ps
}
// Load reads all transmissions + observations from SQLite into memory.
@@ -293,7 +303,7 @@ func (s *PacketStore) Load() error {
s.loaded = true
elapsed := time.Since(t0)
estMB := (len(s.packets)*450 + s.totalObs*100) / (1024 * 1024)
estMB := (len(s.packets)*5120 + s.totalObs*500) / (1024 * 1024)
log.Printf("[store] Loaded %d transmissions (%d observations) in %v (~%dMB est)",
len(s.packets), s.totalObs, elapsed, estMB)
return nil
@@ -542,20 +552,22 @@ func (s *PacketStore) GetPerfStoreStats() map[string]interface{} {
}
s.mu.RUnlock()
// Rough estimate: ~430 bytes per packet + ~200 per observation
estimatedMB := math.Round(float64(totalLoaded*430+totalObs*200)/1048576*10) / 10
// Realistic estimate: ~5KB per packet + ~500 bytes per observation
estimatedMB := math.Round(float64(totalLoaded*5120+totalObs*500)/1048576*10) / 10
evicted := atomic.LoadInt64(&s.evicted)
return map[string]interface{}{
"totalLoaded": totalLoaded,
"totalObservations": totalObs,
"evicted": 0,
"evicted": evicted,
"inserts": atomic.LoadInt64(&s.insertCount),
"queries": atomic.LoadInt64(&s.queryCount),
"inMemory": totalLoaded,
"sqliteOnly": false,
"maxPackets": 2386092,
"retentionHours": s.retentionHours,
"maxMemoryMB": s.maxMemoryMB,
"estimatedMB": estimatedMB,
"maxMB": 1024,
"indexes": map[string]interface{}{
"byHash": hashIdx,
"byTxID": txIdx,
@@ -648,12 +660,12 @@ func (s *PacketStore) GetPerfStoreStatsTyped() PerfPacketStoreStats {
}
s.mu.RUnlock()
estimatedMB := math.Round(float64(totalLoaded*430+totalObs*200)/1048576*10) / 10
estimatedMB := math.Round(float64(totalLoaded*5120+totalObs*500)/1048576*10) / 10
return PerfPacketStoreStats{
TotalLoaded: totalLoaded,
TotalObservations: totalObs,
Evicted: 0,
Evicted: int(atomic.LoadInt64(&s.evicted)),
Inserts: atomic.LoadInt64(&s.insertCount),
Queries: atomic.LoadInt64(&s.queryCount),
InMemory: totalLoaded,
@@ -1699,6 +1711,218 @@ func (s *PacketStore) buildDistanceIndex() {
len(s.distHops), len(s.distPaths))
}
// estimatedMemoryMB returns estimated memory usage of the packet store.
func (s *PacketStore) estimatedMemoryMB() float64 {
return float64(len(s.packets)*5120+s.totalObs*500) / 1048576.0
}
// EvictStale removes packets older than the retention window and/or exceeding
// the memory cap. Must be called with s.mu held (Lock). Returns the number of
// packets evicted.
func (s *PacketStore) EvictStale() int {
if s.retentionHours <= 0 && s.maxMemoryMB <= 0 {
return 0
}
cutoffIdx := 0
// Time-based eviction: find how many packets from the head are too old
if s.retentionHours > 0 {
cutoff := time.Now().UTC().Add(-time.Duration(s.retentionHours*3600) * time.Second).Format(time.RFC3339)
for cutoffIdx < len(s.packets) && s.packets[cutoffIdx].FirstSeen < cutoff {
cutoffIdx++
}
}
// Memory-based eviction: if still over budget, trim more from head
if s.maxMemoryMB > 0 {
for cutoffIdx < len(s.packets) && s.estimatedMemoryMB() > float64(s.maxMemoryMB) {
// Estimate how many more to evict: rough binary approach
overMB := s.estimatedMemoryMB() - float64(s.maxMemoryMB)
// ~5KB per packet, so overMB * 1024*1024 / 5120 packets
extra := int(overMB * 1048576.0 / 5120.0)
if extra < 100 {
extra = 100
}
cutoffIdx += extra
if cutoffIdx > len(s.packets) {
cutoffIdx = len(s.packets)
}
// Recalculate estimated memory with fewer packets
// (we haven't actually removed yet, so simulate)
remainingPkts := len(s.packets) - cutoffIdx
remainingObs := s.totalObs
for _, tx := range s.packets[:cutoffIdx] {
remainingObs -= len(tx.Observations)
}
estMB := float64(remainingPkts*5120+remainingObs*500) / 1048576.0
if estMB <= float64(s.maxMemoryMB) {
break
}
}
}
if cutoffIdx == 0 {
return 0
}
if cutoffIdx > len(s.packets) {
cutoffIdx = len(s.packets)
}
evicting := s.packets[:cutoffIdx]
evictedObs := 0
// Remove from all indexes
for _, tx := range evicting {
delete(s.byHash, tx.Hash)
delete(s.byTxID, tx.ID)
// Remove observations from indexes
for _, obs := range tx.Observations {
delete(s.byObsID, obs.ID)
// Remove from byObserver
if obs.ObserverID != "" {
obsList := s.byObserver[obs.ObserverID]
for i, o := range obsList {
if o.ID == obs.ID {
s.byObserver[obs.ObserverID] = append(obsList[:i], obsList[i+1:]...)
break
}
}
if len(s.byObserver[obs.ObserverID]) == 0 {
delete(s.byObserver, obs.ObserverID)
}
}
evictedObs++
}
// Remove from byPayloadType
if tx.PayloadType != nil {
pt := *tx.PayloadType
ptList := s.byPayloadType[pt]
for i, t := range ptList {
if t.ID == tx.ID {
s.byPayloadType[pt] = append(ptList[:i], ptList[i+1:]...)
break
}
}
if len(s.byPayloadType[pt]) == 0 {
delete(s.byPayloadType, pt)
}
}
// Remove from byNode and nodeHashes
if tx.DecodedJSON != "" {
var decoded map[string]interface{}
if json.Unmarshal([]byte(tx.DecodedJSON), &decoded) == nil {
for _, field := range []string{"pubKey", "destPubKey", "srcPubKey"} {
if v, ok := decoded[field].(string); ok && v != "" {
if hashes, ok := s.nodeHashes[v]; ok {
delete(hashes, tx.Hash)
if len(hashes) == 0 {
delete(s.nodeHashes, v)
}
}
// Remove tx from byNode
nodeList := s.byNode[v]
for i, t := range nodeList {
if t.ID == tx.ID {
s.byNode[v] = append(nodeList[:i], nodeList[i+1:]...)
break
}
}
if len(s.byNode[v]) == 0 {
delete(s.byNode, v)
}
}
}
}
}
// Remove from subpath index
removeTxFromSubpathIndex(s.spIndex, tx)
}
// Remove from distance indexes — filter out records referencing evicted txs
evictedTxSet := make(map[*StoreTx]bool, cutoffIdx)
for _, tx := range evicting {
evictedTxSet[tx] = true
}
newDistHops := s.distHops[:0]
for i := range s.distHops {
if !evictedTxSet[s.distHops[i].tx] {
newDistHops = append(newDistHops, s.distHops[i])
}
}
s.distHops = newDistHops
newDistPaths := s.distPaths[:0]
for i := range s.distPaths {
if !evictedTxSet[s.distPaths[i].tx] {
newDistPaths = append(newDistPaths, s.distPaths[i])
}
}
s.distPaths = newDistPaths
// Trim packets slice
n := copy(s.packets, s.packets[cutoffIdx:])
s.packets = s.packets[:n]
s.totalObs -= evictedObs
evictCount := cutoffIdx
atomic.AddInt64(&s.evicted, int64(evictCount))
freedMB := float64(evictCount*5120+evictedObs*500) / 1048576.0
log.Printf("[store] Evicted %d packets older than %.0fh (freed ~%.1fMB estimated)",
evictCount, s.retentionHours, freedMB)
// Invalidate analytics caches
s.cacheMu.Lock()
s.rfCache = make(map[string]*cachedResult)
s.topoCache = make(map[string]*cachedResult)
s.hashCache = make(map[string]*cachedResult)
s.chanCache = make(map[string]*cachedResult)
s.distCache = make(map[string]*cachedResult)
s.subpathCache = make(map[string]*cachedResult)
s.cacheMu.Unlock()
// Invalidate hash size cache
s.hashSizeInfoMu.Lock()
s.hashSizeInfoCache = nil
s.hashSizeInfoMu.Unlock()
return evictCount
}
// RunEviction acquires the write lock and runs eviction. Safe to call from
// a goroutine. Returns evicted count.
func (s *PacketStore) RunEviction() int {
s.mu.Lock()
defer s.mu.Unlock()
return s.EvictStale()
}
// StartEvictionTicker starts a background goroutine that runs eviction every
// minute. Returns a stop function.
func (s *PacketStore) StartEvictionTicker() func() {
if s.retentionHours <= 0 && s.maxMemoryMB <= 0 {
return func() {} // no-op
}
ticker := time.NewTicker(1 * time.Minute)
done := make(chan struct{})
go func() {
for {
select {
case <-ticker.C:
s.RunEviction()
case <-done:
ticker.Stop()
return
}
}
}()
return func() { close(done) }
}
// computeDistancesForTx computes distance records for a single transmission.
func computeDistancesForTx(tx *StoreTx, nodeByPk map[string]*nodeInfo, repeaterSet map[string]bool, resolveHop func(string) *nodeInfo) ([]distHopRecord, *distPathRecord) {
pathHops := txGetParsedPath(tx)
+2 -2
View File
@@ -270,7 +270,7 @@ func TestPollerBroadcastsMultipleObservations(t *testing.T) {
}()
poller := NewPoller(db, hub, 50*time.Millisecond)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store load failed: %v", err)
}
@@ -359,7 +359,7 @@ func TestIngestNewObservationsBroadcast(t *testing.T) {
db := setupTestDB(t)
defer db.Close()
seedTestData(t, db)
store := NewPacketStore(db)
store := NewPacketStore(db, nil)
if err := store.Load(); err != nil {
t.Fatalf("store load failed: %v", err)
}