diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index dd626716a..178e8c1a0 100644 --- a/cmd/server/chunked_load.go +++ b/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++ diff --git a/cmd/server/node_analytics_characterization_test.go b/cmd/server/node_analytics_characterization_test.go index 8a96a2f4c..1a73c1c44 100644 --- a/cmd/server/node_analytics_characterization_test.go +++ b/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 { diff --git a/cmd/server/node_analytics_coverage.go b/cmd/server/node_analytics_coverage.go new file mode 100644 index 000000000..b13edb3bb --- /dev/null +++ b/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" + } +} diff --git a/cmd/server/node_analytics_coverage_bench_test.go b/cmd/server/node_analytics_coverage_bench_test.go new file mode 100644 index 000000000..0d1541115 --- /dev/null +++ b/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") + } + }) +} diff --git a/cmd/server/node_analytics_coverage_errors_test.go b/cmd/server/node_analytics_coverage_errors_test.go new file mode 100644 index 000000000..5a3ad7d75 --- /dev/null +++ b/cmd/server/node_analytics_coverage_errors_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + sqlite "modernc.org/sqlite" +) + +// Wrap the real SQLite driver to force count, row-iteration and conversion +// failures independently, without production hooks or timing assumptions. +var coverage485RegisterFailureDriver sync.Once + +type coverage485FailureDriver struct{} +type coverage485FailureConn struct { + driver.Conn + failure string +} +type coverage485FailureRows struct { + driver.Rows + failure string + n int +} + +func (coverage485FailureDriver) Open(dsn string) (driver.Conn, error) { + parts := strings.SplitN(dsn, "|", 2) + conn, err := (&sqlite.Driver{}).Open(parts[0]) + if err != nil { + return nil, err + } + return coverage485FailureConn{Conn: conn, failure: parts[1]}, nil +} + +func (c coverage485FailureConn) QueryContext(ctx context.Context, q string, args []driver.NamedValue) (driver.Rows, error) { + if c.failure == "count" && strings.HasPrefix(q, "SELECT COUNT(*) FROM transmissions") { + return nil, errors.New("injected eligible-count failure") + } + rows, err := c.Conn.(driver.QueryerContext).QueryContext(ctx, q, args) + if err != nil { + return nil, err + } + if strings.HasPrefix(q, "SELECT t.id, t.raw_hex") { + return &coverage485FailureRows{Rows: rows, failure: c.failure}, nil + } + return rows, nil +} + +func (r *coverage485FailureRows) Next(values []driver.Value) error { + if r.failure == "cursor" && r.n == 1 { + return errors.New("injected cursor I/O failure") + } + err := r.Rows.Next(values) + if err == nil && r.failure == "scan" && r.n == 0 { + values[0] = "not an integer transmission id" + } + r.n++ + return err +} + +func TestNodeAnalyticsCoverage485_LoadErrorsNeverCertifyFullHistory(t *testing.T) { + coverage485RegisterFailureDriver.Do(func() { sql.Register("coverage485-failure", coverage485FailureDriver{}) }) + for _, mode := range []string{"legacy", "chunked"} { + for _, failure := range []string{"count", "cursor", "scan"} { + t.Run(mode+"/"+failure, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "failure.db") + createTestDBAt(t, path, 2) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + original := db.conn + t.Cleanup(func() { original.Close() }) + db.conn, err = sql.Open("coverage485-failure", path+"|"+failure) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + s := NewPacketStore(db, nil) + if mode == "legacy" { + err = s.Load() + } else { + err = s.LoadChunked(10000) + } + if err == nil && !s.WaitIndexesReady(5*time.Second) { + t.Fatal("indexes not ready") + } + s.mu.RLock() + c := s.nodeAnalyticsCoverageLocked(1, time.Now()) + failed := s.analyticsHistory.failed + s.mu.RUnlock() + if !c.Partial || (err == nil && !failed) { + t.Fatalf("%s error silently certified: err=%v failed=%v coverage=%+v", failure, err, failed, c) + } + if err == nil && c.Seconds > 1 { + t.Fatalf("failed historical subset used as supported history: %+v", c) + } + }) + } + } +} diff --git a/cmd/server/node_analytics_coverage_test.go b/cmd/server/node_analytics_coverage_test.go new file mode 100644 index 000000000..0a1bfb24b --- /dev/null +++ b/cmd/server/node_analytics_coverage_test.go @@ -0,0 +1,399 @@ +package main + +import ( + "database/sql" + "fmt" + "math" + "path/filepath" + "reflect" + "testing" + "time" +) + +const coverage485PK = "c048500000000000000000000000000000000000000000000000000000000000" + +func TestNodeAnalyticsCoverage485_RequestedMonthDoesNotDiluteRetainedTwoDays(t *testing.T) { + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + if _, err := db.conn.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, 'Coverage test', 'repeater')`, coverage485PK); err != nil { + t.Fatal(err) + } + store := NewPacketStore(db, &PacketStoreConfig{RetentionHours: 48}) + now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) + store.loaded = true + store.oldestLoaded = now.Add(-48 * time.Hour).Format(time.RFC3339) + store.analyticsHistory = nodeAnalyticsHistory{known: true, from: now.Add(-48 * time.Hour)} + for h := 0; h < 48; h++ { + tx := &StoreTx{ID: h + 1, Hash: fmt.Sprintf("coverage-%d", h), FirstSeen: now.Add(-48*time.Hour + time.Duration(h)*time.Hour).Format(time.RFC3339)} + store.byNode[coverage485PK] = append(store.byNode[coverage485PK], tx) + store.packets = append(store.packets, tx) + } + full, err := store.getNodeAnalyticsAt(coverage485PK, 30, now) + if err != nil { + t.Fatal(err) + } + if full.ComputedStats.AvgPacketsPerDay != 24 { + t.Fatalf("packets/day=%v, want 24 over the supported two days, not the requested month", full.ComputedStats.AvgPacketsPerDay) + } + if full.ComputedStats.AvailabilityPct != 100 { + t.Fatalf("availability=%v, want 100 for all 48 supported hour bins", full.ComputedStats.AvailabilityPct) + } +} + +func coverage485Store(t *testing.T, now time.Time, floor time.Time) *PacketStore { + t.Helper() + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + if _, err := db.conn.Exec(`INSERT INTO nodes (public_key, name, role) VALUES (?, 'Coverage test', 'repeater')`, coverage485PK); err != nil { + t.Fatal(err) + } + s := NewPacketStore(db, nil) + s.loaded = true + s.analyticsHistory = nodeAnalyticsHistory{known: !floor.IsZero(), from: floor} + return s +} + +func coverage485Packet(s *PacketStore, first string, observations ...string) { + id := len(s.packets) + 1 + tx := &StoreTx{ID: id, Hash: fmt.Sprintf("cov-%d", id), FirstSeen: first} + for _, at := range observations { + tx.Observations = append(tx.Observations, &StoreObs{Timestamp: at}) + } + s.packets = append(s.packets, tx) + s.byNode[coverage485PK] = append(s.byNode[coverage485PK], tx) +} + +func coverage485Pair(t *testing.T, s *PacketStore, days int, now time.Time) *NodeAnalyticsResponse { + t.Helper() + f, err := s.getNodeAnalyticsAt(coverage485PK, days, now) + if err != nil { + t.Fatal(err) + } + l, err := s.getNodeAnalyticsSummaryAt(coverage485PK, days, now) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(f.ComputedStats, l.ComputedStats) || f.Coverage != l.Coverage || f.TimeRange != l.TimeRange { + t.Fatalf("full/summary differ: %+v / %+v", f, l) + } + if math.IsNaN(f.ComputedStats.AvailabilityPct) || math.IsInf(f.ComputedStats.AvgPacketsPerDay, 0) { + t.Fatal("non-finite rates") + } + return f +} + +func TestNodeAnalyticsCoverage485_BoundariesUTCAndPartialHour(t *testing.T) { + now := time.Date(2026, 10, 16, 12, 15, 0, 0, time.FixedZone("east", 2*3600)) // 10:15 UTC + floor := now.UTC().Add(-90 * time.Minute) // 08:45 + s := coverage485Store(t, now, floor) + for _, raw := range []string{"2026-10-16T08:44:59Z", "2026-10-16T10:45:00+02:00", "2026-10-16T09:59:00Z", "2026-10-16T10:15:00Z", "2026-10-16T10:15:01Z", "malformed"} { + coverage485Packet(s, raw) + } + f := coverage485Pair(t, s, 30, now) + if f.ComputedStats.TotalPackets != 3 || f.ComputedStats.AvgPacketsPerDay != 48 || f.ComputedStats.AvailabilityPct != 100 { + t.Fatalf("stats=%+v", f.ComputedStats) + } + if f.Coverage.From != "2026-10-16T08:45:00Z" || f.Coverage.Seconds != 5400 || !f.Coverage.Partial { + t.Fatalf("coverage=%+v", f.Coverage) + } + if f.TimeRange.To != "2026-10-16T10:15:00Z" { + t.Fatalf("non UTC range=%+v", f.TimeRange) + } + for _, b := range f.ActivityTimeline { + at, _ := time.Parse(time.RFC3339Nano, *b.Bucket) + if at.Before(floor) { + t.Fatalf("bucket precedes supported history: %v", b) + } + } +} + +func TestNodeAnalyticsCoverage485_CompleteAndInactiveNode(t *testing.T) { + now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) + s := coverage485Store(t, now, now.Add(-48*time.Hour)) + f := coverage485Pair(t, s, 1, now) + if f.Coverage.Partial || f.Coverage.Seconds != 86400 || f.ComputedStats.TotalPackets != 0 || f.ComputedStats.AvailabilityPct != 0 { + t.Fatalf("complete empty node=%+v", f) + } +} + +func TestNodeAnalyticsCoverage485_EmptyLoadingCappedFailedAndZeroDuration(t *testing.T) { + now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) + for _, tt := range []struct { + name string + history nodeAnalyticsHistory + loaded bool + }{ + {"empty", nodeAnalyticsHistory{}, true}, + {"loading", nodeAnalyticsHistory{loading: true}, false}, + {"capped", nodeAnalyticsHistory{known: true, from: now, cap: true, reason: "load_capped"}, true}, + {"failed", nodeAnalyticsHistory{known: true, from: now, failed: true, reason: "load_failed"}, true}, + {"future", nodeAnalyticsHistory{known: true, from: now.Add(time.Second)}, true}, + } { + t.Run(tt.name, func(t *testing.T) { + s := coverage485Store(t, now, time.Time{}) + s.analyticsHistory = tt.history + s.loaded = tt.loaded + coverage485Packet(s, now.Add(-time.Hour).Format(time.RFC3339)) + f := coverage485Pair(t, s, 7, now) + if !f.Coverage.Partial || f.Coverage.Seconds != 0 || f.ComputedStats.TotalPackets != 0 || f.ComputedStats.AvailabilityPct != 0 || f.ComputedStats.AvgPacketsPerDay != 0 { + t.Fatalf("unknown/zero history=%+v", f) + } + }) + } +} + +func TestNodeAnalyticsCoverage485_ReheardOldDoesNotProveOldHistory(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + s := coverage485Store(t, now, time.Time{}) + coverage485Packet(s, now.Add(-365*24*time.Hour).Format(time.RFC3339), now.Add(-time.Hour).Format(time.RFC3339)) + coverage485Packet(s, now.Add(-30*time.Minute).Format(time.RFC3339), now.Add(-30*time.Minute).Format(time.RFC3339)) + s.beginNodeAnalyticsLoadLocked(now.Add(-30 * 24 * time.Hour).Format(time.RFC3339)) + s.analyticsHistory.observedFrom = now.Add(-time.Hour) + s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, false, false) + f := coverage485Pair(t, s, 30, now) + if f.Coverage.Seconds != 3600 || f.ComputedStats.TotalPackets != 1 || f.ComputedStats.AvgPacketsPerDay != 24 { + t.Fatalf("reheard=%+v", f) + } +} + +func TestNodeAnalyticsCoverage485_LoadLifecycleShortDBAndEviction(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + s := coverage485Store(t, now, time.Time{}) + coverage485Packet(s, now.Add(-40*24*time.Hour).Format(time.RFC3339), now.Add(-24*time.Hour).Format(time.RFC3339), now.Add(-time.Hour).Format(time.RFC3339)) + coverage485Packet(s, now.Add(-time.Hour).Format(time.RFC3339), now.Add(-time.Hour).Format(time.RFC3339)) + s.beginNodeAnalyticsLoadLocked(now.Add(-30 * 24 * time.Hour).Format(time.RFC3339)) + s.analyticsHistory.observedFrom = now.Add(-24 * time.Hour) + s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, false, true) + f := coverage485Pair(t, s, 30, now) + if !f.Coverage.Loading || f.Coverage.Seconds != 86400 { + t.Fatalf("short DB=%+v", f.Coverage) + } + s.analyticsHistory.loading = false + s.invalidateNodeAnalyticsEvictionLocked(s.packets[:1]) + // The evicted tx was first seen 40 days ago; its late observations must + // not move the edge (F1). Coverage stays at the 24 h loaded edge. + f = coverage485Pair(t, s, 30, now) + if f.Coverage.Reason != "memory_eviction" || f.Coverage.Seconds != 86400 || f.ComputedStats.TotalPackets != 1 { + t.Fatalf("reheard eviction=%+v", f) + } + // A later background completion cannot undo the invalidated edge. + s.finishNodeAnalyticsLoadLocked(now.Add(-30*24*time.Hour), false, false) + if s.analyticsHistory.from.Before(now.Add(-24 * time.Hour)) { + t.Fatal("background completion restored evicted coverage") + } +} + +func TestNodeAnalyticsCoverage485_EvictingReheardOldTxDoesNotJumpFrom(t *testing.T) { + now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) + floor := now.Add(-72 * time.Hour) + s := coverage485Store(t, now, floor) + // Old tx first seen at the edge, re-heard ~2 days later (production max). + coverage485Packet(s, floor.Add(time.Minute).Format(time.RFC3339), floor.Add(time.Minute).Format(time.RFC3339), now.Add(-time.Hour).Format(time.RFC3339), now.Add(-30*time.Minute).Format(time.RFC3339)) + coverage485Packet(s, now.Add(-10*time.Hour).Format(time.RFC3339), now.Add(-10*time.Hour).Format(time.RFC3339)) + s.invalidateNodeAnalyticsEvictionLocked(s.packets[:1]) + want := floor.Add(time.Minute).Add(time.Nanosecond) + if got := s.analyticsHistory.from; !got.Equal(want) { + t.Fatalf("from=%v want eviction boundary %v (late observation must not move it)", got, want) + } + cov := s.nodeAnalyticsCoverageLocked(30, now) + if cov.LateEvictedObservations != 2 { + t.Fatalf("late evicted observations=%d want 2", cov.LateEvictedObservations) + } + if want := now.Sub(floor.Add(time.Minute)).Seconds(); cov.Seconds < want-1 || cov.Seconds > want { + t.Fatalf("seconds=%v want ~%v", cov.Seconds, want) + } +} + +func TestNodeAnalyticsCoverage485_InitiallyEmptyGainsLiveEvidence(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + s := coverage485Store(t, now, time.Time{}) + s.observeNodeAnalyticsLiveLocked("invalid") + if s.analyticsHistory.known { + t.Fatal("invalid timestamp proved history") + } + s.observeNodeAnalyticsLiveLocked(now.Add(-time.Minute).Format(time.RFC3339)) + if !s.analyticsHistory.known || s.nodeAnalyticsCoverageLocked(1, now).Seconds != 60 { + t.Fatalf("first live evidence=%+v", s.analyticsHistory) + } +} + +func TestNodeAnalyticsCoverage485_LiveEvidenceSurvivesEmptyBackgroundHandoff(t *testing.T) { + now := time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) + s := coverage485Store(t, now, time.Time{}) + s.beginNodeAnalyticsLoadLocked(now.Add(-48 * time.Hour).Format(time.RFC3339)) + s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, false, true) + at := now.Add(-time.Minute) + coverage485Packet(s, at.Format(time.RFC3339), at.Format(time.RFC3339)) + s.observeNodeAnalyticsLiveLocked(at.Format(time.RFC3339)) + if !s.nodeAnalyticsCoverageLocked(1, now).Loading { + t.Fatal("unfinished background load claimed completion") + } + s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, false, false) + f := coverage485Pair(t, s, 1, now) + if f.Coverage.Loading || f.Coverage.Seconds != 60 || f.ComputedStats.TotalPackets != 1 || !s.analyticsHistory.known { + t.Fatalf("accepted live observation was lost at handoff: %+v", f) + } +} + +func BenchmarkNodeAnalyticsCoverage485Snapshot(b *testing.B) { + now := time.Now().UTC() + s := &PacketStore{loaded: true, analyticsHistory: nodeAnalyticsHistory{known: true, from: now.Add(-48 * time.Hour)}} + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _ = s.nodeAnalyticsCoverageLocked(30, now) + } +} + +func TestNodeAnalyticsCoverage485_RealStartupLoadersUseObservationEvidence(t *testing.T) { + for _, mode := range []string{"legacy", "chunked", "background"} { + t.Run(mode, func(t *testing.T) { + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + seedTestData(t, db) + if _, err := db.conn.Exec(`ALTER TABLE transmissions ADD COLUMN last_seen INTEGER; UPDATE transmissions SET last_seen=COALESCE((SELECT MAX(timestamp) FROM observations o WHERE o.transmission_id=transmissions.id),0); UPDATE transmissions SET first_seen='2020-01-01T00:00:00Z' WHERE id=1`); err != nil { + t.Fatal(err) + } + db.hasLastSeenFlag.forceTrue() + cfg := &PacketStoreConfig{RetentionHours: 30 * 24} + if mode == "background" { + cfg.HotStartupHours = 2 + } + s := NewPacketStore(db, cfg) + if mode == "legacy" { + if err := s.Load(); err != nil { + t.Fatal(err) + } + } else { + s.OnChunkLoaded(func(_, _ int) { + s.mu.RLock() + c := s.nodeAnalyticsCoverageLocked(30, time.Now()) + s.mu.RUnlock() + if !c.Partial || !c.Loading { + t.Error("in-flight loader claimed complete coverage") + } + }) + if err := s.RunStartupLoad(1); err != nil { + t.Fatal(err) + } + } + if !s.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + now := time.Now().UTC() + full, err := s.getNodeAnalyticsAt(nodeAnalyticsSummaryTestPK, 30, now) + if err != nil { + t.Fatal(err) + } + if !full.Coverage.Partial || full.Coverage.Seconds < 23*3600 || full.Coverage.Seconds > 25*3600 || full.ComputedStats.TotalPackets != 1 { + t.Fatalf("%s coverage=%+v stats=%+v", mode, full.Coverage, full.ComputedStats) + } + if full.Coverage.Loading { + t.Fatal("finished loader still reports loading") + } + }) + } +} + +func TestNodeAnalyticsCoverage485_BudgetOnlyCapsWhenRowsOmitted(t *testing.T) { + for _, mode := range []string{"legacy", "chunked"} { + for _, n := range []int{1000, 1001} { + t.Run(fmt.Sprintf("%s/%d", mode, n), func(t *testing.T) { + path := filepath.Join(t.TempDir(), "coverage.db") + createTestDBAt(t, path, n) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + s := NewPacketStore(db, &PacketStoreConfig{MaxMemoryMB: 1}) + if mode == "legacy" { + err = s.Load() + } else { + err = s.LoadChunked(1000) + } + if err != nil { + t.Fatal(err) + } + if !s.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + wantCap := n > 1000 + if s.analyticsHistory.cap != wantCap { + t.Fatalf("loaded=%d/%d cap=%v want%v", len(s.packets), n, s.analyticsHistory.cap, wantCap) + } + if !wantCap && s.nodeAnalyticsCoverageLocked(365, time.Now()).Seconds < 24*3600 { + t.Fatal("complete 1000-row history was discarded") + } + }) + } + } +} + +func TestNodeAnalyticsCoverage485_ExcludedRetentionRowsAreNotMemoryCap(t *testing.T) { + for _, mode := range []string{"legacy", "chunked"} { + t.Run(mode, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "retained.db") + createTestDBAt(t, path, 1001) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.Close() }) + recent := time.Now().UTC().Add(-time.Hour).Truncate(time.Second).Format(time.RFC3339) + writer, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + _, err = writer.Exec(`UPDATE transmissions SET first_seen=?; UPDATE observations SET timestamp=?; UPDATE transmissions SET first_seen='2020-01-01T00:00:00Z' WHERE id=1001; UPDATE observations SET timestamp='2020-01-01T00:00:00Z' WHERE transmission_id=1001`, recent, recent) + writer.Close() + if err != nil { + t.Fatal(err) + } + s := NewPacketStore(db, &PacketStoreConfig{MaxMemoryMB: 1, RetentionHours: 24}) + if mode == "legacy" { + err = s.Load() + } else { + err = s.LoadChunked(1000) + } + if err != nil { + t.Fatal(err) + } + if !s.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + if len(s.packets) != 1000 || s.analyticsHistory.cap { + t.Fatalf("retained population incorrectly capped: loaded=%d history=%+v", len(s.packets), s.analyticsHistory) + } + if s.nodeAnalyticsCoverageLocked(30, time.Now()).Seconds < 3599 { + t.Fatal("eligible historical interval discarded") + } + }) + } +} + +// F2: the canonical-UTC fast path must be indistinguishable from the +// time.Parse fallback for every input shape, valid or not. +func TestParseNodeAnalyticsTime485_FastPathMatchesFallback(t *testing.T) { + slow := func(raw string) (time.Time, bool) { + tm, err := time.Parse(time.RFC3339Nano, raw) + if err != nil { + tm, err = time.ParseInLocation("2006-01-02 15:04:05", raw, time.UTC) + } + return tm.UTC(), err == nil && tm.Unix() > 0 + } + for _, raw := range []string{ + "2026-10-16T12:00:00Z", "2024-02-29T23:59:59Z", "2025-02-29T00:00:00Z", + "2026-13-01T00:00:00Z", "2026-00-10T00:00:00Z", "2026-04-31T00:00:00Z", + "2026-10-16T24:00:00Z", "2026-10-16T12:60:00Z", "2026-10-16T12:00:60Z", + "2026-10-16T12:00:00z", "2026-10-16T12:00:0aZ", "2026-10-16 12:00:00", + "2026-10-16T12:00:00+02:00", "2026-10-16T12:00:00.123456789Z", + "1970-01-01T00:00:00Z", "1969-12-31T23:59:59Z", "0000-01-01T00:00:00Z", + "2026-10-16T12:00:00Z ", "", "x", "2026-10-16T12:00:00ZZ", "2026+10-16T12:00:00Z", + } { + gt, gok := parseNodeAnalyticsTime(raw) + wt, wok := slow(raw) + if gok != wok || (gok && !gt.Equal(wt)) { + t.Errorf("%q: got (%v,%v) want (%v,%v)", raw, gt, gok, wt, wok) + } + } +} diff --git a/cmd/server/node_analytics_summary.go b/cmd/server/node_analytics_summary.go index a897e5625..cf47c1c91 100644 --- a/cmd/server/node_analytics_summary.go +++ b/cmd/server/node_analytics_summary.go @@ -89,8 +89,8 @@ func (a *nodeAnalyticsAccumulator) accumulate(p *StoreTx) { a.totalPackets++ // Activity timeline bucket. - if len(p.FirstSeen) >= 13 { - a.timelineBuckets[p.FirstSeen[:13]+":00:00Z"]++ + if at, ok := parseNodeAnalyticsTime(p.FirstSeen); ok { + a.timelineBuckets[at.Truncate(time.Hour).Format(time.RFC3339)]++ } // SNR — folded into Welford's online mean/M2 in constant memory (mean/ @@ -248,7 +248,7 @@ func nodeAnalyticsPeerCandidates(decoded map[string]interface{}, pubkey string) // finalizeComputedStats derives all 11 ComputedNodeStats fields — identical // whether or not withDisplayArrays was set, since every field is sourced // from the "always accumulated" state above. -func (a *nodeAnalyticsAccumulator) finalizeComputedStats(days int) ComputedNodeStats { +func (a *nodeAnalyticsAccumulator) finalizeComputedStats(days int, coverage ...NodeAnalyticsCoverage) ComputedNodeStats { bucketKeys := make([]string, 0, len(a.timelineBuckets)) for k := range a.timelineBuckets { bucketKeys = append(bucketKeys, k) @@ -257,17 +257,41 @@ func (a *nodeAnalyticsAccumulator) finalizeComputedStats(days int) ComputedNodeS distinctHours := len(bucketKeys) totalHours := float64(days) * 24 + activeHours := float64(distinctHours) + if len(coverage) > 0 { + c := coverage[0] + totalHours = c.Seconds / 3600 + activeHours = 0 + from, _ := time.Parse(time.RFC3339Nano, c.From) + to, _ := time.Parse(time.RFC3339Nano, c.To) + for _, key := range bucketKeys { + start, err := time.Parse(time.RFC3339, key) + if err != nil { + continue + } + end := start.Add(time.Hour) + if start.Before(from) { + start = from + } + if end.After(to) { + end = to + } + if end.After(start) { + activeHours += end.Sub(start).Hours() + } + } + } availabilityPct := 0.0 if totalHours > 0 { - availabilityPct = round(float64(distinctHours)*100.0/totalHours, 1) + availabilityPct = round(activeHours*100.0/totalHours, 1) if availabilityPct > 100 { availabilityPct = 100 } } var avgPacketsPerDay float64 - if days > 0 { - avgPacketsPerDay = round(float64(a.totalPackets)/float64(days), 1) + if totalHours > 0 { + avgPacketsPerDay = round(float64(a.totalPackets)*24/totalHours, 1) } var longestSilenceMs int @@ -342,7 +366,7 @@ func (a *nodeAnalyticsAccumulator) finalizeComputedStats(days int) ComputedNodeS // finalizeDisplayArrays builds the full endpoint's seven heavy client // arrays from the accumulator. Only meaningful when withDisplayArrays was // true — the summary endpoint never calls this. -func (a *nodeAnalyticsAccumulator) finalizeDisplayArrays() ( +func (a *nodeAnalyticsAccumulator) finalizeDisplayArrays(coverage ...NodeAnalyticsCoverage) ( activityTimeline []TimeBucket, snrTrend []SnrTrendEntry, packetTypeBreakdown []PayloadTypeCount, @@ -359,6 +383,13 @@ func (a *nodeAnalyticsAccumulator) finalizeDisplayArrays() ( activityTimeline = make([]TimeBucket, 0, len(bucketKeys)) for _, k := range bucketKeys { b := k + if len(coverage) > 0 { + at, _ := time.Parse(time.RFC3339, k) + from, _ := time.Parse(time.RFC3339Nano, coverage[0].From) + if at.Before(from) { + b = coverage[0].From + } + } activityTimeline = append(activityTimeline, TimeBucket{Bucket: &b, Count: a.timelineBuckets[k]}) } @@ -440,11 +471,14 @@ func (a *nodeAnalyticsAccumulator) finalizeDisplayArrays() ( // GRP_TXT packet whose message text contains a node's pubkey is not a // packet *for* that node (issue673). Caller must hold s.mu (at least // RLock). -func (s *PacketStore) filterNodePacketsLocked(pubkey, fromISO string) []*StoreTx { +func (s *PacketStore) filterNodePacketsLocked(pubkey string, from, to time.Time) []*StoreTx { indexed := s.byNode[pubkey] var packets []*StoreTx + if !to.After(from) { + return packets + } for _, p := range indexed { - if p.FirstSeen > fromISO { + if at, ok := parseNodeAnalyticsTime(p.FirstSeen); ok && !at.Before(from) && !at.After(to) { packets = append(packets, p) } } @@ -477,13 +511,16 @@ func (s *PacketStore) getNodeAnalyticsSummaryAt(pubkey string, days int, now tim return nil, err } + now = now.UTC() fromISO := now.Add(-time.Duration(days) * 24 * time.Hour).Format(time.RFC3339) toISO := now.Format(time.RFC3339) s.mu.RLock() defer s.mu.RUnlock() - packets := s.filterNodePacketsLocked(pubkey, fromISO) + coverage := s.nodeAnalyticsCoverageLocked(days, now) + from, _ := time.Parse(time.RFC3339Nano, coverage.From) + packets := s.filterNodePacketsLocked(pubkey, from, now) acc := newNodeAnalyticsAccumulator(pubkey, false) for _, p := range packets { @@ -492,6 +529,7 @@ func (s *PacketStore) getNodeAnalyticsSummaryAt(pubkey string, days int, now tim return &NodeAnalyticsSummaryResponse{ TimeRange: TimeRangeResp{From: fromISO, To: toISO, Days: days}, - ComputedStats: acc.finalizeComputedStats(days), + Coverage: coverage, + ComputedStats: acc.finalizeComputedStats(days, coverage), }, nil } diff --git a/cmd/server/node_analytics_summary_test.go b/cmd/server/node_analytics_summary_test.go index a83783962..12521dcb0 100644 --- a/cmd/server/node_analytics_summary_test.go +++ b/cmd/server/node_analytics_summary_test.go @@ -12,7 +12,7 @@ import ( const nodeAnalyticsSummaryTestPK = "aabbccdd11223344" // seeded by seedTestData // TestNodeAnalyticsSummary_OnlyTimeRangeAndComputedStats asserts the wire -// JSON has exactly two top-level keys — no node, no clockSkew, none of the +// JSON has exactly three top-level keys — no node, no clockSkew, none of the // heavy display arrays the full /analytics endpoint carries. func TestNodeAnalyticsSummary_OnlyTimeRangeAndComputedStats(t *testing.T) { _, router := setupTestServer(t) @@ -28,12 +28,12 @@ func TestNodeAnalyticsSummary_OnlyTimeRangeAndComputedStats(t *testing.T) { if err := json.Unmarshal(w.Body.Bytes(), &raw); err != nil { t.Fatalf("invalid JSON: %v", err) } - if len(raw) != 2 { + if len(raw) != 3 { gotKeys := make([]string, 0, len(raw)) for k := range raw { gotKeys = append(gotKeys, k) } - t.Fatalf("expected exactly 2 top-level keys, got %d: %v", len(raw), gotKeys) + t.Fatalf("expected exactly 3 top-level keys, got %d: %v", len(raw), gotKeys) } if _, ok := raw["timeRange"]; !ok { t.Error("missing timeRange") @@ -41,6 +41,9 @@ func TestNodeAnalyticsSummary_OnlyTimeRangeAndComputedStats(t *testing.T) { if _, ok := raw["computedStats"]; !ok { t.Error("missing computedStats") } + if _, ok := raw["coverage"]; !ok { + t.Error("missing coverage") + } } // TestNodeAnalyticsSummary_MatchesFullComputedStats asserts full and diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 82a325f69..4fda022ac 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -121,8 +121,8 @@ func routeDescriptions() map[string]routeMeta { }}, "GET /api/nodes/{pubkey}/health": {Summary: "Get node health", Description: "Health of one node: stats, recent packets, the rolling 24 h activity and its Heard By lists. directObservers received the node's own transmission; relayObservers saw only other evidence (see HeardDirectObserver). Both lists describe observed routing evidence, not radio range.", Tag: "nodes", Response: schemaRef("NodeHealthResponse")}, "GET /api/nodes/{pubkey}/paths": {Summary: "Get node routing paths", Tag: "nodes"}, - "GET /api/nodes/{pubkey}/analytics": {Summary: "Get node analytics", Description: "Per-node packet counts, timing, and RF stats.", Tag: "nodes"}, - "GET /api/nodes/{pubkey}/analytics/summary": {Summary: "Get node analytics summary", Description: "Fase 5.2a: light sibling of /analytics — timeRange + computedStats only (availabilityPct, longestSilenceMs/Start, signalGrade, snrMean/snrStdDev, relayPct, totalPackets, uniqueObservers, uniquePeers, avgPacketsPerDay). No node object, no clockSkew, and none of the heavy display arrays (activityTimeline/snrTrend/packetTypeBreakdown/observerCoverage/hopDistribution/peerInteractions/uptimeHeatmap) — for callers that only need the summary numbers, not the full endpoint's per-packet detail. Shares its per-packet computation with /analytics, so computedStats is identical between the two for the same pubkey/days.", Tag: "nodes", + "GET /api/nodes/{pubkey}/analytics": {Summary: "Get node analytics", Description: "Per-node packet counts, timing, and RF stats. timeRange is the requested range; coverage describes the effective UTC retained-history interval, partial/loading status and reason (late_evicted_observations counts late observations of evicted transmissions, which never move coverage.from). Rates and packet arrays use that interval; zero supported duration yields zero rates. This is not radio uptime.", Tag: "nodes"}, + "GET /api/nodes/{pubkey}/analytics/summary": {Summary: "Get node analytics summary", Description: "Light sibling of /analytics: requested timeRange, effective UTC retained-history coverage, and computedStats. No node object, clockSkew, or heavy display arrays. Shares coverage, packet membership and computed statistics with the full endpoint for the same pubkey/days/instant. Partial/loading/unknown history is explicit; rates use coverage.seconds, not the requested days. Zero supported duration yields zero rates. This is available store data, not measured radio uptime.", Tag: "nodes", QueryParams: []paramMeta{ {Name: "days", Description: "Time window in days, 1-365. Default 7.", Type: "integer"}, }}, diff --git a/cmd/server/store.go b/cmd/server/store.go index 37ba45e29..15e9b82c8 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -197,23 +197,24 @@ func (tx *StoreTx) ParsedDecoded() map[string]interface{} { // All other locks are acquired independently (no nesting). // When adding new lock acquisitions, respect this ordering. type PacketStore struct { - mu sync.RWMutex - db *DB - packets []*StoreTx // sorted by first_seen ASC (oldest first; newest at tail) - byHash map[string]*StoreTx // hash → *StoreTx - byTxID map[int]*StoreTx // transmission_id → *StoreTx - byObsID map[int]*StoreObs // observation_id → *StoreObs - maxTxID int // highest transmission_id in store - maxObsID int // highest observation_id in store - byObserver map[string][]*StoreObs // observer_id → observations - byNode map[string][]*StoreTx // pubkey → transmissions - nodeHashes map[string]map[string]bool // pubkey → Set - byPathHop map[string][]*StoreTx // lowercase hop/pubkey → transmissions with that hop in path - byPayloadType map[int][]*StoreTx // payload_type → transmissions - loaded bool - totalObs int - insertCount int64 - queryCount int64 + mu sync.RWMutex + db *DB + packets []*StoreTx // sorted by first_seen ASC (oldest first; newest at tail) + byHash map[string]*StoreTx // hash → *StoreTx + byTxID map[int]*StoreTx // transmission_id → *StoreTx + byObsID map[int]*StoreObs // observation_id → *StoreObs + maxTxID int // highest transmission_id in store + maxObsID int // highest observation_id in store + byObserver map[string][]*StoreObs // observer_id → observations + byNode map[string][]*StoreTx // pubkey → transmissions + nodeHashes map[string]map[string]bool // pubkey → Set + byPathHop map[string][]*StoreTx // lowercase hop/pubkey → transmissions with that hop in path + byPayloadType map[int][]*StoreTx // payload_type → transmissions + loaded bool + analyticsHistory nodeAnalyticsHistory // guarded by mu; retained analytics coverage (#485) + totalObs int + insertCount int64 + queryCount int64 // Response caches (separate mutex to avoid contention with store RWMutex) cacheMu sync.RWMutex rfCache map[string]*cachedResult // region → cached RF result @@ -890,11 +891,9 @@ func (s *PacketStore) Load() error { t0 := time.Now() - // Count total transmissions for logging. - var totalInDB int - if err := s.db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&totalInDB); err != nil { - totalInDB = -1 // non-fatal - } + // Count the eligible window below, before applying the memory cap. + // A configured budget is not proof that any retained rows were omitted. + totalInDB := -1 // Calculate max packets to load based on memory budget. var maxPackets int64 @@ -957,6 +956,13 @@ func (s *PacketStore) Load() error { loadConditions = append(loadConditions, fmt.Sprintf("t.first_seen >= '%s'", hotCutoffStr)) } } + countSQL := "SELECT COUNT(*) FROM transmissions t" + if len(loadConditions) > 0 { + countSQL += " WHERE " + strings.Join(loadConditions, " AND ") + } + if err := s.db.conn.QueryRow(countSQL).Scan(&totalInDB); err != nil { + totalInDB = -1 + } if maxPackets > 0 { if s.db.hasLastSeen() { loadConditions = append(loadConditions, fmt.Sprintf( @@ -970,6 +976,7 @@ func (s *PacketStore) Load() error { if len(loadConditions) > 0 { filterClause = "\n\t\t\tWHERE " + strings.Join(loadConditions, "\n\t\t\t AND ") } + s.beginNodeAnalyticsLoadLocked(hotCutoffStr) if s.db.isV3() { loadSQL = `SELECT t.id, t.raw_hex, t.hash, t.first_seen, t.route_type, @@ -1046,6 +1053,7 @@ func (s *PacketStore) Load() error { } if err := rows.Scan(scanArgs...); err != nil { log.Printf("[store] scan error: %v", err) + s.analyticsHistory.failed = true continue } if s.loadScannedRowHook != nil { @@ -1118,6 +1126,7 @@ func (s *PacketStore) Load() error { 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++ @@ -1141,6 +1150,9 @@ func (s *PacketStore) Load() error { } } + if rows.Err() != nil { + s.analyticsHistory.failed = true + } // Post-load: pick best observation (longest path) for each transmission, // then re-index so relay hops from resolved_path land in byNode. // indexByNode was called earlier (on StoreTx creation) before observations @@ -1180,6 +1192,8 @@ func (s *PacketStore) Load() error { s.oldestLoaded = s.packets[0].FirstSeen } + s.analyticsHistory.failed = s.analyticsHistory.failed || totalInDB < 0 + s.finishNodeAnalyticsLoadLocked(s.analyticsHistory.loadCutoff, maxPackets > 0 && totalInDB > len(s.packets), s.hotStartupHours > 0) s.loaded = true elapsed := time.Since(t0) if maxPackets > 0 && totalInDB > len(s.packets) { @@ -1374,7 +1388,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { } if err := rows.Scan(scanArgs...); err != nil { log.Printf("[store] loadChunk scan error: %v", err) - continue + return fmt.Errorf("analytics history chunk scan: %w", err) } hashStr := nullStrVal(hash) @@ -1483,8 +1497,14 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { return err } + // Evidence is reduced outside s.mu; publish only the scalar with the + // final atomic chunk merge, not an extra full-store scan under a lock. + var localObservedFrom time.Time // Pick best observation for each local packet before merging. for _, tx := range localPackets { + for _, obs := range tx.Observations { + localObservedFrom = earlierNodeAnalyticsEvidence(localObservedFrom, obs.Timestamp) + } pickBestObservation(tx) localTrackedBytes += rechargeTx(tx) } @@ -1619,6 +1639,9 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // 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) + if !localObservedFrom.IsZero() && (s.analyticsHistory.observedFrom.IsZero() || localObservedFrom.Before(s.analyticsHistory.observedFrom)) { + s.analyticsHistory.observedFrom = localObservedFrom + } s.totalObs += localTotalObs s.trackedBytes += localTrackedBytes if localMaxTxID > s.maxTxID { @@ -1709,6 +1732,9 @@ func (s *PacketStore) loadBackgroundChunks() { } if s.retentionHours <= 0 { + s.mu.Lock() + s.analyticsHistory.loading = false + s.mu.Unlock() s.backgroundLoadDone.Store(true) return } @@ -1851,6 +1877,10 @@ func (s *PacketStore) loadBackgroundChunks() { } bgErr := s.backgroundLoadErr s.bgErrMu.Unlock() + s.mu.Lock() + s.analyticsHistory.failed = s.analyticsHistory.failed || chunkErrors > 0 || totalInDB < 0 || (totalInDB > 0 && loadedCount < totalInDB) + s.finishNodeAnalyticsLoadLocked(target, false, false) + s.mu.Unlock() // Only flip done=true when the store truly reflects the DB. Operators // (and the analytics warm-up gate, #1659) read this as "no more rows @@ -3442,6 +3472,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac tx.Observations = append(tx.Observations, obs) tx.mergeObservedPathHashSize(r.pathJSON) + s.observeNodeAnalyticsLiveLocked(obs.Timestamp) tx.obsKeys[dk] = true if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] { tx.observerSet[obs.ObserverID] = true @@ -3832,6 +3863,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] tx.Observations = append(tx.Observations, obs) tx.mergeObservedPathHashSize(r.pathJSON) + s.observeNodeAnalyticsLiveLocked(obs.Timestamp) tx.obsKeys[dk] = true if obs.ObserverID != "" && !tx.observerSet[obs.ObserverID] { tx.observerSet[obs.ObserverID] = true @@ -5786,6 +5818,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { } evicting := s.packets[:cutoffIdx] + s.invalidateNodeAnalyticsEvictionLocked(evicting) evictedObs := 0 var evictedBytes int64 @@ -10721,13 +10754,16 @@ func (s *PacketStore) getNodeAnalyticsAt(pubkey string, days int, now time.Time) return nil, err } + now = now.UTC() fromISO := now.Add(-time.Duration(days) * 24 * time.Hour).Format(time.RFC3339) toISO := now.Format(time.RFC3339) s.mu.RLock() defer s.mu.RUnlock() - packets := s.filterNodePacketsLocked(pubkey, fromISO) + coverage := s.nodeAnalyticsCoverageLocked(days, now) + from, _ := time.Parse(time.RFC3339Nano, coverage.From) + packets := s.filterNodePacketsLocked(pubkey, from, now) // Fase 5.2a: full analytics and the light /analytics/summary endpoint // share this single per-packet pass (see node_analytics_summary.go) so @@ -10737,8 +10773,8 @@ func (s *PacketStore) getNodeAnalyticsAt(pubkey string, days int, now time.Time) for _, p := range packets { acc.accumulate(p) } - activityTimeline, snrTrend, packetTypeBreakdown, observerCoverage, hopDistribution, peerInteractions, uptimeHeatmap := acc.finalizeDisplayArrays() - computedStats := acc.finalizeComputedStats(days) + activityTimeline, snrTrend, packetTypeBreakdown, observerCoverage, hopDistribution, peerInteractions, uptimeHeatmap := acc.finalizeDisplayArrays(coverage) + computedStats := acc.finalizeComputedStats(days, coverage) // Compute clock skew (already under RLock). clockSkew := s.getNodeClockSkewLocked(pubkey) @@ -10746,6 +10782,7 @@ func (s *PacketStore) getNodeAnalyticsAt(pubkey string, days int, now time.Time) return &NodeAnalyticsResponse{ Node: node, TimeRange: TimeRangeResp{From: fromISO, To: toISO, Days: days}, + Coverage: coverage, ActivityTimeline: activityTimeline, SnrTrend: snrTrend, PacketTypeBreakdown: packetTypeBreakdown, diff --git a/cmd/server/types.go b/cmd/server/types.go index 2f664c689..798a6939f 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -443,11 +443,11 @@ const ( // is the reliable half of a coverage picture: "where do we know wardriving // signal actually reached." type WardrivingObserverCoverage struct { - ObserverID string `json:"observerId"` - ObserverName string `json:"observerName"` - IATA string `json:"iata,omitempty"` - Lat *float64 `json:"lat,omitempty"` - Lon *float64 `json:"lon,omitempty"` + ObserverID string `json:"observerId"` + ObserverName string `json:"observerName"` + IATA string `json:"iata,omitempty"` + Lat *float64 `json:"lat,omitempty"` + Lon *float64 `json:"lon,omitempty"` // PositionSource is "real" when Lat/Lon came from the observer's own // node position, "iata" when it's the configured IATA centroid used // as a fallback (no real position known), or "none" when neither is @@ -1054,6 +1054,7 @@ type ComputedNodeStats struct { type NodeAnalyticsResponse struct { Node map[string]interface{} `json:"node"` TimeRange TimeRangeResp `json:"timeRange"` + Coverage NodeAnalyticsCoverage `json:"coverage"` ActivityTimeline []TimeBucket `json:"activityTimeline"` SnrTrend []SnrTrendEntry `json:"snrTrend"` PacketTypeBreakdown []PayloadTypeCount `json:"packetTypeBreakdown"` @@ -1066,12 +1067,13 @@ type NodeAnalyticsResponse struct { } // NodeAnalyticsSummaryResponse is the light sibling of NodeAnalyticsResponse -// (Fase 5.2a) — GET /api/nodes/{pubkey}/analytics/summary. Only timeRange + -// computedStats: no node object, no clockSkew, none of the heavy display +// (Fase 5.2a) — GET /api/nodes/{pubkey}/analytics/summary. Only timeRange, +// coverage and computedStats: no node object, no clockSkew, none of the heavy display // arrays. See PacketStore.GetNodeAnalyticsSummary. type NodeAnalyticsSummaryResponse struct { - TimeRange TimeRangeResp `json:"timeRange"` - ComputedStats ComputedNodeStats `json:"computedStats"` + TimeRange TimeRangeResp `json:"timeRange"` + Coverage NodeAnalyticsCoverage `json:"coverage"` + ComputedStats ComputedNodeStats `json:"computedStats"` } // HopAnalyticsPacket is one transmission that passed through a specific node diff --git a/docs/api-spec.md b/docs/api-spec.md index 41a300429..7386b6bfe 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -1090,6 +1090,46 @@ Per-node analytics over a time range. --- +### Node analytics history coverage (`/analytics` and `/analytics/summary`) + +Both endpoints preserve the requested `timeRange` and return an additional +`coverage` object: `{ from, to, seconds, partial, reason, loading, late_evicted_observations }`. The full +and lightweight endpoints use exactly the same coverage and computed statistics +when evaluated at the same instant. All boundaries are UTC and inclusive; +invalid timestamps and future transmissions are excluded. There is no implicit +retention-edge tolerance. + +`coverage.from` is the later of the requested start, configured retention cutoff, +successfully loaded contiguous window, and earliest persisted **observation** +evidence available in that window. An old transmission's `first_seen`, including +an old hash heard again recently, does not prove earlier coverage. Loader state +and the evidence edge are captured at startup; requests do not scan SQLite. +While startup/background loading continues, `partial` and `loading` are true. +Capped/failed loads conservatively certify only history after their completion; +memory eviction advances the edge to the evicted transmissions' `first_seen` +(the eviction boundary; packet membership is by `first_seen`). Late observations +of evicted, re-heard transmissions never move `from`; they are counted in +`late_evicted_observations` as a known, bounded gap. A later background completion cannot restore +an evicted interval. Reasons include `complete`, `available_history`, `loading`, +`unknown_history`, `load_capped`, `load_failed`, and `memory_eviction`. + +The packet set (including all display arrays and the summary numerator) is +filtered to this effective interval, without changing sender/recipient/relay +membership rules. Packets/day uses `totalPackets * 86400 / coverage.seconds`. +Availability is the fraction of supported UTC-hour time bins containing at least +one packet: a partially covered first or last hour contributes only its actual +intersection duration, to both numerator and denominator. The first timeline bin +is clipped to `coverage.from`. Zero supported duration yields zero rates and no +packets; an empty/unknown store does not claim that missing history was silent. +A node with no packets in an otherwise known supported interval has zero rates. + +This measures **available retained store history**, not uninterrupted recorder +operation, radio uptime, successful reception, or an exact physical availability +measurement. No new schema/config, historical SQL fallback, or acquisition-start +guarantee is introduced. The UI explicitly warns when the requested range is +only partially supported. It distinguishes unknown/loading history from an +inactive node in a known interval. + ## GET /api/nodes/:pubkey/reach Per-node RF reach report (two-way link quality). Computes **directional** link counts from raw diff --git a/public/node-analytics.js b/public/node-analytics.js index f8225acda..0609e4f90 100644 --- a/public/node-analytics.js +++ b/public/node-analytics.js @@ -159,6 +159,22 @@ `; + // The API's coverage is retained store evidence, not physical radio uptime. + // Use textContent for all server-provided metadata and insert one notice; + // do not interpolate coverage fields into this page's HTML template. + if (data.coverage && data.coverage.partial) { + const notice = document.createElement('p'); + notice.id = 'nodeAnalyticsCoverageNotice'; + notice.setAttribute('role', 'status'); + notice.style.cssText = 'padding:10px 12px;color:var(--text-muted);border:1px solid var(--border);border-radius:6px'; + const coverage = data.coverage; + notice.textContent = coverage.seconds > 0 + ? 'Partial history: statistics use available retained data from ' + coverage.from + ' to ' + coverage.to + ', not the entire requested ' + days + ' days. Missing history is not inactivity; this is not measured radio uptime.' + : 'Partial history: a supported history interval is not yet known. Rates are shown as 0 until retained data is available; this is not measured radio uptime.'; + if (coverage.loading) notice.textContent += ' History is still loading.'; + container.querySelector('.analytics-stats').before(notice); + } + // Time range buttons container.querySelectorAll('#timeRangeBtns button').forEach(btn => { btn.addEventListener('click', () => { diff --git a/scripts/e2e-shards.json b/scripts/e2e-shards.json index aa6289f89..e6bd00b9f 100644 --- a/scripts/e2e-shards.json +++ b/scripts/e2e-shards.json @@ -1,6 +1,14 @@ { "version": 1, "suites": [ + { + "file": "test-node-analytics-coverage-e2e.js", + "shard": 2, + "env": { "BASE_URL": "$BASE_URL" }, + "output": "e2e", + "meanSeconds": 0, + "measurementNote": "New suite; synthetic coverage metadata against the real local analytics page." + }, { "file": "test-issue-480-no-scope-e2e.js", "shard": 2, diff --git a/test-node-analytics-coverage-e2e.js b/test-node-analytics-coverage-e2e.js new file mode 100644 index 000000000..f34dfa48a --- /dev/null +++ b/test-node-analytics-coverage-e2e.js @@ -0,0 +1,65 @@ +#!/usr/bin/env node +'use strict'; + +// Exercise the real analytics page with explicit synthetic API coverage. +// No live mesh identities/data or non-local host may be used by this suite. +const assert = require('node:assert/strict'); +const { chromium } = require('playwright'); +const BASE = process.env.BASE_URL || 'http://localhost:13581'; +assert.ok(['localhost', '127.0.0.1', '[::1]'].includes(new URL(BASE).hostname), 'local server required'); +const PK = 'c048500000000000000000000000000000000000000000000000000000000000'; +const FROM = '2026-10-15T12:00:00Z'; +const TO = '2026-10-16T12:00:00Z'; + +function analytics(coverage) { + return { + node: { name: 'Coverage fixture', public_key: PK, role: 'repeater' }, + timeRange: { from: '2026-10-09T12:00:00Z', to: TO, days: 7 }, + coverage, + computedStats: { totalPackets: 24, availabilityPct: 100, avgPacketsPerDay: 24, signalGrade: 'B', uniqueObservers: 2, relayPct: 50, longestSilenceMs: 0 }, + activityTimeline: [], snrTrend: [], packetTypeBreakdown: [], observerCoverage: [], hopDistribution: [], peerInteractions: [], uptimeHeatmap: [], + }; +} + +(async () => { + const browser = await chromium.launch({ headless: true, ...(process.env.CHROMIUM_PATH ? { executablePath: process.env.CHROMIUM_PATH } : {}) }); + let passed = 0; + try { + for (const width of [1440, 375]) { + for (const state of ['partial', 'complete', 'unknown']) { + const context = await browser.newContext({ viewport: { width, height: 900 } }); + const errors = []; + const coverage = { from: FROM, to: TO, seconds: state === 'unknown' ? 0 : 86400, partial: state !== 'complete', loading: state === 'unknown', reason: state === 'unknown' ? 'loading' : state === 'complete' ? 'complete' : 'available_history' }; + await context.route(url => url.pathname === `/api/nodes/${PK}/analytics`, route => route.fulfill({ status: 200, contentType: 'application/json', body: JSON.stringify(analytics(coverage)) })); + await context.route(url => url.pathname.startsWith(`/api/nodes/${PK}/`) && !url.pathname.endsWith('/analytics'), route => route.fulfill({ status: 200, contentType: 'application/json', body: JSON.stringify({ packets: [], samples: [] }) })); + const page = await context.newPage(); + page.on('pageerror', error => errors.push(error.message)); + await page.goto(`${BASE}/#/nodes/${PK}/analytics`, { waitUntil: 'domcontentloaded' }); + await page.locator('.analytics-stats').waitFor(); + const notice = page.locator('#nodeAnalyticsCoverageNotice'); + if (state === 'complete') { + assert.equal(await notice.count(), 0, 'a complete range must not show partial notice'); + } else { + await notice.waitFor({ state: 'visible' }); + const text = await notice.textContent(); + assert.match(text, /Partial history/); + assert.match(text, /not measured radio uptime/); + assert.equal(await notice.getAttribute('role'), 'status'); + if (state === 'partial') { + assert.ok(text.includes(FROM) && text.includes(TO), 'effective interval must be visible'); + assert.match(text, /not the entire requested 7 days/); + } else { + assert.match(text, /supported history interval is not yet known/); + assert.match(text, /still loading/); + } + assert.ok(await notice.evaluate(el => el.getBoundingClientRect().right <= window.innerWidth), 'notice must fit mobile/desktop viewport'); + } + assert.deepEqual(errors, [], 'analytics page errors'); + console.log(` ✓ ${state} coverage notice at ${width}px`); + passed++; + await context.close(); + } + } + } finally { await browser.close(); } + console.log(`Node analytics coverage: ${passed}/6 passed`); +})().catch(error => { console.error(error); process.exitCode = 1; });