diff --git a/cmd/server/evict_order_test.go b/cmd/server/evict_order_test.go new file mode 100644 index 00000000..e0350ed3 --- /dev/null +++ b/cmd/server/evict_order_test.go @@ -0,0 +1,82 @@ +package main + +import ( + "fmt" + "testing" + "time" +) + +// s.packets is declared "sorted by first_seen ASC (oldest first; newest at +// tail)" (store.go:177), and retention eviction depends on it: it walks from +// the head and stops at the first transmission inside the window, so a slice +// out of order is under-evicted silently rather than noisily wrong. +// +// The background chunk loader broke that. Chunks are windowed on last_seen, so +// a transmission first heard weeks ago and heard again recently rides in with +// a recent chunk carrying its old first_seen; the chunk was put in front with +// `append(localPackets, s.packets...)` and never re-sorted, so a later, older +// chunk prepended in front of it left that ancient row behind newer ones. +// +// Not a corner case: on a production database 2071 of the 236080 transmissions +// in a 14 day window have a first_seen more than a day older than their +// last_seen, 1848 of them more than a week. + +// mergeChunkIntoPackets is what keeps that invariant true, so pin it directly. +func TestMergeChunkIntoPackets_KeepsFirstSeenOrder(t *testing.T) { + at := func(h int) string { return time.Now().UTC().Add(-time.Duration(h) * time.Hour).Format(time.RFC3339) } + + existing := []*StoreTx{ + {ID: 2, Hash: "b", FirstSeen: at(30)}, + {ID: 4, Hash: "d", FirstSeen: at(10)}, + } + // Interleaves with what is already loaded, and is deliberately not in + // order: the chunk query sorts, but the merge must not depend on it. + chunk := []*StoreTx{ + {ID: 3, Hash: "c", FirstSeen: at(20)}, + {ID: 1, Hash: "a", FirstSeen: at(40)}, + {ID: 5, Hash: "e", FirstSeen: at(5)}, + } + + merged := mergeChunkIntoPackets(chunk, existing) + + if len(merged) != 5 { + t.Fatalf("merged %d packets, want 5", len(merged)) + } + got := "" + for _, tx := range merged { + got += tx.Hash + } + if got != "abcde" { + t.Fatalf("merge order = %q, want %q", got, "abcde") + } + for i := 1; i < len(merged); i++ { + if merged[i-1].FirstSeen > merged[i].FirstSeen { + t.Fatalf("merge left the slice out of order at index %d", i) + } + } +} + +// The merge runs under s.mu once per chunk, so it must stay linear. This +// guards against it being "simplified" back into a sort of the whole slice, +// which on a loaded instance means sorting hundreds of thousands of packets +// while ingest waits for the lock. +func BenchmarkMergeChunkIntoPackets(b *testing.B) { + base := time.Now().UTC().Add(-14 * 24 * time.Hour) + mk := func(n, stride, off int) []*StoreTx { + out := make([]*StoreTx, n) + for i := 0; i < n; i++ { + out[i] = &StoreTx{ + ID: i, + Hash: fmt.Sprintf("h%06d", i), + FirstSeen: base.Add(time.Duration(i*stride+off) * time.Second).Format(time.RFC3339), + } + } + return out + } + existing := mk(200000, 6, 0) + chunk := mk(20000, 60, 3) + b.ResetTimer() + for i := 0; i < b.N; i++ { + mergeChunkIntoPackets(chunk, existing) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index 330ea5df..cee8c609 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -1446,7 +1446,7 @@ 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...) + s.packets = mergeChunkIntoPackets(localPackets, s.packets) s.totalObs += localTotalObs s.trackedBytes += localTrackedBytes if localMaxTxID > s.maxTxID { @@ -4743,6 +4743,47 @@ func (s *PacketStore) EvictStale() int { return s.evictStaleInternal(nil) } +// mergeChunkIntoPackets merges a background chunk into the packet slice while +// keeping the invariant s.packets is declared with: "sorted by first_seen ASC +// (oldest first; newest at tail)". Retention eviction depends on it, walking +// from the head and stopping at the first transmission inside the window, so a +// slice that is out of order is silently under-evicted rather than noisily +// wrong. +// +// The chunk cannot simply be put in front. Chunks are selected by last_seen, +// so a transmission first heard weeks ago and heard again recently arrives in +// a recent chunk carrying its old FirstSeen. On a production database 2071 of +// the 236080 transmissions in a 14 day window have a first_seen more than a +// day older than their last_seen, 1848 of them more than a week. +// +// Linear on purpose: this runs under s.mu once per chunk, and re-sorting the +// whole slice there would mean sorting hundreds of thousands of packets while +// ingest waits. The chunk itself is sorted first, which is the only +// comparison sort involved and is bounded by one chunk. LoadChunked does its +// own sort once at the end of the initial load; this keeps that invariant true +// for every chunk merged afterwards. +func mergeChunkIntoPackets(chunk, existing []*StoreTx) []*StoreTx { + less := func(i, j int) bool { return chunk[i].FirstSeen < chunk[j].FirstSeen } + if !sort.SliceIsSorted(chunk, less) { + sort.SliceStable(chunk, less) + } + + out := make([]*StoreTx, 0, len(chunk)+len(existing)) + i, j := 0, 0 + for i < len(chunk) && j < len(existing) { + if chunk[i].FirstSeen <= existing[j].FirstSeen { + out = append(out, chunk[i]) + i++ + } else { + out = append(out, existing[j]) + j++ + } + } + out = append(out, chunk[i:]...) + out = append(out, existing[j:]...) + return out +} + func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { if s.retentionHours <= 0 && s.maxMemoryMB <= 0 { return 0