fix(nodes): calculate analytics over available retained history

cmd/server/chunked_load.go

@@ -222,6 +222,10 @@ func (s *PacketStore) RunStartupLoad(chunkSize int) error {
 	s.bgErrMu.Unlock()
 
 	if err := s.LoadChunked(chunkSize); err != nil {
+		s.mu.Lock()
+		s.analyticsHistory.failed = true
+		s.analyticsHistory.loading = false
+		s.mu.Unlock()
 		// Pick a terminal state on the error path so /api/healthz +
 		// backgroundLoadComplete don't stay undefined forever. Both
 		// done AND failed are set true: the LoadChunked read-protocol
@@ -358,6 +362,9 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 			loadConditions = append(loadConditions, fmt.Sprintf("t2.first_seen >= '%s'", hotCutoffStr))
 		}
 	}
+	s.mu.Lock()
+	s.beginNodeAnalyticsLoadLocked(hotCutoffStr)
+	s.mu.Unlock()
 
 	// COUNT honours the same retention/hot-startup filter the chunk
 	// loop applies, so the logged "DB total" matches the rows the
@@ -522,6 +529,8 @@ func (s *PacketStore) LoadChunked(chunkSize int) error {
 	} else if len(s.packets) > 0 {
 		s.oldestLoaded = s.packets[0].FirstSeen
 	}
+	s.analyticsHistory.failed = s.analyticsHistory.failed || totalInDB < 0 || (maxPackets == 0 && totalLoaded < totalInDB)
+	s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, maxPackets > 0 && totalLoaded < totalInDB, s.hotStartupHours > 0)
 	s.loaded = true
 	s.mu.Unlock()
 
@@ -591,6 +600,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 		}
 		if err := rows.Scan(scanArgs...); err != nil {
 			log.Printf("[store] LoadChunked scan error: %v", err)
+			s.analyticsHistory.failed = true
 			continue
 		}
 		if s.loadScannedRowHook != nil {
@@ -674,6 +684,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 			tx.mergeObservedPathHashSize(obsPJ)
 			tx.Observations = append(tx.Observations, obs)
 			tx.obsKeys[dk] = true
+			s.analyticsHistory.observedFrom = earlierNodeAnalyticsEvidence(s.analyticsHistory.observedFrom, obs.Timestamp)
 			if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] {
 				tx.observerSet[obs.ObserverID] = true
 				tx.UniqueObserverCount++

cmd/server/node_analytics_characterization_test.go

@@ -40,6 +40,16 @@ func loadPacketsIntoStore(ps *PacketStore, pubkey string, packets []*StoreTx) {
 		ps.byTxID[p.ID] = p
 	}
 	ps.byNode[pubkey] = packets
+	// This synthetic fixture bypasses the loader. Certify the complete
+	// interval explicitly so this test still characterizes all non-coverage
+	// analytics fields rather than accidentally exercising a loading store.
+	ps.loaded = true
+	if len(packets) > 0 {
+		at, ok := parseNodeAnalyticsTime(packets[0].FirstSeen)
+		if ok {
+			ps.analyticsHistory = nodeAnalyticsHistory{known: true, from: at.Add(-365 * 24 * time.Hour)}
+		}
+	}
 }
 
 func mustParseRFC3339UTC(t *testing.T, s string) time.Time {

cmd/server/node_analytics_coverage.go

@@ -0,0 +1,228 @@
+package main
+
+import "time"
+
+// NodeAnalyticsCoverage describes available retained store data, not radio
+// uptime. The loader's contiguous time window is clamped to persisted
+// observation evidence; an old re-heard transmission's FirstSeen is never
+// evidence that the intervening history was loaded.
+type NodeAnalyticsCoverage struct {
+	From    string  `json:"from"`
+	To      string  `json:"to"`
+	Seconds float64 `json:"seconds"`
+	Partial bool    `json:"partial"`
+	Reason  string  `json:"reason"`
+	Loading bool    `json:"loading"`
+	// LateEvictedObservations counts observations (since startup) of evicted
+	// transmissions that were heard after the eviction edge. They are a known,
+	// bounded gap in retained evidence and never move From.
+	LateEvictedObservations int `json:"late_evicted_observations"`
+}
+
+// All fields are guarded by PacketStore.mu. This is constant-sized and is
+// updated only on load/eviction, never by a per-request SQL scan.
+type nodeAnalyticsHistory struct {
+	from         time.Time
+	known        bool
+	loading      bool
+	failed       bool
+	cap          bool
+	reason       string
+	loadCutoff   time.Time
+	evictedFrom  time.Time
+	observedFrom time.Time
+	lateEvicted  int
+}
+
+// parseCanonicalUTCSeconds is the hot-path parser for exactly
+// "YYYY-MM-DDTHH:MM:SSZ", the form the store writes. It rejects anything it
+// is not certain about (including out-of-range fields) so the caller can fall
+// back to time.Parse with identical results.
+func parseCanonicalUTCSeconds(raw string) (time.Time, bool) {
+	if len(raw) != 20 || raw[4] != '-' || raw[7] != '-' || raw[10] != 'T' ||
+		raw[13] != ':' || raw[16] != ':' || raw[19] != 'Z' {
+		return time.Time{}, false
+	}
+	num := func(i, n int) (int, bool) {
+		v := 0
+		for ; n > 0; n, i = n-1, i+1 {
+			c := raw[i]
+			if c < '0' || c > '9' {
+				return 0, false
+			}
+			v = v*10 + int(c-'0')
+		}
+		return v, true
+	}
+	y, ok1 := num(0, 4)
+	mo, ok2 := num(5, 2)
+	d, ok3 := num(8, 2)
+	h, ok4 := num(11, 2)
+	mi, ok5 := num(14, 2)
+	sec, ok6 := num(17, 2)
+	if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6) || mo < 1 || mo > 12 || d < 1 ||
+		h > 23 || mi > 59 || sec > 59 {
+		return time.Time{}, false
+	}
+	t := time.Date(y, time.Month(mo), d, h, mi, sec, 0, time.UTC)
+	if t.Day() != d { // day overflowed the month (e.g. Feb 30)
+		return time.Time{}, false
+	}
+	return t, true
+}
+
+func parseNodeAnalyticsTime(raw string) (time.Time, bool) {
+	if t, ok := parseCanonicalUTCSeconds(raw); ok {
+		return t, t.Unix() > 0
+	}
+	t, err := time.Parse(time.RFC3339Nano, raw)
+	if err != nil {
+		t, err = time.ParseInLocation("2006-01-02 15:04:05", raw, time.UTC)
+	}
+	return t.UTC(), err == nil && t.Unix() > 0
+}
+
+func (s *PacketStore) beginNodeAnalyticsLoadLocked(cutoff string) {
+	floor, _ := parseNodeAnalyticsTime(cutoff)
+	s.analyticsHistory = nodeAnalyticsHistory{loading: true, loadCutoff: floor}
+}
+
+func earlierNodeAnalyticsEvidence(earliest time.Time, raw string) time.Time {
+	if at, ok := parseNodeAnalyticsTime(raw); ok && (earliest.IsZero() || at.Before(earliest)) {
+		return at
+	}
+	return earliest
+}
+
+// finishNodeAnalyticsLoadLocked certifies only the interval actually read.
+// Earliest persisted observation is a conservative evidence edge, not proof
+// that the recorder was continuously running or that RF reception succeeded.
+// A capped/error load cannot certify its historical subset; only subsequent
+// live FirstSeen traffic is eligible, from completion onward.
+func (s *PacketStore) finishNodeAnalyticsLoadLocked(cutoff time.Time, capped, loading bool) {
+	h := &s.analyticsHistory
+	earliest := h.observedFrom
+	h.loading = loading
+	h.cap = h.cap || capped
+	if earliest.IsZero() {
+		h.known = false
+		h.from = time.Now().UTC()
+		h.reason = "unknown_history"
+		return
+	}
+	floor := earliest
+	if cutoff.After(floor) {
+		floor = cutoff
+	}
+	if h.cap || h.failed {
+		floor = time.Now().UTC()
+		h.reason = "load_capped"
+		if h.failed {
+			h.reason = "load_failed"
+		}
+	} else {
+		h.reason = "available_history"
+	}
+	// Eviction may have removed evidence while background chunks loaded.
+	// Never move a previously invalidated edge backward during completion.
+	if h.evictedFrom.After(floor) {
+		floor = h.evictedFrom
+		h.reason = "memory_eviction"
+	}
+	h.from, h.known = floor.UTC(), true
+}
+
+func (s *PacketStore) nodeAnalyticsCoverageLocked(days int, now time.Time) NodeAnalyticsCoverage {
+	now = now.UTC()
+	requested := now.Add(-time.Duration(days) * 24 * time.Hour)
+	h := s.analyticsHistory
+	from := requested
+	if s.retentionHours > 0 {
+		if floor := now.Add(-time.Duration(s.retentionHours * float64(time.Hour))); floor.After(from) {
+			from = floor
+		}
+	}
+	reason := h.reason
+	if !h.known {
+		from = now
+		reason = "unknown_history"
+	} else if h.from.After(from) {
+		from = h.from
+	}
+	if from.After(now) {
+		from = now
+	}
+	loading := !h.failed && (h.loading || !s.loaded)
+	if loading {
+		reason = "loading"
+	}
+	if h.failed {
+		reason = "load_failed"
+	}
+	partial := loading || h.failed || !h.known || from.After(requested)
+	if !partial {
+		reason = "complete"
+	} else if reason == "" || reason == "available_history" {
+		reason = "available_history"
+	}
+	return NodeAnalyticsCoverage{From: from.Format(time.RFC3339Nano), To: now.Format(time.RFC3339Nano), Seconds: now.Sub(from).Seconds(), Partial: partial, Reason: reason, Loading: loading, LateEvictedObservations: h.lateEvicted}
+}
+
+// invalidateNodeAnalyticsEvictionLocked advances the coverage edge to the
+// eviction boundary: the evicted transmission's FirstSeen (+1 ns to exclude
+// ties). Analytics membership is by FirstSeen, so nothing at or before that
+// instant can still be complete. Late observations of an evicted, re-heard
+// transmission (up to ~2 days after FirstSeen in production) are a known,
+// bounded gap that is only counted, never allowed to move the edge: routine
+// minute-by-minute eviction would otherwise make coverage jump forward.
+// An unparseable FirstSeen cannot bound anything and stays conservative.
+func (s *PacketStore) invalidateNodeAnalyticsEvictionLocked(removed []*StoreTx) {
+	h := &s.analyticsHistory
+	for _, tx := range removed {
+		edge, ok := parseNodeAnalyticsTime(tx.FirstSeen)
+		if !ok {
+			h.from = time.Now().UTC()
+		} else {
+			edge = edge.Add(time.Nanosecond)
+			if edge.After(h.from) {
+				h.from = edge
+			}
+			for _, obs := range tx.Observations {
+				if at, ok := parseNodeAnalyticsTime(obs.Timestamp); ok && !at.Before(edge) {
+					h.lateEvicted++
+				}
+			}
+		}
+		if h.from.After(h.evictedFrom) {
+			h.evictedFrom = h.from
+		}
+	}
+	if len(removed) > 0 {
+		h.reason = "memory_eviction"
+	}
+}
+
+// An initially empty database gains a conservative edge at its first live
+// observation, without inventing any history before the receiver evidence.
+func (s *PacketStore) observeNodeAnalyticsLiveLocked(timestamp string) {
+	h := &s.analyticsHistory
+	// A live row may arrive between an empty hot load and background
+	// completion. Keep its evidence even though the interval cannot be
+	// certified until the loader finishes. Steady-state known history pays
+	// no extra parse/reduction on this hot path.
+	if h.loading || !h.known {
+		h.observedFrom = earlierNodeAnalyticsEvidence(h.observedFrom, timestamp)
+	}
+	if h.known || h.loading || !s.loaded {
+		return
+	}
+	if at, ok := parseNodeAnalyticsTime(timestamp); ok {
+		if h.from.After(at) {
+			at = h.from
+		}
+		if h.evictedFrom.After(at) {
+			at = h.evictedFrom
+		}
+		h.from, h.known, h.reason = at, true, "available_history"
+	}
+}

cmd/server/node_analytics_coverage_bench_test.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// This benchmark intentionally uses only the pre-existing API so the same
+// source can run unchanged on the parent revision for a before/after result.
+func BenchmarkNodeAnalyticsCoverage485Endpoints(b *testing.B) {
+	for _, n := range []int{1000, 30000} {
+		for _, full := range []bool{false, true} {
+			b.Run(fmt.Sprintf("packets=%d/full=%t", n, full), func(b *testing.B) {
+				db := setupTestDB(b)
+				defer db.Close()
+				now := time.Now().UTC().Truncate(time.Second)
+				pk := "c048500000000000000000000000000000000000000000000000000000000000"
+				if _, err := db.conn.Exec(`INSERT INTO nodes (public_key,name,role) VALUES (?, 'Coverage bench','repeater')`, pk); err != nil {
+					b.Fatal(err)
+				}
+				tx, err := db.conn.Begin()
+				if err != nil {
+					b.Fatal(err)
+				}
+				stmt, err := tx.Prepare(`INSERT INTO transmissions (id,raw_hex,hash,first_seen,payload_type,decoded_json) VALUES (?,'1000',?,?,4,?)`)
+				if err != nil {
+					b.Fatal(err)
+				}
+				obs, err := tx.Prepare(`INSERT INTO observations (transmission_id,path_json,timestamp) VALUES (?,'[]',?)`)
+				if err != nil {
+					b.Fatal(err)
+				}
+				for i := 1; i <= n; i++ {
+					at := now.Add(-24*time.Hour + time.Duration(i)*time.Second)
+					if _, err := stmt.Exec(i, fmt.Sprintf("coverage-bench-%d", i), at.Format(time.RFC3339), `{"pubKey":"`+pk+`"}`); err != nil {
+						b.Fatal(err)
+					}
+					if _, err := obs.Exec(i, at.Unix()); err != nil {
+						b.Fatal(err)
+					}
+				}
+				stmt.Close()
+				obs.Close()
+				if err := tx.Commit(); err != nil {
+					b.Fatal(err)
+				}
+				s := NewPacketStore(db, nil)
+				if err := s.Load(); err != nil {
+					b.Fatal(err)
+				}
+				if !s.WaitIndexesReady(5 * time.Second) {
+					b.Fatal("indexes not ready")
+				}
+				b.ReportAllocs()
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					if full {
+						if _, err := s.getNodeAnalyticsAt(pk, 30, now); err != nil {
+							b.Fatal(err)
+						}
+					} else {
+						if _, err := s.getNodeAnalyticsSummaryAt(pk, 30, now); err != nil {
+							b.Fatal(err)
+						}
+					}
+				}
+			})
+		}
+	}
+}
+
+// F2: per-timestamp cost of the canonical-UTC fast path vs the time.Parse
+// fallback it replaces on the request path.
+func BenchmarkParseNodeAnalyticsTime485(b *testing.B) {
+	const canonical = "2026-10-16T12:34:56Z"
+	b.Run("fast_canonical", func(b *testing.B) {
+		b.ReportAllocs()
+		for i := 0; i < b.N; i++ {
+			parseNodeAnalyticsTime(canonical)
+		}
+	})
+	b.Run("fallback_time_Parse", func(b *testing.B) {
+		b.ReportAllocs()
+		for i := 0; i < b.N; i++ {
+			parseNodeAnalyticsTime("2026-10-16T12:34:56.5Z")
+		}
+	})
+}