From 5a0bb3ffdea1ebe65d7b2cbbc48cebe2aeb745ea Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:15:45 +0000 Subject: [PATCH 1/2] test(store): reproduce background chunks breaking first_seen order (#114) Background chunks are windowed on last_seen (#1690), so a chunk can hold transmissions whose first_seen is newer than an old transmission that is in the hot set because it was heard again recently. loadChunk publishes the chunk with s.packets = append(localPackets, s.packets...). chunk_merge_order_114_test.go loads a real SQLite fixture through LoadChunked (1h hot window) and loadChunk, then checks order, indexes and eviction. On master 3 of 4 fail: the store is out of order after one and after two chunk merges, and retention eviction stops at the first in-window head entry, leaving a 30h-old transmission behind with a 25h retention. The memory-eviction control passes (the oldest chunk entry is at the head either way). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/chunk_merge_order_114_test.go | 203 +++++++++++++++++++++++ 1 file changed, 203 insertions(+) create mode 100644 cmd/server/chunk_merge_order_114_test.go diff --git a/cmd/server/chunk_merge_order_114_test.go b/cmd/server/chunk_merge_order_114_test.go new file mode 100644 index 000000000..1fda50489 --- /dev/null +++ b/cmd/server/chunk_merge_order_114_test.go @@ -0,0 +1,203 @@ +package main + +import ( + "database/sql" + "path/filepath" + "sort" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +// Issue #114: PacketStore.packets must stay ordered by first_seen ASC, +// because eviction walks it from the head and stops at the first +// transmission inside the retention window. Background chunks are windowed +// on last_seen (#1690), so a chunk can hold transmissions whose first_seen +// is NEWER than transmissions already in the store (an old transmission that +// was heard again recently is in the hot set). loadChunk published a chunk +// with s.packets = append(localPackets, s.packets...), which then leaves an +// out-of-order store and makes eviction stop early. + +type fixtureTx114 struct { + id int + hash string + firstSeen time.Time + lastSeen time.Time +} + +// createChunkOrderDB writes transmissions with explicit first_seen and +// last_seen (unix seconds, as the live schema stores it) and one observation +// each. +func createChunkOrderDB(t *testing.T, txs []fixtureTx114) string { + t.Helper() + dbPath := filepath.Join(t.TempDir(), "chunk-order.db") + conn, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + defer conn.Close() + exec := func(q string, args ...interface{}) { + if _, err := conn.Exec(q, args...); err != nil { + t.Fatalf("setup: %v\n%s", err, q) + } + } + // PREFLIGHT: async=true reason="test fixture: t.TempDir SQLite created from empty, never a real DB" + exec(`CREATE TABLE transmissions ( + id INTEGER PRIMARY KEY, raw_hex TEXT, hash TEXT, first_seen TEXT, + route_type INTEGER, payload_type INTEGER, payload_version INTEGER, + decoded_json TEXT, last_seen INTEGER)`) + // PREFLIGHT: async=true reason="test fixture, tmpdir DB" + exec(`CREATE TABLE observations ( + id INTEGER PRIMARY KEY, transmission_id INTEGER, observer_id TEXT, observer_name TEXT, + direction TEXT, snr REAL, rssi REAL, score INTEGER, path_json TEXT, timestamp TEXT, + raw_hex TEXT, resolved_path TEXT)`) + // PREFLIGHT: async=true reason="test fixture, tmpdir DB" + exec(`CREATE TABLE observers (rowid INTEGER PRIMARY KEY, id TEXT, name TEXT, iata TEXT)`) + // PREFLIGHT: async=true reason="test fixture, tmpdir DB" + exec(`CREATE TABLE nodes (public_key TEXT PRIMARY KEY, name TEXT, role TEXT, lat REAL, lon REAL, + last_seen TEXT, first_seen TEXT, advert_count INTEGER DEFAULT 0)`) + // PREFLIGHT: async=true reason="test fixture, tmpdir DB" + exec(`CREATE TABLE schema_version (version INTEGER)`) + exec(`INSERT INTO schema_version (version) VALUES (1)`) + for _, tx := range txs { + exec(`INSERT INTO transmissions VALUES (?,?,?,?,0,5,1,'{}',?)`, + tx.id, "aa", tx.hash, tx.firstSeen.UTC().Format(time.RFC3339), tx.lastSeen.Unix()) + exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, raw_hex, resolved_path) + VALUES (?,?,?,?,?,?,?,?,?,?,?,NULL)`, + tx.id, tx.id, "obs1", "Obs1", "RX", -10.0, -80.0, 5, `["aa"]`, tx.lastSeen.UTC().Format(time.RFC3339), "") + } + return dbPath +} + +func assertSortedByFirstSeen(t *testing.T, s *PacketStore) { + t.Helper() + if !sort.SliceIsSorted(s.packets, func(i, j int) bool { return s.packets[i].FirstSeen < s.packets[j].FirstSeen }) { + var order []string + for _, tx := range s.packets { + order = append(order, tx.Hash+"@"+tx.FirstSeen) + } + t.Fatalf("s.packets is not ordered by first_seen: %v", order) + } +} + +func assertEveryPacketIndexedOnce(t *testing.T, s *PacketStore, want int) { + t.Helper() + if len(s.packets) != want { + t.Fatalf("%d packets in the store, want %d", len(s.packets), want) + } + seen := map[int]bool{} + for _, tx := range s.packets { + if seen[tx.ID] { + t.Fatalf("tx %d is in s.packets twice", tx.ID) + } + seen[tx.ID] = true + if s.byHash[tx.Hash] != tx || s.byTxID[tx.ID] != tx { + t.Fatalf("tx %d is in s.packets without its byHash/byTxID entries", tx.ID) + } + } +} + +// chunkOrderFixture: two transmissions in the hot set (heard in the last +// hour), one of them first seen 30h ago, and a background chunk whose +// transmissions straddle it, with a first_seen tie on either side. +func chunkOrderFixture(now time.Time) []fixtureTx114 { + h := func(d time.Duration) time.Time { return now.Add(-d).Truncate(time.Second) } + return []fixtureTx114{ + {1, "hot-old-first-seen", h(30 * time.Hour), h(10 * time.Minute)}, + {2, "hot-new", h(20 * time.Minute), h(5 * time.Minute)}, + {3, "chunk-40h", h(40 * time.Hour), h(20 * time.Hour)}, + {4, "chunk-20h", h(20 * time.Hour), h(5 * time.Hour)}, + {5, "chunk-35h", h(35 * time.Hour), h(30 * time.Hour)}, + {6, "chunk-tie-30h", h(30 * time.Hour), h(3 * time.Hour)}, + {7, "chunk-2h", h(2 * time.Hour), h(90 * time.Minute)}, + } +} + +func loadHotThenChunk(t *testing.T, now time.Time, txs []fixtureTx114) *PacketStore { + t.Helper() + db, err := OpenDB(createChunkOrderDB(t, txs)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.conn.Close() }) + if !db.hasLastSeen() { + t.Fatal("fixture: last_seen column not detected; chunks would not be windowed on last_seen") + } + s := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 72, HotStartupHours: 1}) + s.graph.Store(NewNeighborGraph()) + if err := s.LoadChunked(0); err != nil { + t.Fatalf("LoadChunked: %v", err) + } + if len(s.packets) != 2 { + t.Fatalf("fixture: hot load has %d packets, want the 2 heard in the last hour", len(s.packets)) + } + if err := s.loadChunk(now.Add(-72*time.Hour), now.Add(-time.Hour)); err != nil { + t.Fatalf("loadChunk: %v", err) + } + return s +} + +func TestBackgroundChunkMergeKeepsFirstSeenOrder_114(t *testing.T) { + now := time.Now().UTC() + s := loadHotThenChunk(t, now, chunkOrderFixture(now)) + assertSortedByFirstSeen(t, s) + assertEveryPacketIndexedOnce(t, s, 7) +} + +// Time-based eviction must remove every transmission older than retention, +// not stop at the first in-window one it meets at the head. +func TestEvictionAfterChunkMergeRemovesAllExpired_114(t *testing.T) { + now := time.Now().UTC() + s := loadHotThenChunk(t, now, chunkOrderFixture(now)) + s.retentionHours = 25 + s.mu.Lock() + n := s.EvictStale() + s.mu.Unlock() + cutoff := now.Add(-25 * time.Hour).Format(time.RFC3339) + for _, tx := range s.packets { + if tx.FirstSeen < cutoff { + t.Errorf("tx %s (first_seen %s) is older than retention but survived eviction", tx.Hash, tx.FirstSeen) + } + } + if n != 4 { + t.Errorf("evicted %d, want 4 (40h, 35h and both 30h)", n) + } + assertEveryPacketIndexedOnce(t, s, 3) +} + +// Memory-based eviction also evicts from the head: after a merge, the head +// must be the oldest transmissions. +func TestMemoryEvictionAfterChunkMergeTakesOldest_114(t *testing.T) { + now := time.Now().UTC() + s := loadHotThenChunk(t, now, chunkOrderFixture(now)) + s.retentionHours = 0 + s.maxMemoryMB = 1 + s.trackedBytes = 2 * 1048576 // over the watermark; the 25% cap evicts 1 of 7 + s.mu.Lock() + n := s.EvictStale() + s.mu.Unlock() + if n != 1 { + t.Fatalf("evicted %d, want 1", n) + } + if _, ok := s.byHash["chunk-40h"]; ok { + t.Error("memory eviction kept the oldest transmission (chunk-40h)") + } +} + +// A second chunk merged later (repeated last_seen activity moves rows +// between windows) keeps the order too. +func TestRepeatedChunkMergesKeepOrder_114(t *testing.T) { + now := time.Now().UTC() + txs := chunkOrderFixture(now) + txs = append(txs, + fixtureTx114{8, "late-chunk-25h", now.Add(-25 * time.Hour).Truncate(time.Second), now.Add(-80 * time.Hour)}, + fixtureTx114{9, "late-chunk-50h", now.Add(-50 * time.Hour).Truncate(time.Second), now.Add(-90 * time.Hour)}, + ) + s := loadHotThenChunk(t, now, txs) + if err := s.loadChunk(now.Add(-100*time.Hour), now.Add(-72*time.Hour)); err != nil { + t.Fatalf("second loadChunk: %v", err) + } + assertSortedByFirstSeen(t, s) + assertEveryPacketIndexedOnce(t, s, 9) +} From 1ca172d8578b773d270ed15ad451f285d68c71e9 Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:18:32 +0000 Subject: [PATCH 2/2] fix(store): merge background chunks into s.packets by first_seen (#114) loadChunk published a chunk with s.packets = append(localPackets, s.packets...). Chunks are windowed on last_seen, so a chunk overlaps the store by first_seen and the prepend left s.packets out of order; eviction walks from the head and stops at the first in-window entry, so it under-evicted. The final publication step now calls mergeByFirstSeen: existing entries older than the chunk are copied as one block (binary search), the overlap is merged linearly, the rest copied. O(existing + chunk) like the copy the prepend made, with O(log n + overlap) comparisons, one allocation, and no sort of the store under the write lock. The chunk itself is ordered by the query; loadChunk re-sorts it outside the lock only if it is not. Index-before-publication, route masks and the resolved/node indexes are untouched: only the publication line changed. Tests: random runs with ties (ordered, nothing lost or duplicated, inputs unchanged) and a comparison bound that a full sort fails (242,743 comparisons against a bound of 2,021). Benchmark, 20k chunk into a 500k store with a 5% overlap: merge about 3.1 ms, the old prepend about 2.7 ms, a full stable sort about 45 ms. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/chunk_merge_order_114_test.go | 129 +++++++++++++++++++++++ cmd/server/store.go | 52 ++++++++- 2 files changed, 178 insertions(+), 3 deletions(-) diff --git a/cmd/server/chunk_merge_order_114_test.go b/cmd/server/chunk_merge_order_114_test.go index 1fda50489..aa5a09106 100644 --- a/cmd/server/chunk_merge_order_114_test.go +++ b/cmd/server/chunk_merge_order_114_test.go @@ -2,8 +2,10 @@ package main import ( "database/sql" + "math/rand" "path/filepath" "sort" + "strconv" "testing" "time" @@ -201,3 +203,130 @@ func TestRepeatedChunkMergesKeepOrder_114(t *testing.T) { assertSortedByFirstSeen(t, s) assertEveryPacketIndexedOnce(t, s, 9) } + +func txsAt(prefix string, secs ...int) []*StoreTx { + base := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + out := make([]*StoreTx, len(secs)) + for i, s := range secs { + out[i] = &StoreTx{ID: len(prefix)*1000000 + i, Hash: prefix + strconv.Itoa(i), FirstSeen: base.Add(time.Duration(s) * time.Second).Format(time.RFC3339)} + } + return out +} + +func TestMergeByFirstSeenRandomRuns_114(t *testing.T) { + rng := rand.New(rand.NewSource(114)) + for trial := 0; trial < 2000; trial++ { + a := make([]int, rng.Intn(40)) + b := make([]int, rng.Intn(40)) + for i := range a { + a[i] = rng.Intn(60) // few distinct values: many ties + } + for i := range b { + b[i] = rng.Intn(60) + } + sort.Ints(a) + sort.Ints(b) + ex, in := txsAt("e", a...), txsAt("i", b...) + exCopy, inCopy := append([]*StoreTx(nil), ex...), append([]*StoreTx(nil), in...) + got := mergeByFirstSeen(ex, in) + if !sort.SliceIsSorted(got, func(i, j int) bool { return got[i].FirstSeen < got[j].FirstSeen }) { + t.Fatalf("trial %d: merged run out of order", trial) + } + if len(got) != len(ex)+len(in) { + t.Fatalf("trial %d: %d entries, want %d", trial, len(got), len(ex)+len(in)) + } + seen := map[*StoreTx]int{} + for _, tx := range got { + seen[tx]++ + } + for _, tx := range append(append([]*StoreTx(nil), ex...), in...) { + if seen[tx] != 1 { + t.Fatalf("trial %d: %s appears %d times", trial, tx.Hash, seen[tx]) + } + } + for i := range ex { + if ex[i] != exCopy[i] { + t.Fatal("existing run modified") + } + } + for i := range in { + if in[i] != inCopy[i] { + t.Fatal("incoming run modified") + } + } + } +} + +// The merge must never degrade into a comparison sort of the whole store +// under the write lock: it compares O(log n) times to find where the chunk +// starts, then at most once per entry of the overlapping part. +func TestMergeByFirstSeenComparisonBound_114(t *testing.T) { + const n, chunk = 200000, 2000 + secs := make([]int, n) + for i := range secs { + secs[i] = i * 10 + } + existing := txsAt("e", secs...) + for _, c := range []struct { + name string + start int // incoming first_seen offset, seconds + overlap int // existing entries at or after the chunk start + }{ + {"chunk older than the store", -chunk*10 - 5, n}, + {"chunk overlaps the last 1% of the store", (n - n/100) * 10, n / 100}, + {"chunk newer than the store", n*10 + 5, 0}, + } { + t.Run(c.name, func(t *testing.T) { + in := make([]int, chunk) + for i := range in { + in[i] = c.start + i*5 + } + incoming := txsAt("i", in...) + cmp := 0 + got := mergeSortedRuns(existing, incoming, func(a, b *StoreTx) bool { cmp++; return a.FirstSeen < b.FirstSeen }) + if len(got) != n+chunk { + t.Fatalf("%d entries", len(got)) + } + bound := 20 + min(c.overlap, chunk*2) + chunk // log2(200000) < 18 + if c.overlap == n { + bound = 20 + chunk + 1 + } + if cmp > bound { + t.Fatalf("%d comparisons, bound %d (a full sort would be ~%d)", cmp, bound, (n+chunk)*18) + } + }) + } +} + +// BenchmarkMergeChunkUnderLock_114 merges a one-day chunk (20k) into a +// production-shaped store (500k) whose newest 5% overlaps the chunk by +// first_seen; "fullSort" is what sorting the whole store would cost instead. +func BenchmarkMergeChunkUnderLock_114(b *testing.B) { + const n, chunk = 500000, 20000 + secs := make([]int, n) + for i := range secs { + secs[i] = i * 2 + } + existing := txsAt("e", secs...) + in := make([]int, chunk) + for i := range in { + in[i] = (n-n/20)*2 + i*5 + 1 + } + incoming := txsAt("i", in...) + b.Run("merge", func(b *testing.B) { + for i := 0; i < b.N; i++ { + mergeByFirstSeen(existing, incoming) + } + }) + b.Run("prepend(old,unordered)", func(b *testing.B) { + for i := 0; i < b.N; i++ { + _ = append(append(make([]*StoreTx, 0, n+chunk), incoming...), existing...) + } + }) + b.Run("fullSort", func(b *testing.B) { + for i := 0; i < b.N; i++ { + all := append(append(make([]*StoreTx, 0, n+chunk), incoming...), existing...) + sort.SliceStable(all, func(x, y int) bool { return all[x].FirstSeen < all[y].FirstSeen }) + } + }) +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 1852d3d54..72fd6b63a 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -1118,8 +1118,9 @@ func (s *PacketStore) Load() error { // write lock, builds local data structures, then merges them into the store // under s.mu.Lock(). It is the building block for the background loader. // -// The chunk is assumed to be older than the data already in the store, so -// localPackets are prepended to s.packets. +// Chunks are windowed on last_seen, so a chunk is older than the store by +// last_seen but can overlap it by first_seen. localPackets are merged into +// s.packets by first_seen (mergeByFirstSeen, #114). // // byPayloadType is updated here incrementally. byPathHop, spIndex, and // distHops are NOT updated here — the caller (loadBackgroundChunks) rebuilds @@ -1396,6 +1397,12 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { if len(localPackets) == 0 { return nil } + // The query orders by first_seen already; sorting the (one-window) + // chunk here, outside the lock, makes the merge below rely on nothing + // but this function. + if !sort.SliceIsSorted(localPackets, func(i, j int) bool { return localPackets[i].FirstSeen < localPackets[j].FirstSeen }) { + sort.SliceStable(localPackets, func(i, j int) bool { return localPackets[i].FirstSeen < localPackets[j].FirstSeen }) + } // PR #1187 r3 MUST-FIX 1: index↔slice consistency. // @@ -1512,7 +1519,11 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // critical section. After this point the new state is fully visible; // before it readers see the old slice (which is still fully indexed). s.mu.Lock() - s.packets = append(localPackets, s.packets...) + // Merge, not prepend: eviction walks s.packets from the head and stops + // at the first in-window transmission, so it must stay ordered by + // first_seen (#114). O(len(s.packets) + len(localPackets)), like the + // copy the prepend made; no comparison sort under the lock. + s.packets = mergeByFirstSeen(s.packets, localPackets) s.totalObs += localTotalObs s.trackedBytes += localTrackedBytes if localMaxTxID > s.maxTxID { @@ -1531,6 +1542,41 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { return nil } +// mergeByFirstSeen merges two runs that are each ordered by FirstSeen ASC +// into a new slice ordered by FirstSeen ASC (#114). Existing entries older +// than the first incoming one are copied as a block (found by binary +// search), the overlap is merged linearly, and the rest is copied, so the +// work is O(len(existing) + len(incoming)) with O(log n + overlap) +// comparisons. On equal timestamps the incoming entry goes first, as the +// old prepend did. Neither input is modified. +func mergeByFirstSeen(existing, incoming []*StoreTx) []*StoreTx { + return mergeSortedRuns(existing, incoming, func(a, b *StoreTx) bool { return a.FirstSeen < b.FirstSeen }) +} + +// mergeSortedRuns is mergeByFirstSeen with the order as a parameter (tests +// count comparisons through it). +func mergeSortedRuns(existing, incoming []*StoreTx, less func(a, b *StoreTx) bool) []*StoreTx { + out := make([]*StoreTx, 0, len(existing)+len(incoming)) + if len(incoming) == 0 { + return append(out, existing...) + } + // existing[:i] is strictly older than every incoming entry. + i := sort.Search(len(existing), func(k int) bool { return !less(existing[k], incoming[0]) }) + out = append(out, existing[:i]...) + j := 0 + for i < len(existing) && j < len(incoming) { + if less(existing[i], incoming[j]) { + out = append(out, existing[i]) + i++ + } else { + out = append(out, incoming[j]) + j++ + } + } + out = append(out, incoming[j:]...) + return append(out, existing[i:]...) +} + // loadBackgroundChunks fills the remaining retentionHours window by loading // daily chunks from oldestLoaded back to the retention cutoff. After all // chunks are merged it rebuilds analytics indexes once. Chunk errors are