From 738d5fef39ff3a86cfb8f48f2475902f5ca65d9e Mon Sep 17 00:00:00 2001
From: Kpa-clawbot <259247574+Kpa-clawbot@users.noreply.github.com>
Date: Tue, 31 Mar 2026 23:28:46 -0700
Subject: [PATCH] fix: poller uses store max IDs to prevent replaying entire DB
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When GetMaxTransmissionID() fails silently (e.g., corrupted DB returns 0
from COALESCE), the poller starts from ID 0 and replays the entire
database over WebSocket — broadcasting thousands of old packets per second.

Fix: after querying the DB, use the in-memory store's MaxTransmissionID
and MaxObservationID as a floor. Since Load() already read the full DB
successfully, the store has the correct max IDs.

Root cause discovered on staging: DB corruption caused MAX(id) query to
fail, returning 0. Poller log showed 'starting from transmission ID 0'
followed by 1000-2000 broadcasts per tick walking through 76K rows.

Also adds MaxObservationID() to PacketStore for observation cursor safety.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 cmd/server/store.go     | 14 ++++++++++++++
 cmd/server/websocket.go | 11 +++++++++++
 2 files changed, 25 insertions(+)

diff --git a/cmd/server/store.go b/cmd/server/store.go
index 1261ea24..d3a30cee 100644
--- a/cmd/server/store.go
+++ b/cmd/server/store.go
@@ -1344,6 +1344,20 @@ func (s *PacketStore) MaxTransmissionID() int {
 	return maxID
 }
 
+// MaxObservationID returns the highest observation ID in the store.
+func (s *PacketStore) MaxObservationID() int {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	maxID := 0
+	for id := range s.byObsID {
+		if id > maxID {
+			maxID = id
+		}
+	}
+	return maxID
+}
+
 // --- Internal filter/query helpers ---
 
 // filterPackets applies PacketQuery filters to the in-memory packet list.
diff --git a/cmd/server/websocket.go b/cmd/server/websocket.go
index e4696bc4..96f544e1 100644
--- a/cmd/server/websocket.go
+++ b/cmd/server/websocket.go
@@ -166,6 +166,17 @@ func NewPoller(db *DB, hub *Hub, interval time.Duration) *Poller {
 func (p *Poller) Start() {
 	lastID := p.db.GetMaxTransmissionID()
 	lastObsID := p.db.GetMaxObservationID()
+	// If the store already loaded data, use its max IDs as a floor.
+	// This prevents replaying the entire DB when the DB query fails
+	// (e.g., corrupted DB returns 0 from COALESCE).
+	if p.store != nil {
+		if storeMax := p.store.MaxTransmissionID(); storeMax > lastID {
+			lastID = storeMax
+		}
+		if storeMaxObs := p.store.MaxObservationID(); storeMaxObs > lastObsID {
+			lastObsID = storeMaxObs
+		}
+	}
 	log.Printf("[poller] starting from transmission ID %d, obs ID %d, interval %v", lastID, lastObsID, p.interval)
 
 	ticker := time.NewTicker(p.interval)