diff --git a/AGENTS.md b/AGENTS.md index 293ba0426..fd35106a0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -211,7 +211,8 @@ All colors MUST use CSS variables. Never hardcode `#hex` values outside of `:roo ### last_heard vs last_seen - `last_seen` = DB timestamp, only updates on adverts/direct upserts -- `last_heard` = from in-memory packet store, updates on ALL traffic +- `last_heard` = the newer of `last_seen` and the certain last heard from the in-memory packet store (#2260): the node's own packets plus relay hops whose prefix can only be this node. A 1-byte hop shared with other repeaters is NOT evidence, and neither is `last_relayed` (it counts those shared prefixes too). Liveness — status, Last Heard, the notify offline check — uses only this. +- Status from `getNodeStatus(n)` is `'active'`, `'stale'` or, for a repeater or room, `'uncertain'`: nothing certain within the silent window, but `last_relayed` (its shared prefix in a path) is. Treat `'uncertain'` as not proven alive (dim it like stale); never as active. - Always prefer `n.last_heard || n.last_seen` for display and status calculation ### Packet Filter (packet-filter.js) diff --git a/cmd/server/adaptive_offline.go b/cmd/server/adaptive_offline.go new file mode 100644 index 000000000..4dd16fd6b --- /dev/null +++ b/cmd/server/adaptive_offline.go @@ -0,0 +1,264 @@ +package main + +import ( + "math" + "sort" + "strings" + "time" +) + +// Adaptive offline threshold (#2260, part 2). The fixed infraSilentHours +// (72 h by default) treats every repeater the same: one heard every minute +// that goes quiet for an hour is suspicious, a node that sends a few adverts +// a day is not. For node.offline, a watched node's threshold is learned from +// its own certain traffic (certainFor), so an outage is mailed within the +// hour instead of three days later. +// +// Learning: certain packets per hour over the week before the node's last +// certain packet. Ending there, not at now, keeps a silence in progress out +// of the rate: otherwise the outage itself drags the rate down until the +// fixed threshold returns and the node reads "back online". +// +// Threshold: arrivals treated as Poisson, P(no packet in t) = e^(-λt); for +// that below one in a million t = ln(10⁶)/λ ≈ 14/λ hours, with λ the 10th +// percentile of the hourly counts so quiet nights do not alert. +// +// Empty runs of hours are classified first. A run of at most +// adaptiveMaxNormalRun hours that recurs (another empty run on a different +// day covers one of the same hours of the day) is the node's normal +// silence: it stays in the percentile, and the threshold is never shorter +// than the longest such run plus an hour, so it does not mail offline and +// online every night. Any other run is an outage: it is left out of the +// learning entirely, so one outage neither stretches the threshold for a +// week nor makes the node unlearnable. +// +// Clamped to [floor, the role's silent window]. Nodes with too little +// certain traffic (λ < 1/hour, under a day of history) keep the fixed +// window, and so do nodes whose 10th-percentile hour was heard by fewer +// than minObservers observers: when one observer carries the node, that +// observer dropping out must not read as the node going offline. + +const ( + adaptiveLearnWindow = 7 * 24 * time.Hour + adaptiveMinHistory = 24 // hours of history before a rate is trusted + adaptiveMinRate = 1.0 // certain packets per hour at the 10th percentile + adaptiveSurpriseHours = 13.815510557964 // ln(10⁶): P(silence) < 10⁻⁶ + adaptiveMaxNormalRun = 6 // longest empty run in hours that may be normal silence + adaptiveRefreshEvery = 10 * time.Minute +) + +// certainActivity is a node's certain traffic in its learning window. +type certainActivity struct { + From time.Time // start of Hourly[0], UTC, on the hour + Hourly []int // certain packets per hour, oldest first, zero hours included + HourObservers []int // distinct observers per hour of those packets +} + +// CertainActivity returns, per pubkey with a certain last heard, its certain +// packets per hour over adaptiveLearnWindow up to that last heard (capped at +// the store's covered window) and how many observers heard each hour. +// O(packets of the given nodes), with the read lock taken per node so ingest +// waits for one node's walk at most, not all of them. +func (s *PacketStore) CertainActivity(pubkeys []string, now time.Time) map[string]certainActivity { + now = now.UTC() // FirstSeen is UTC; a local offset breaks the string bounds + _, pm := s.getCachedNodesAndPM() + s.mu.RLock() + covered, _ := s.coverageStart(now) + s.mu.RUnlock() + covered = covered.UTC() + out := make(map[string]certainActivity, len(pubkeys)) + for _, pk := range pubkeys { + if act, ok := s.certainActivityOf(pk, covered, pm); ok { + out[pk] = act + } + } + return out +} + +// certainActivityOf is CertainActivity for one node, under its own read lock. +func (s *PacketStore) certainActivityOf(pk string, covered time.Time, pm *prefixMap) (certainActivity, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + last, ok := parseRelayTS(s.certainHeard[pk]) + if !ok { + return certainActivity{}, false + } + end := last.UTC().Truncate(time.Hour).Add(time.Hour) + from := end.Add(-adaptiveLearnWindow) + if covered.After(from) { + from = covered.Truncate(time.Hour).Add(time.Hour) // first whole hour held + } + hours := int(end.Sub(from) / time.Hour) + if hours <= 0 { + return certainActivity{}, false + } + act := certainActivity{From: from, Hourly: make([]int, hours), HourObservers: make([]int, hours)} + observers := make([]map[string]struct{}, hours) + // Pre-filter on the date only: legacy FirstSeen values use a space + // instead of T and would sort below a same-day RFC3339 bound. The parsed + // time below decides. + fromDate := from.Format("2006-01-02") + for _, tx := range s.byNode[pk] { + if tx == nil || len(tx.FirstSeen) < 10 || tx.FirstSeen[:10] < fromDate || !certainFor(tx, pk, pm) { + continue + } + t, ok := parseStoreTime(tx.FirstSeen) + if !ok || t.Before(from) || !t.Before(end) { + continue + } + h := int(t.Sub(from) / time.Hour) + act.Hourly[h]++ + for _, o := range tx.Observations { + if o != nil && o.ObserverID != "" { + if observers[h] == nil { + observers[h] = map[string]struct{}{} + } + observers[h][o.ObserverID] = struct{}{} + } + } + } + for h, set := range observers { + act.HourObservers[h] = len(set) + } + return act, true +} + +// adaptiveOfflineAfter is the learned silence after which a node with this +// activity counts as offline, floored at floor; ok is false when the +// activity is too thin to learn from (see the file comment). The caller +// caps it at the role's silent window. +func adaptiveOfflineAfter(act certainActivity, floor time.Duration, minObservers int) (time.Duration, bool) { + outage, normalRun := classifySilences(act) + var counts, observers []int + for h, c := range act.Hourly { + if outage[h] { + continue + } + counts = append(counts, c) + if c > 0 && h < len(act.HourObservers) { + observers = append(observers, act.HourObservers[h]) + } + } + if len(counts) < adaptiveMinHistory || len(observers) == 0 { + return 0, false + } + sort.Ints(counts) + rate := float64(counts[len(counts)/10]) + if rate < adaptiveMinRate { + return 0, false + } + sort.Ints(observers) + if observers[len(observers)/10] < minObservers { + return 0, false + } + after := time.Duration(math.Ceil(adaptiveSurpriseHours / rate * float64(time.Hour))) + if normalRun > 0 { + // k empty buckets in a row can hide up to k+1 hours of silence. + if seen := time.Duration(normalRun+1) * time.Hour; after < seen { + after = seen + } + } + if after < floor { + after = floor + } + return after, true +} + +// classifySilences marks the hours of outage runs (empty runs that are not +// normal silence, see the file comment) and returns the longest normal run. +func classifySilences(act certainActivity) (outage []bool, longestNormal int) { + type run struct{ start, n int } + var runs []run + for h := 0; h < len(act.Hourly); h++ { + if act.Hourly[h] != 0 { + continue + } + r := run{start: h} + for h < len(act.Hourly) && act.Hourly[h] == 0 { + r.n++ + h++ + } + runs = append(runs, r) + } + day := func(h int) int { return int(act.From.Add(time.Duration(h)*time.Hour).Unix() / 86400) } + hod := func(h int) int { return act.From.Add(time.Duration(h) * time.Hour).Hour() } + recurs := func(i int) bool { + a := runs[i] + for j, b := range runs { + if j == i || day(b.start) == day(a.start) { + continue + } + for x := a.start; x < a.start+a.n; x++ { + for y := b.start; y < b.start+b.n; y++ { + if hod(x) == hod(y) { + return true + } + } + } + } + return false + } + outage = make([]bool, len(act.Hourly)) + for i, r := range runs { + if r.n <= adaptiveMaxNormalRun && recurs(i) { + if r.n > longestNormal { + longestNormal = r.n + } + continue + } + for h := r.start; h < r.start+r.n; h++ { + outage[h] = true + } + } + return outage, longestNormal +} + +// adaptiveEntry is one cached threshold; ok false caches "not learnable". +type adaptiveEntry struct { + after time.Duration + ok bool + at time.Time +} + +// offlineWindows returns the learned offline window per watched pubkey that +// has one, recomputing entries older than adaptiveRefreshEvery: the rate +// moves slowly and the walk over a week of packets need not run every tick. +// Entries for nodes no longer watched are dropped. Loop goroutine only. +func (n *notifier) offlineWindows(pubkeys []string, now time.Time) map[string]time.Duration { + set := n.a.set.notify.adaptive + if !set.enabled || len(pubkeys) == 0 { + n.adaptive = nil + return nil + } + if n.adaptive == nil { + n.adaptive = map[string]adaptiveEntry{} + } + watched := make(map[string]bool, len(pubkeys)) + var stale []string + for _, pk := range pubkeys { + pk = strings.ToLower(pk) + watched[pk] = true + if e, ok := n.adaptive[pk]; !ok || now.Sub(e.at) >= adaptiveRefreshEvery { + stale = append(stale, pk) + } + } + for pk := range n.adaptive { + if !watched[pk] { + delete(n.adaptive, pk) + } + } + if len(stale) > 0 { + acts := n.src.certainActivity(stale, now) + for _, pk := range stale { + after, ok := adaptiveOfflineAfter(acts[pk], set.floor, set.minObservers) + n.adaptive[pk] = adaptiveEntry{after: after, ok: ok, at: now} + } + } + out := make(map[string]time.Duration, len(n.adaptive)) + for pk, e := range n.adaptive { + if e.ok { + out[pk] = e.after + } + } + return out +} diff --git a/cmd/server/adaptive_offline_test.go b/cmd/server/adaptive_offline_test.go new file mode 100644 index 000000000..d5fd2867c --- /dev/null +++ b/cmd/server/adaptive_offline_test.go @@ -0,0 +1,453 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http/httptest" + "testing" + "time" + + "github.com/meshcore-analyzer/users" +) + +// chStoreWithPM is chStore with the three-repeater prefix map cached, so +// getCachedNodesAndPM does not go to SQLite. +func chStoreWithPM() *PacketStore { + s := chStore() + s.nodePM = chPM() + s.nodeCacheTime = time.Now() + return s +} + +func TestCertainFor(t *testing.T) { + pm := chPM() + at := time.Date(2026, 10, 18, 16, 0, 0, 0, time.UTC) + // Zero-hop advert: certain only through its sender. (certainFor assumes + // tx is in byNode[pk]; a flood packet with chC's unique prefix would + // rightly count as chC relaying it.) + advert := chTx(1, 2, 4, 1, at) + advert.DecodedJSON = fmt.Sprintf(`{"pubKey":%q}`, chA) + anon := chTx(2, 2, 7, 1, at) // direct route, but the sender is certain + anon.DecodedJSON = fmt.Sprintf(`{"srcPubKey":%q}`, chA) + cases := []struct { + name string + tx *StoreTx + pk string + want bool + }{ + {"own advert", advert, chA, true}, + {"own anon request, direct route", anon, chA, true}, + {"2-byte flood hop, unique", chTx(3, 1, 5, 2, at), chA, true}, + {"1-byte flood hop shared with chB", chTx(4, 1, 5, 1, at), chA, false}, + {"1-byte flood hop, unique first byte", chTx(5, 1, 5, 1, at), chC, true}, + {"direct-route hop", chTx(6, 2, 5, 2, at), chA, false}, + {"TRACE", chTx(7, 1, packetpathTrace, 2, at), chA, false}, + {"someone else's advert", advert, chC, false}, + {"nil tx", nil, chA, false}, + } + for _, c := range cases { + if got := certainFor(c.tx, c.pk, pm); got != c.want { + t.Errorf("%s: %v; want %v", c.name, got, c.want) + } + } +} + +func TestLongestSilence(t *testing.T) { + now := time.Date(2026, 10, 22, 12, 30, 0, 0, time.UTC) + from := time.Date(2026, 10, 22, 6, 30, 0, 0, time.UTC) + h := func(s string) string { return "2026-10-" + s + ":00:00Z" } + cases := []struct { + name string + buckets []string + from time.Time + wantMs int + start interface{} + }{ + {"no certain bucket: the whole window", nil, from, 6 * 3600000, "2026-10-22T06:30:00Z"}, + {"gap between buckets", []string{h("22T07"), h("22T09"), h("22T10"), h("22T11"), h("22T12")}, from, 2 * 3600000, h("22T07")}, + {"silence before the first bucket", []string{h("22T10"), h("22T11"), h("22T12")}, from, int(3.5 * 3600000), "2026-10-22T06:30:00Z"}, + // #2260: the outage in progress is the longest silence. + {"running silence", []string{h("18T15"), h("18T16")}, time.Date(2026, 10, 18, 15, 0, 0, 0, time.UTC), int(now.Sub(time.Date(2026, 10, 18, 16, 0, 0, 0, time.UTC)).Milliseconds()), h("18T16")}, + {"current hour is not a silence", []string{h("22T12")}, time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC), 0, nil}, + } + for _, c := range cases { + ms, start := longestSilence(c.buckets, c.from, now) + if ms != c.wantMs || start != c.start { + t.Errorf("%s: %d from %v; want %d from %v", c.name, ms, start, c.wantMs, c.start) + } + } +} + +func hourly(n, v int) []int { + out := make([]int, n) + for i := range out { + out[i] = v + } + return out +} + +// learnFrom is a midnight-UTC start, so hour index h is hour h%24 of day h/24. +var learnFrom = time.Date(2026, 10, 11, 0, 0, 0, 0, time.UTC) + +// act builds a learning week with obs observers on every hour that has +// packets. +func act(counts []int, obs int) certainActivity { + o := make([]int, len(counts)) + for i, c := range counts { + if c > 0 { + o[i] = obs + } + } + return certainActivity{From: learnFrom, Hourly: counts, HourObservers: o} +} + +func zeroAt(counts []int, hours ...int) []int { + out := append([]int(nil), counts...) + for _, h := range hours { + out[h] = 0 + } + return out +} + +func span(from, n int) []int { + out := make([]int, n) + for i := range out { + out[i] = from + i + } + return out +} + +func TestAdaptiveOfflineAfter(t *testing.T) { + floor := 30 * time.Minute + learned := func(rate float64) time.Duration { + return time.Duration(adaptiveSurpriseHours / rate * float64(time.Hour)) + } + + quietNights := hourly(168, 20) + var nightly2h, nightly4h, everyFifth []int + for d := 0; d < 7; d++ { + for j := 0; j < 6; j++ { + quietNights[d*24+j] = 3 // six quiet hours a night: the 10th percentile + } + nightly2h = append(nightly2h, d*24+2, d*24+3) + nightly4h = append(nightly4h, d*24+1, d*24+2, d*24+3, d*24+4) + } + for i := 0; i < 30; i++ { + everyFifth = append(everyFifth, i*5) + } + busy := hourly(168, 35) + + dominant := act(busy, 1) // one observer carries the node every hour + dominant.HourObservers[5] = 3 + + cases := []struct { + name string + act certainActivity + want time.Duration + wantOK bool + }{ + // 839 packets/day (#2260): 14/35 h ≈ 24 min, so the floor applies. + {"busy repeater hits the floor", act(busy, 32), floor, true}, + {"5 per hour", act(hourly(168, 5), 5), learned(5), true}, + {"quiet nights set the rate", act(quietNights, 5), learned(3), true}, + {"under a day of history", act(hourly(23, 35), 32), 0, false}, + {"too few observers in the 10th-percentile hour", act(busy, 2), 0, false}, + // Review: a node heard 99 % through one observer passed a weekly + // count of 3; per hour it is carried by one. + {"one observer carries the node", dominant, 0, false}, + // Two empty hours every night recur: normal silence, the window is + // at least 3 h (8 % of hours empty keeps the rate at 35). + {"recurring 2-hour nightly silence lengthens the window", act(zeroAt(busy, nightly2h...), 5), 3 * time.Hour, true}, + // Four empty hours every night are 17 % of the week: too gappy to learn. + {"recurring 4-hour nightly silence: not learnable", act(zeroAt(busy, nightly4h...), 5), 0, false}, + // Review: one 6-hour gap is an outage, not normal silence. Left out, + // it neither lengthens the window nor blocks learning. + {"one-off 6-hour gap is an outage", act(zeroAt(busy, span(100, 6)...), 5), floor, true}, + // Review: an outage over ~17 h used to make the node unlearnable + // for a week (the zero hours took the 10th percentile to 0). + {"30-hour outage stays learnable", act(zeroAt(busy, span(50, 30)...), 5), floor, true}, + // Recurring but longer than adaptiveMaxNormalRun: an outage. + {"recurring 8-hour gaps are outages", act(zeroAt(busy, append(span(10, 8), span(34, 8)...)...), 5), floor, true}, + // A single empty hour every 5 hours: hour-of-day only repeats on a + // later day for 12 of the 30, those are normal silence (window at + // least 2 h); the 18 one-offs are left out of the learning. + {"scattered empty hours: only recurring ones count", act(zeroAt(hourly(168, 40), everyFifth...), 5), 2 * time.Hour, true}, + {"nothing", certainActivity{}, 0, false}, + } + for _, c := range cases { + got, ok := adaptiveOfflineAfter(c.act, floor, 3) + // Ceil to the nanosecond in the code; compare to the second. + if ok != c.wantOK || got.Round(time.Second) != c.want.Round(time.Second) { + t.Errorf("%s: %v %v; want %v %v", c.name, got, ok, c.want, c.wantOK) + } + } +} + +// CertainActivity learns from the week before the last certain packet, so +// an outage does not drag the rate down, and counts only certain packets. +func TestCertainActivityEndsAtLastCertainPacket(t *testing.T) { + s := chStoreWithPM() + now := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC) + lastCertain := time.Date(2026, 10, 18, 16, 49, 0, 0, time.UTC) + hopsSeen := map[string]bool{} + id := 0 + for h := 0; h < 48; h++ { + at := lastCertain.Add(-time.Duration(h) * time.Hour) + for k := 0; k < 3; k++ { + id++ + tx := chTx(id, 1, 5, 2, at.Add(-time.Duration(k)*time.Minute)) + tx.Observations = []*StoreObs{{ObserverID: fmt.Sprintf("obs%d", k)}} + s.indexResolvedPathHops(tx, []string{chA}, hopsSeen, s.nodePM) + } + } + // After the outage only the shared 1-byte prefix: attributed, not certain. + for h := 1; h < 60; h++ { + id++ + s.indexResolvedPathHops(chTx(id, 1, 5, 1, lastCertain.Add(time.Duration(h)*time.Hour)), []string{chA}, hopsSeen, s.nodePM) + } + + a := s.CertainActivity([]string{chA, chC}, now)[chA] + if want := int(adaptiveLearnWindow / time.Hour); len(a.Hourly) != want { + t.Fatalf("hours = %d; want %d ending at the last certain packet", len(a.Hourly), want) + } + total := 0 + for _, c := range a.Hourly { + total += c + } + if total != 48*3 { + t.Fatalf("total %d; want 144 certain packets", total) + } + if got := a.HourObservers[len(a.HourObservers)-1]; got != 3 { + t.Errorf("observers in the last hour = %d; want 3", got) + } + if !a.From.Equal(lastCertain.Truncate(time.Hour).Add(time.Hour).Add(-adaptiveLearnWindow)) { + t.Errorf("From = %v", a.From) + } + if got := a.Hourly[len(a.Hourly)-1]; got != 3 { + t.Errorf("last hour = %d; want the 3 packets of the last certain hour", got) + } + if _, ok := s.CertainActivity([]string{chC}, now)[chC]; ok { + t.Error("a node never certainly heard has no activity") + } +} + +// End to end through the notifier: a busy repeater learned at ~3 per hour +// goes offline in mail after its learned window, not after 72 h; a node +// without enough observers keeps the fixed window; with the setting off +// nothing is learned or asked. +func TestNotifierAdaptiveOffline(t *testing.T) { + cases := []struct { + name string + enabled bool + observers int + silentFor time.Duration + want string + }{ + {"learned window passed", true, 5, 5 * time.Hour, users.NotifyBad}, + {"inside the learned window", true, 5, 4 * time.Hour, users.NotifyGood}, + {"too few observers: fixed 72 h", true, 2, 5 * time.Hour, users.NotifyGood}, + {"setting off: fixed 72 h", false, 5, 5 * time.Hour, users.NotifyGood}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + in := evInput() + last := evNow.Add(-c.silentFor) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: "repeater", LastSeen: last.Format(time.RFC3339)} + in.Heard[evPkA] = last + + src := &fakeNotifySource{acts: map[string]certainActivity{evPkA: act(hourly(168, 3), c.observers)}} + n := ¬ifier{a: &authService{set: &userMgmtSettings{}}, src: src} + n.a.set.notify.adaptive = resolveAdaptiveOffline(&AdaptiveOfflineConfig{Enabled: c.enabled}) + in.OfflineAfter = n.offlineWindows([]string{evPkA}, evNow) + + if got, _ := evState(evaluateNotifications(in), evUser, users.NotifyNodeOffline, evPkA); got != c.want { + t.Fatalf("node.offline = %q; want %q (window %v)", got, c.want, in.OfflineAfter) + } + if !c.enabled && len(src.actsAsked) != 0 { + t.Fatal("activity asked with the setting off") + } + }) + } +} + +func TestNotifierAdaptiveCacheRefreshAndPrune(t *testing.T) { + src := &fakeNotifySource{acts: map[string]certainActivity{ + evPkA: act(hourly(168, 3), 5), + evPkB: act(hourly(168, 3), 5), + }} + n := ¬ifier{a: &authService{set: &userMgmtSettings{}}, src: src} + n.a.set.notify.adaptive = resolveAdaptiveOffline(&AdaptiveOfflineConfig{Enabled: true}) + + n.offlineWindows([]string{evPkA, evPkB}, evNow) + n.offlineWindows([]string{evPkA, evPkB}, evNow.Add(adaptiveRefreshEvery-time.Second)) + if len(src.actsAsked) != 1 { + t.Fatalf("asked %d times inside the refresh period; want 1", len(src.actsAsked)) + } + n.offlineWindows([]string{evPkA}, evNow.Add(adaptiveRefreshEvery)) + if len(src.actsAsked) != 2 || len(src.actsAsked[1]) != 1 { + t.Fatalf("asked %v; want one refresh of the still-watched node", src.actsAsked) + } + if _, ok := n.adaptive[evPkB]; ok { + t.Fatal("an unwatched node stays cached") + } +} + +func TestResolveAdaptiveOffline(t *testing.T) { + if s := resolveAdaptiveOffline(nil); s.enabled { + t.Fatal("absent block must be off") + } + if s := resolveAdaptiveOffline(&AdaptiveOfflineConfig{}); s.enabled { + t.Fatal("enabled false must be off") + } + s := resolveAdaptiveOffline(&AdaptiveOfflineConfig{Enabled: true}) + if s.floor != 30*time.Minute || s.minObservers != 3 { + t.Fatalf("defaults = %+v", s) + } + s = resolveAdaptiveOffline(&AdaptiveOfflineConfig{Enabled: true, FloorMinutes: 45, MinObservers: 5}) + if s.floor != 45*time.Minute || s.minObservers != 5 { + t.Fatalf("configured = %+v", s) + } +} + +// The node analytics endpoint splits each timeline bucket and takes +// availability from certain buckets only. +func TestNodeAnalyticsCertainTimeline(t *testing.T) { + srv, router := setupTestServer(t) + const pk = "aabbccdd11223344" + // A second repeater sharing the first byte makes 1-byte hop "aa" ambiguous. + srv.db.conn.Exec(`INSERT INTO nodes (public_key, name, role) VALUES ('aa00000000000000', 'Namesake', 'repeater')`) + srv.store.InvalidateNodeCache() + + hour := time.Now().UTC().Add(-3 * time.Hour).Truncate(time.Hour) + srv.store.mu.Lock() + srv.store.indexResolvedPathHops(chTx(900001, 1, 5, 1, hour.Add(10*time.Minute)), []string{pk}, map[string]bool{}, nil) + srv.store.mu.Unlock() + + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/nodes/"+pk+"/analytics?days=1", nil)) + if w.Code != 200 { + t.Fatalf("status %d: %s", w.Code, w.Body.String()) + } + var body struct { + ActivityTimeline []TimeBucket `json:"activityTimeline"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + want := hour.Format(time.RFC3339) + for _, b := range body.ActivityTimeline { + if b.Bucket != nil && *b.Bucket == want { + if b.Certain == nil || *b.Certain != 0 || b.Count != 1 { + t.Fatalf("bucket %s: count %d certain %v; want 1 packet, 0 certain", want, b.Count, b.Certain) + } + return + } + } + t.Fatalf("bucket %s missing from %d buckets", want, len(body.ActivityTimeline)) +} + +// CertainActivity walks a week of packets per watched node under the read +// lock, at most once per node per adaptiveRefreshEvery. 50 watched busy +// repeaters with 6,000 packets each (~840 a day, the #2260 node). Quote in +// the PR. +func BenchmarkCertainActivity(b *testing.B) { + s := chStoreWithPM() + end := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC) + keys := make([]string, 0, 50) + s.certainHeard = map[string]string{} + id := 0 + for n := 0; n < 50; n++ { + pk := fmt.Sprintf("%064x", n+1) + keys = append(keys, pk) + for i := 0; i < 6000; i++ { + id++ + tx := chTx(id, 1, 5, 2, end.Add(-time.Duration(i)*100*time.Second)) + tx.Observations = []*StoreObs{{ObserverID: "obs1"}, {ObserverID: "obs2"}} + s.byNode[pk] = append(s.byNode[pk], tx) + } + s.certainHeard[pk] = end.Format(time.RFC3339) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.CertainActivity(keys, end) + } +} + +// Review: a node stored offline comes back only on evidence from that moment +// on. Here its learned window is gone (retention dropped the history before +// the outage, or a restart) and the fixed 72 h window has not passed: it +// must stay offline, not be mailed "back online". +func TestOfflineNodeNeedsNewEvidenceToComeBack(t *testing.T) { + lastCertain := evNow.Add(-49 * time.Hour) + in := evInput() + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: "repeater", LastSeen: lastCertain.Format(time.RFC3339)} + in.Heard[evPkA] = lastCertain + row := evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyBad) + row.ChangedAt = lastCertain.Add(30 * time.Minute) // mailed offline by the learned window + in.States = []users.NotifyState{row} + res := evaluateNotifications(in) + // Unchanged states are not stored again; a good one would be. + if got, _ := evState(res, evUser, users.NotifyNodeOffline, evPkA); got == users.NotifyGood { + t.Fatal("node.offline turned good; nothing new was heard") + } + if len(res.Changes[evUser]) != 0 { + t.Fatalf("changes = %+v; want no back-online mail", res.Changes[evUser]) + } + + // Heard again after it went offline: back online. + in.Heard[evPkA] = evNow.Add(-time.Minute) + if ch := evaluateNotifications(in).Changes[evUser]; len(ch) != 1 || ch[0].To != users.NotifyGood { + t.Fatalf("changes = %+v; want one back-online change", ch) + } +} + +// Review (#2262): with a 30-minute learned window and the 30-minute stale +// threshold, a feed that stalls for 29 minutes must not mail a node whose +// last packet came 3 minutes before the stall. Silence is measured up to +// the newest packet in the store. +func TestStalledFeedStopsTheSilenceClock(t *testing.T) { + in := evInput() + last := evNow.Add(-32 * time.Minute) + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: "repeater", LastSeen: last.Format(time.RFC3339)} + in.Heard[evPkA] = last + in.OfflineAfter = map[string]time.Duration{evPkA: 30 * time.Minute} + in.States = []users.NotifyState{evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood)} + + in.NewestPacket = evNow.Add(-29 * time.Minute) + if ch := evaluateNotifications(in).Changes[evUser]; len(ch) != 0 { + t.Fatalf("changes = %+v; the feed stalled, the node did not", ch) + } + in.NewestPacket = evNow // the feed is live: the node is really silent + if ch := evaluateNotifications(in).Changes[evUser]; len(ch) != 1 || ch[0].To != users.NotifyBad { + t.Fatalf("changes = %+v; want offline", ch) + } +} + +// Review (#2262): the window bounds are UTC whatever the caller's zone, and +// a legacy "2006-01-02 15:04:05" FirstSeen on the window's first day counts. +func TestCertainActivityUTCBoundsAndLegacyTimestamps(t *testing.T) { + s := chStoreWithPM() + s.retentionHours = 30 // the window start binds on retention, not on the week + lastCertain := time.Date(2026, 10, 22, 21, 10, 0, 0, time.UTC) + hopsSeen := map[string]bool{} + for h := 0; h < 28; h++ { + s.indexResolvedPathHops(chTx(h+1, 1, 5, 2, lastCertain.Add(-time.Duration(h)*time.Hour)), []string{chA}, hopsSeen, s.nodePM) + } + // Local now just after midnight (+02:00): the UTC date is a day earlier. + now := time.Date(2026, 10, 23, 0, 30, 0, 0, time.FixedZone("CEST", 2*3600)) + a := s.CertainActivity([]string{chA}, now)[chA] + if a.From.Location() != time.UTC { + t.Fatalf("From = %v; want UTC", a.From) + } + firstHour := a.From + legacy := chTx(999, 1, 5, 2, firstHour) + legacy.FirstSeen = firstHour.Add(20 * time.Minute).Format("2006-01-02 15:04:05") + s.mu.Lock() + s.byNode[chA] = append(s.byNode[chA], legacy) + s.mu.Unlock() + b := s.CertainActivity([]string{chA}, now)[chA] + if b.Hourly[0] != a.Hourly[0]+1 { + t.Fatalf("first hour %d -> %d; the legacy timestamp was dropped", a.Hourly[0], b.Hourly[0]) + } +} diff --git a/cmd/server/certain_heard.go b/cmd/server/certain_heard.go new file mode 100644 index 000000000..0f10aa13e --- /dev/null +++ b/cmd/server/certain_heard.go @@ -0,0 +1,218 @@ +package main + +import ( + "strings" + "time" + + "github.com/meshcore-analyzer/packetpath" +) + +// Certain liveness (#2260). certainHeard holds, per pubkey, the newest +// FirstSeen of a transmission that can only have come from that node: +// +// - its own packets (decoded pubKey / srcPubKey), and +// - flood relay hops whose prefix, at the packet's path hash size, matches +// exactly one node that can appear in a path, and no retired one +// (inactive_nodes) that may still relay. +// +// A 1-byte hop shared by several repeaters is not evidence: the resolver may +// still attribute it (affinity, geography) for analytics, but a silent node +// must not stay "Active" on traffic from a namesake prefix. The firmware +// lets the originator pick the hash size for the whole path +// (Mesh::sendFlood) and every relay appends its own hash at that size +// (Mesh::routeRecvPacket), so a multibyte repeater legitimately appears as a +// 1-byte hop; only its uniqueness decides. Direct-route paths list the hops +// still to come (Mesh::removeSelfFromPath), and TRACE reuses the path bytes +// for SNR, so neither proves a hop transmitted. +// +// Liveness reads it: Last Heard, the node status, the nodes list and the +// notify offline check. One entry per node that was ever certainly heard, +// dropped with the node's byNode list on eviction. + +// noteCertainHeard records tx as certain evidence for pubkey. Caller holds +// s.mu for writing. +func (s *PacketStore) noteCertainHeard(pubkey string, tx *StoreTx) { + if pubkey == "" || tx == nil || tx.FirstSeen == "" { + return + } + if s.certainHeard == nil { + s.certainHeard = make(map[string]string) + } + if tx.FirstSeen > s.certainHeard[pubkey] { + s.certainHeard[pubkey] = tx.FirstSeen + } +} + +// signatureFailed reports a decoded advert whose signature did not verify +// (signatureValid false): a corrupted or forged advert is no evidence the +// node transmitted. Absent means not checked, which is not a failure. +func (tx *StoreTx) signatureFailed() bool { + v, ok := tx.ParsedDecoded()["signatureValid"].(bool) + return ok && !v +} + +// relayHopHexLen is the hex length of one relay hop in tx's path, or 0 when +// the path is no evidence of who transmitted: not a flood route, a TRACE, or +// an unreadable header. +func relayHopHexLen(tx *StoreTx) int { + if tx.PayloadType != nil && !packetpath.PathBytesAreHops(byte(*tx.PayloadType)) { + return 0 + } + routeType, pathByte, ok := packetpath.PathByte(tx.RawHex) + if !ok || (routeType != packetpath.RouteFlood && routeType != packetpath.RouteTransportFlood) { + return 0 + } + size := int(pathByte>>6) + 1 + if size > 3 { // reserved (firmware Packet::isValidPathLen) + return 0 + } + return 2 * size +} + +// noteCertainRelays records tx for each resolved relay hop pubkey whose hop +// prefix is unique among path-capable nodes in pm. pm nil records nothing. +// Caller holds s.mu for writing. +func (s *PacketStore) noteCertainRelays(tx *StoreTx, pks []string, pm *prefixMap) { + if pm == nil || len(pks) == 0 { + return + } + n := relayHopHexLen(tx) + if n == 0 { + return + } + for _, pk := range pks { + if uniqueHopPrefix(pk, n, pm) { + s.noteCertainHeard(pk, tx) + } + } +} + +// uniqueHopPrefix reports whether pk's first n hex characters, a relay hop +// at that hash size, match exactly one path-capable node in pm, pk itself, +// and no retired one. A repeater that never advertised is unknown to both +// and cannot be ruled out. +func uniqueHopPrefix(pk string, n int, pm *prefixMap) bool { + if pm == nil || n == 0 || len(pk) < n { + return false + } + prefix := strings.ToLower(pk[:n]) + if _, shared := pm.retired[prefix]; shared { + return false + } + c := pm.relayCandidates(prefix) + return len(c) == 1 && strings.EqualFold(c[0].PublicKey, pk) +} + +// markRetired loads the hop prefixes (1 to 3 bytes) of the path-capable +// nodes in inactive_nodes into pm.retired. Rows of nodes also in nodes are +// skipped: retention copies a node there and nothing removes it when the +// node is heard again, and a node is no namesake of itself. A database +// without the table leaves it empty. +func (pm *prefixMap) markRetired(db *DB) error { + if db == nil || !db.inactiveNodesPresent() { + return nil + } + rows, err := db.conn.Query(`SELECT public_key, COALESCE(role, '') FROM inactive_nodes + WHERE lower(public_key) NOT IN (SELECT lower(public_key) FROM nodes)`) + if err != nil { + return err + } + defer rows.Close() + retired := map[string]struct{}{} + for rows.Next() { + var pk, role string + if err := rows.Scan(&pk, &role); err != nil { + return err + } + if !canAppearInPath(role) { + continue + } + pk = strings.ToLower(pk) + for n := 2; n <= 6 && n <= len(pk); n += 2 { + retired[pk[:n]] = struct{}{} + } + } + if err := rows.Err(); err != nil { + return err + } + pm.retired = retired + return nil +} + +// certainFor reports whether tx, listed in byNode[pk], is certain evidence +// for pk under the same rule as certainHeard: pk sent it, or it is a flood +// relay whose hop prefix only pk has. Packet-level counterpart of +// certainHeard for analytics and the adaptive offline threshold; it uses the +// prefix map of now, certainHeard the one at ingest. Reads tx only. +func certainFor(tx *StoreTx, pk string, pm *prefixMap) bool { + if tx == nil || pk == "" { + return false + } + if tx.DecodedJSON != "" && strings.Contains(tx.DecodedJSON, "ubKey") && !tx.signatureFailed() { + if d := tx.ParsedDecoded(); d != nil { + for _, f := range []string{"pubKey", "srcPubKey"} { + if v, _ := d[f].(string); v != "" && strings.EqualFold(v, pk) { + return true + } + } + } + } + return uniqueHopPrefix(pk, relayHopHexLen(tx), pm) +} + +// lastHeardValue is a node's last_heard in /api/nodes and the node detail: +// the newer of its DB last_seen (adverts; nil or unparsable counts as +// never) and heard, its certain last heard from LastHeardMap (zero for +// none). Before #2260 last_heard was a plain copy of last_seen. +func lastHeardValue(lastSeen interface{}, heard time.Time) interface{} { + if heard.IsZero() { + return lastSeen + } + if s, ok := lastSeen.(string); ok { + if t, ok := parseRelayTS(s); ok && !heard.After(t) { + return lastSeen + } + } + return heard.UTC().Format("2006-01-02T15:04:05.000Z") +} + +// longestSilence is the longest gap between consecutive hourly buckets +// (sorted RFC3339 keys) and where it started, including the silence from +// the start of the covered window (from) to the first bucket and the one +// still running from the last bucket to now: an outage in progress is +// exactly the silence worth seeing. No bucket at all is silence over the +// whole window. +func longestSilence(buckets []string, from, now time.Time) (int, interface{}) { + var longest int + var start interface{} + consider := func(begin string, to time.Time) { + t, err := time.Parse(time.RFC3339, begin) + if err != nil { + return + } + if gap := int(to.Sub(t).Milliseconds()); gap > longest { + longest, start = gap, begin + } + } + fromStr := from.UTC().Format(time.RFC3339) + if len(buckets) == 0 { + consider(fromStr, now) + return longest, start + } + if t, err := time.Parse(time.RFC3339, buckets[0]); err == nil && t.Sub(from) > time.Hour { + consider(fromStr, t) + } + for i := 1; i < len(buckets); i++ { + if t, err := time.Parse(time.RFC3339, buckets[i]); err == nil { + consider(buckets[i-1], t) + } + } + if len(buckets) > 0 { + // The last bucket covers its whole hour. + last := buckets[len(buckets)-1] + if t, err := time.Parse(time.RFC3339, last); err == nil && now.Sub(t) > time.Hour { + consider(last, now) + } + } + return longest, start +} diff --git a/cmd/server/certain_heard_loadpaths_test.go b/cmd/server/certain_heard_loadpaths_test.go new file mode 100644 index 000000000..7d49813ad --- /dev/null +++ b/cmd/server/certain_heard_loadpaths_test.go @@ -0,0 +1,122 @@ +package main + +import ( + "database/sql" + "strings" + "testing" + "time" +) + +// Review (#2261): every load path must feed certainHeard. A nil prefix map +// at any of the call sites in Load, loadChunk, scanAndMergeChunk, live +// ingest or late observations would leave the unit tests green; these go +// through the real paths and assert LastHeardMap. + +const lpRelay = "4f2a" + "5555555555555555555555555555555555555555555555555555555555555" + +// lpFixture is one flood transmission whose single 2-byte hop is lpRelay, +// a registered repeater, observed a minute ago. withResolved keeps the +// observation's resolved_path (the ingestor resolved the hop); otherwise it +// is NULL and the server re-resolves from path_json. hopInFirstObs false +// leaves the first observation without a path (for the late-observation +// case). +func lpFixture(t *testing.T, withResolved, hopInFirstObs bool) (*DB, *sql.DB, string) { + t.Helper() + path := createTestDBWithResolvedPath(t, 1, []string{lpRelay}) + w, err := sql.Open("sqlite3", path+"?_journal_mode=WAL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { w.Close() }) + at := time.Now().UTC().Add(-time.Minute).Format(time.RFC3339) + raw := "1541" + lpRelay[:4] + "00" // GRP_TXT, flood, path byte: 2-byte hashes x1 + pathJSON, resolved := `["`+strings.ToUpper(lpRelay[:4])+`"]`, interface{}(`["`+lpRelay+`"]`) + if !withResolved { + resolved = nil + } + if !hopInFirstObs { + pathJSON, resolved = `[]`, nil + } + for _, q := range []struct { + sql string + args []interface{} + }{ + {`ALTER TABLE transmissions ADD COLUMN from_pubkey TEXT`, nil}, + {`UPDATE transmissions SET raw_hex=?, route_type=1, payload_type=5, first_seen=?`, []interface{}{raw, at}}, + {`UPDATE observations SET raw_hex=?, timestamp=?, path_json=?, resolved_path=?`, []interface{}{raw, at, pathJSON, resolved}}, + {`INSERT INTO nodes (public_key, name, role) VALUES (?, 'Relay', 'repeater')`, []interface{}{lpRelay}}, + } { + if _, err := w.Exec(q.sql, q.args...); err != nil { + t.Fatalf("%s: %v", q.sql, err) + } + } + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { db.conn.Close() }) + return db, w, at +} + +func TestCertainHeardFedByEveryLoadPath(t *testing.T) { + type mode struct { + name string + withResolved, hopInFirstObs bool + run func(t *testing.T, s *PacketStore, db *DB, w *sql.DB, at string) + } + load := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) { + if err := s.Load(); err != nil { + t.Fatal(err) + } + } + chunk := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) { + if err := s.loadChunk(time.Now().Add(-time.Hour), time.Now()); err != nil { + t.Fatal(err) + } + } + chunked := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) { + if err := s.LoadChunked(1); err != nil { + t.Fatal(err) + } + } + live := func(t *testing.T, s *PacketStore, _ *DB, _ *sql.DB, _ string) { + s.IngestNewFromDB(0, 100) + } + late := func(t *testing.T, s *PacketStore, db *DB, w *sql.DB, at string) { + if err := s.Load(); err != nil { + t.Fatal(err) + } + if heard, _ := s.LastHeardMap([]string{lpRelay}); len(heard) != 0 { + t.Fatal("setup: the first observation carries no hop") + } + maxObs := db.GetMaxObservationID() + if _, err := w.Exec(`INSERT INTO observations (id, transmission_id, observer_id, observer_name, direction, snr, rssi, score, path_json, timestamp, raw_hex, resolved_path) + VALUES (2, 1, 'obs2', 'Obs2', 'RX', -9, -79, 5, ?, ?, '', NULL)`, `["`+strings.ToUpper(lpRelay[:4])+`"]`, at); err != nil { + t.Fatal(err) + } + s.IngestNewObservations(maxObs, 100) + } + for _, m := range []mode{ + {"Load/resolved_path", true, true, load}, + {"Load/path_json", false, true, load}, + {"loadChunk/resolved_path", true, true, chunk}, + {"loadChunk/path_json", false, true, chunk}, + {"LoadChunked/resolved_path", true, true, chunked}, + {"LoadChunked/path_json", false, true, chunked}, + {"IngestNewFromDB", false, true, live}, + {"IngestNewObservations", false, false, late}, + } { + t.Run(m.name, func(t *testing.T) { + db, w, at := lpFixture(t, m.withResolved, m.hopInFirstObs) + s := NewPacketStore(db, &PacketStoreConfig{}) + s.useResolvedPathIndex = true + s.initResolvedPathIndex() + m.run(t, s, db, w, at) + heard, _ := s.LastHeardMap([]string{lpRelay}) + want, _ := time.Parse(time.RFC3339, at) + if !heard[lpRelay].Equal(want) { + t.Fatalf("LastHeardMap[relay] = %v; want %v: this path does not feed certainHeard", heard[lpRelay], want) + } + }) + } +} diff --git a/cmd/server/certain_heard_test.go b/cmd/server/certain_heard_test.go new file mode 100644 index 000000000..3a6703092 --- /dev/null +++ b/cmd/server/certain_heard_test.go @@ -0,0 +1,408 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/meshcore-analyzer/users" +) + +// Three repeaters: two share the first pubkey byte 0x67 (as on the +// instance in #2260, where nine nodes did), one has a unique first byte. +var ( + chA = "67a7" + strings.Repeat("1", 60) // the silent repeater + chB = "67f4" + strings.Repeat("2", 60) // its 1-byte namesake, still relaying + chC = "40aa" + strings.Repeat("3", 60) // unique first byte +) + +func chPM() *prefixMap { + return buildPrefixMap([]nodeInfo{ + {PublicKey: chA, Role: "repeater"}, + {PublicKey: chB, Role: "repeater"}, + {PublicKey: chC, Role: "repeater"}, + }) +} + +// chTx builds a transmission whose raw header carries the route type and the +// path hash size (firmware docs/packet_format.md: header, transport codes for +// transport routes, then the path byte with the hash size in its top bits). +func chTx(id, routeType, payloadType, hashSize int, firstSeen time.Time) *StoreTx { + raw := fmt.Sprintf("%02x", payloadType<<2|routeType) + if routeType == 0 || routeType == 3 { + raw += "00000000" + } + raw += fmt.Sprintf("%02x", (hashSize-1)<<6|1) + strings.Repeat("ab", hashSize) + "00" + pt := payloadType + return &StoreTx{ID: id, RawHex: raw, PayloadType: &pt, FirstSeen: firstSeen.UTC().Format(time.RFC3339)} +} + +func TestRelayHopHexLen(t *testing.T) { + at := time.Date(2026, 10, 18, 16, 0, 0, 0, time.UTC) + cases := []struct { + name string + tx *StoreTx + want int + }{ + {"flood 1-byte", chTx(1, 1, 5, 1, at), 2}, + {"flood 2-byte", chTx(2, 1, 5, 2, at), 4}, + {"flood 3-byte", chTx(3, 1, 5, 3, at), 6}, + {"transport flood 2-byte", chTx(4, 0, 5, 2, at), 4}, + {"direct: path lists hops still to come", chTx(5, 2, 5, 1, at), 0}, + {"transport direct", chTx(6, 3, 5, 1, at), 0}, + {"TRACE: path bytes are SNR", chTx(7, 1, packetpathTrace, 1, at), 0}, + {"no raw", &StoreTx{ID: 8}, 0}, + } + for _, c := range cases { + if got := relayHopHexLen(c.tx); got != c.want { + t.Errorf("%s: %d; want %d", c.name, got, c.want) + } + } +} + +const packetpathTrace = 0x09 + +func chStore() *PacketStore { + s := makeTestStore(0, time.Now(), 0) + s.byPathHop = make(map[string][]*StoreTx) + s.initResolvedPathIndex() + return s +} + +func TestNoteCertainRelays(t *testing.T) { + pm := chPM() + t0 := time.Date(2026, 10, 18, 16, 49, 25, 0, time.UTC) + s := chStore() + hopsSeen := map[string]bool{} + + // 2-byte flood hop 67a7: only chA has it, certain. + s.indexResolvedPathHops(chTx(1, 1, 5, 2, t0), []string{chA}, hopsSeen, pm) + // Later 1-byte flood hops 67 (shared with chB) and 40 (unique): the + // resolver attributed 67 to chA, but only 40 is certain. + later := t0.Add(70 * time.Hour) + s.indexResolvedPathHops(chTx(2, 1, 5, 1, later), []string{chA, chC}, hopsSeen, pm) + // Direct route and TRACE never count; a nil prefix map records nothing. + s.indexResolvedPathHops(chTx(3, 2, 5, 2, later), []string{chA}, hopsSeen, pm) + s.indexResolvedPathHops(chTx(4, 1, packetpathTrace, 2, later), []string{chA}, hopsSeen, pm) + s.indexResolvedPathHops(chTx(5, 1, 5, 2, later), []string{chA}, hopsSeen, nil) + + if got := s.certainHeard[chA]; got != t0.Format(time.RFC3339) { + t.Errorf("chA certain = %q; want the 2-byte hop at %s", got, t0.Format(time.RFC3339)) + } + if got := s.certainHeard[chC]; got != later.Format(time.RFC3339) { + t.Errorf("chC certain = %q; want the unique 1-byte hop", got) + } + if len(s.byNode[chA]) != 5 { + t.Errorf("byNode[chA] = %d packets; attribution for analytics must be unchanged", len(s.byNode[chA])) + } + + // The cold-load path (resolved_path NULL) feeds liveness the same way. + s.indexColdLoadRelayHops(chTx(6, 1, 5, 2, later.Add(time.Hour)), []string{chB}, pm) + if got := s.certainHeard[chB]; got != later.Add(time.Hour).Format(time.RFC3339) { + t.Errorf("chB certain from cold load = %q", got) + } +} + +func TestIndexByNodeSenderIsCertain(t *testing.T) { + s := chStore() + at := time.Date(2026, 10, 18, 16, 45, 51, 0, time.UTC) + tx := chTx(1, 1, 4, 1, at) + tx.DecodedJSON = fmt.Sprintf(`{"pubKey":%q,"destPubKey":%q}`, chA, chC) + s.indexByNode(tx) + if s.certainHeard[chA] != at.Format(time.RFC3339) { + t.Errorf("sender not certain: %q", s.certainHeard[chA]) + } + if _, ok := s.certainHeard[chC]; ok { + t.Error("a destination is no evidence the node is alive") + } +} + +// #2260 regression: a repeater last certainly heard on 18 Oct, after which +// only its 1-byte namesake prefix shows up in paths. Liveness must stay at +// 18 Oct and the offline mail must go out once the 72 h window has passed. +func TestSilentRepeaterWithNamesakePrefixGoesOffline(t *testing.T) { + pm := chPM() + lastCertain := time.Date(2026, 10, 18, 16, 49, 25, 0, time.UTC) + s := chStore() + hopsSeen := map[string]bool{} + s.indexResolvedPathHops(chTx(1, 1, 5, 2, lastCertain), []string{chA}, hopsSeen, pm) + for i := 0; i < 20; i++ { + ts := lastCertain.Add(time.Duration(70+i) * time.Hour) + s.indexResolvedPathHops(chTx(2+i, 1, 5, 1, ts), []string{chA}, hopsSeen, pm) + } + + heard, _ := s.LastHeardMap([]string{chA}) + if !heard[chA].Equal(lastCertain) { + t.Fatalf("LastHeardMap = %v; want %v", heard[chA], lastCertain) + } + + in := evInput() + in.Now = lastCertain.Add(73 * time.Hour) + in.Watches = []users.NotifyWatch{{UserID: evUser, Pubkey: chA}} + in.Nodes[chA] = notifyNode{Pubkey: chA, Name: "Silent", Role: "repeater", LastSeen: lastCertain.Add(-4 * time.Minute).Format(time.RFC3339)} + in.Heard = heard + if got, _ := evState(evaluateNotifications(in), evUser, users.NotifyNodeOffline, chA); got != users.NotifyBad { + t.Fatalf("node.offline = %q; want bad", got) + } +} + +func TestCertainHeardEvictedWithNode(t *testing.T) { + now := time.Now().UTC() + s := makeTestStore(4, now.Add(-48*time.Hour), 0) + for i, tx := range s.packets { + s.noteCertainHeard(fmt.Sprintf("pk%04d", i), tx) // makeTestStore's sender keys + } + if len(s.certainHeard) != 4 { + t.Fatalf("certainHeard = %d entries; want 4", len(s.certainHeard)) + } + s.retentionHours = 24 + s.EvictStale() + if len(s.certainHeard) != 0 { + t.Fatalf("certainHeard = %v after evicting every packet; want empty", s.certainHeard) + } +} + +func TestLastHeardValue(t *testing.T) { + heard := time.Date(2026, 10, 22, 9, 0, 0, 0, time.UTC) + cases := []struct { + name string + lastSeen interface{} + heard time.Time + want interface{} + }{ + {"certain newer than advert", "2026-10-19T08:00:00Z", heard, "2026-10-22T09:00:00.000Z"}, + {"advert newer", "2026-10-22T10:00:00Z", heard, "2026-10-22T10:00:00Z"}, + {"nothing certain", "2026-10-19T08:00:00Z", time.Time{}, "2026-10-19T08:00:00Z"}, + {"never advertised", nil, heard, "2026-10-22T09:00:00.000Z"}, + {"nothing at all", nil, time.Time{}, nil}, + } + for _, c := range cases { + if got := lastHeardValue(c.lastSeen, c.heard); got != c.want { + t.Errorf("%s: %v; want %v", c.name, got, c.want) + } + } +} + +// /api/nodes and the node detail report the certain last heard, not a copy +// of last_seen, and the health endpoint's lastHeard is the same value. +func TestNodeEndpointsLastHeardIsCertain(t *testing.T) { + srv, router := setupTestServer(t) + const pk = "aabbccdd11223344" + heardAt := time.Now().UTC().Add(time.Minute).Truncate(time.Second) + srv.store.mu.Lock() + srv.store.certainHeard = map[string]string{pk: heardAt.Format(time.RFC3339)} + srv.store.mu.Unlock() + + get := func(url string, out interface{}) { + t.Helper() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", url, nil)) + if w.Code != 200 { + t.Fatalf("%s: %d", url, w.Code) + } + if err := json.Unmarshal(w.Body.Bytes(), out); err != nil { + t.Fatal(err) + } + } + type node struct { + PublicKey string `json:"public_key"` + LastSeen string `json:"last_seen"` + LastHeard string `json:"last_heard"` + } + want := heardAt.Format("2006-01-02T15:04:05.000Z") + + var list struct { + Nodes []node `json:"nodes"` + } + get("/api/nodes?limit=50", &list) + found := false + for _, n := range list.Nodes { + if n.PublicKey == pk { + found = true + if n.LastHeard != want || n.LastSeen == want { + t.Errorf("list: last_heard %q last_seen %q; want last_heard %q", n.LastHeard, n.LastSeen, want) + } + } else if n.LastHeard != n.LastSeen { + t.Errorf("list %s: nothing certain, last_heard %q must be last_seen %q", n.PublicKey, n.LastHeard, n.LastSeen) + } + } + if !found { + t.Fatalf("node %s not listed", pk) + } + + var detail struct { + Node node `json:"node"` + } + get("/api/nodes/"+pk, &detail) + if detail.Node.LastHeard != want { + t.Errorf("detail last_heard %q; want %q", detail.Node.LastHeard, want) + } + + var health struct { + Stats struct { + LastHeard string `json:"lastHeard"` + } `json:"stats"` + } + get("/api/nodes/"+pk+"/health", &health) + if health.Stats.LastHeard != heardAt.Format(time.RFC3339) { + t.Errorf("health lastHeard %q; want %q", health.Stats.LastHeard, heardAt.Format(time.RFC3339)) + } +} + +// noteCertainRelays runs per observation on ingest, under the write lock. +// An 8-hop 2-byte flood path against a 2,000-repeater prefix map: two hex +// bytes of header parsing plus one map lookup per hop. Quote in the PR. +func BenchmarkNoteCertainRelays(b *testing.B) { + nodes := make([]nodeInfo, 0, 2000) + for i := 0; i < 2000; i++ { + nodes = append(nodes, nodeInfo{PublicKey: fmt.Sprintf("%064x", i*7919), Role: "repeater"}) + } + pm := buildPrefixMap(nodes) + pks := make([]string, 0, 8) + for i := 0; i < 8; i++ { + pks = append(pks, nodes[i*250].PublicKey) + } + s := chStore() + tx := chTx(1, 1, 5, 2, time.Date(2026, 10, 18, 16, 0, 0, 0, time.UTC)) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.noteCertainRelays(tx, pks, pm) + } +} + +func TestIndexByNodeFailedSignatureIsNotCertain(t *testing.T) { + s := chStore() + at := time.Date(2026, 10, 18, 16, 45, 51, 0, time.UTC) + forged := chTx(1, 1, 4, 1, at) + forged.DecodedJSON = fmt.Sprintf(`{"pubKey":%q,"signatureValid":false}`, chA) + s.indexByNode(forged) + if _, ok := s.certainHeard[chA]; ok { + t.Fatal("an advert whose signature failed is no evidence") + } + if len(s.byNode[chA]) != 1 { + t.Fatal("it is still indexed for analytics") + } + valid := chTx(2, 1, 4, 1, at) + valid.DecodedJSON = fmt.Sprintf(`{"pubKey":%q,"signatureValid":true}`, chA) + s.indexByNode(valid) + if s.certainHeard[chA] != at.Format(time.RFC3339) { + t.Fatalf("a verified advert is evidence: %q", s.certainHeard[chA]) + } +} + +func TestNodeSearchLastHeardIsCertain(t *testing.T) { + srv, router := setupTestServer(t) + const pk = "aabbccdd11223344" + heardAt := time.Now().UTC().Add(time.Minute).Truncate(time.Second) + srv.store.mu.Lock() + srv.store.certainHeard = map[string]string{pk: heardAt.Format(time.RFC3339)} + srv.store.mu.Unlock() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/nodes/search?q=TestRepeater", nil)) + var body struct { + Nodes []struct { + PublicKey string `json:"public_key"` + LastHeard string `json:"last_heard"` + } `json:"nodes"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + for _, n := range body.Nodes { + if n.PublicKey == pk { + if want := heardAt.Format("2006-01-02T15:04:05.000Z"); n.LastHeard != want { + t.Fatalf("search last_heard %q; want %q", n.LastHeard, want) + } + return + } + } + t.Fatalf("node not found in %s", w.Body.String()) +} + +// Review (#2261): a repeater moved to inactive_nodes by retention may still +// relay, so its prefix keeps a hop ambiguous although the nodes table has +// only one candidate. +func TestRetiredNamesakeKeepsHopUncertain(t *testing.T) { + db := setupTestDB(t) + if _, err := db.conn.Exec(`CREATE TABLE IF NOT EXISTS inactive_nodes (public_key TEXT PRIMARY KEY, name TEXT, role TEXT)`); err != nil { + t.Fatal(err) + } + for _, q := range []string{ + `INSERT INTO inactive_nodes (public_key, role) VALUES ('40bb` + strings.Repeat("9", 60) + `', 'repeater')`, + `INSERT INTO inactive_nodes (public_key, role) VALUES ('6700` + strings.Repeat("8", 60) + `', 'companion')`, + } { + if _, err := db.conn.Exec(q); err != nil { + t.Fatal(err) + } + } + pm := chPM() + if err := pm.markRetired(db); err != nil { + t.Fatal(err) + } + if uniqueHopPrefix(chC, 2, pm) { + t.Error("1-byte hop 40 is shared with a retired repeater") + } + if !uniqueHopPrefix(chC, 4, pm) { + t.Error("2-byte hop 40aa is still only chC") + } + if _, ok := pm.retired["67"]; ok { + t.Error("a retired companion cannot relay; its prefix must not count") + } +} + +// Review (#2261): retention copies a row into inactive_nodes and nothing +// removes it when the node is heard again, so a node can be its own +// "retired namesake". That row must not take its certain relays away. +func TestReturnedNodeIsNotItsOwnRetiredNamesake(t *testing.T) { + db := setupTestDB(t) + if _, err := db.conn.Exec(`CREATE TABLE IF NOT EXISTS inactive_nodes (public_key TEXT PRIMARY KEY, name TEXT, role TEXT)`); err != nil { + t.Fatal(err) + } + for _, q := range []string{ + `INSERT INTO nodes (public_key, role) VALUES ('` + chC + `', 'repeater')`, + `INSERT INTO inactive_nodes (public_key, role) VALUES ('` + strings.ToUpper(chC) + `', 'repeater')`, + } { + if _, err := db.conn.Exec(q); err != nil { + t.Fatal(err) + } + } + pm := chPM() + if err := pm.markRetired(db); err != nil { + t.Fatal(err) + } + if !uniqueHopPrefix(chC, 4, pm) { + t.Error("2-byte hop 40aa is only chC; its own stale inactive_nodes row must not count") + } +} + +// The owner of an "uncertain" repeater (#2260) gets the offline mail, worded +// so it says why: nothing certain, only its shared prefix still in paths. +func TestOfflineMailExplainsUncertain(t *testing.T) { + last := evNow.Add(-80 * time.Hour) + in := evInput() + in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Name: "Silent", Role: "repeater", LastSeen: last.Format(time.RFC3339)} + in.Heard[evPkA] = last + in.SharedSeen = map[string]time.Time{evPkA: evNow.Add(-4 * time.Minute)} + in.States = []users.NotifyState{evStateRow(evUser, users.NotifyNodeOffline, evPkA, users.NotifyGood)} + ch := evaluateNotifications(in).Changes[evUser] + if len(ch) != 1 || ch[0].To != users.NotifyBad || !ch[0].SharedSeen.Equal(evNow.Add(-4*time.Minute)) { + t.Fatalf("changes = %+v; want one offline change carrying the shared-prefix time", ch) + } + txt := notifyChangeText(ch[0], "") + if !strings.Contains(txt, "offline (nothing that can only be this node") || !strings.Contains(txt, "possibly another node") { + t.Fatalf("mail line %q does not explain the uncertain state", txt) + } + + // A shared-prefix sighting older than the last certain packet adds nothing. + in.SharedSeen[evPkA] = last.Add(-time.Hour) + if ch := evaluateNotifications(in).Changes[evUser]; len(ch) != 1 || !ch[0].SharedSeen.IsZero() { + t.Fatalf("changes = %+v; want a plain offline change", ch) + } + if txt := notifyChangeText(evaluateNotifications(in).Changes[evUser][0], ""); !strings.HasSuffix(strings.SplitN(txt, ", ", 2)[0], "offline") { + t.Fatalf("plain offline line = %q", txt) + } +} diff --git a/cmd/server/chunked_load.go b/cmd/server/chunked_load.go index db9313a54..a37242c97 100644 --- a/cmd/server/chunked_load.go +++ b/cmd/server/chunked_load.go @@ -635,7 +635,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold rpStr := nullStrVal(resolvedPathStr) if rpStr != "" { pks := resolvedPubkeysFromJSON(rpStr) - s.indexResolvedPathHops(tx, pks, hopsSeen) + s.indexResolvedPathHops(tx, pks, hopsSeen, relayPM) } else if relayPM != nil && obsPJ != "" && obsPJ != "[]" { // resolved_path is NULL on live (since #1287 relay data is // persisted as neighbor_edges, not per-observation). Re-resolve @@ -647,9 +647,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold // resolved_path/path-hop indexes must NOT be populated here. // PR #1643 R1 munger #1: unique_prefix-only gate. rp := resolvePathForObsColdLoad(obsPJ, obsIDStr, tx, relayPM, coldLoadAmbiguousHopsSkipped) - for _, pk := range extractResolvedPubkeys(rp) { - s.addToByNode(tx, pk) - } + s.indexColdLoadRelayHops(tx, extractResolvedPubkeys(rp), relayPM) } tx.addObservation(obs) diff --git a/cmd/server/node_analytics_coverage_test.go b/cmd/server/node_analytics_coverage_test.go index 850edaba8..f0baf5c4c 100644 --- a/cmd/server/node_analytics_coverage_test.go +++ b/cmd/server/node_analytics_coverage_test.go @@ -8,6 +8,10 @@ import ( const coverageNodePK = "c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00c0ffee00" +// coverageAdvert is the node's own advert: certain evidence (#2260), so it +// counts toward availability. +const coverageAdvert = `{"pubKey":"` + coverageNodePK + `"}` + // setupCoverageStore holds one packet per hour for the node over the last // memHours hours, and records that window as loaded the way a bounded load // does (oldestLoaded): the in-memory window starts memHours ago (#748 M3a). @@ -27,6 +31,7 @@ func setupCoverageStore(t *testing.T, memHours int) *PacketStore { ID: memHours - h, Hash: fmt.Sprintf("cov%03d", h), PayloadType: &pt, + DecodedJSON: coverageAdvert, FirstSeen: now.Add(-time.Duration(h)*time.Hour - time.Minute).Format(time.RFC3339), } ps.packets = append(ps.packets, tx) @@ -50,7 +55,7 @@ func rate(p *float64) float64 { func addCoverageTx(ps *PacketStore, id int, firstSeen time.Time, front bool) { pt := 4 - tx := &StoreTx{ID: id, Hash: fmt.Sprintf("extra%d", id), PayloadType: &pt, FirstSeen: firstSeen.Format(time.RFC3339)} + tx := &StoreTx{ID: id, Hash: fmt.Sprintf("extra%d", id), PayloadType: &pt, DecodedJSON: coverageAdvert, FirstSeen: firstSeen.Format(time.RFC3339)} ps.mu.Lock() defer ps.mu.Unlock() if front { diff --git a/cmd/server/notify_eval.go b/cmd/server/notify_eval.go index d5c9aa4df..525e21bdc 100644 --- a/cmd/server/notify_eval.go +++ b/cmd/server/notify_eval.go @@ -33,17 +33,29 @@ type notifyObserver struct { // notifyInput is everything one evaluation reads. type notifyInput struct { - Now time.Time - Health HealthThresholds // resolved, as Config.GetHealthThresholds returns it - LowMv int // Config.LowBatteryMv - Accounts map[int64]users.User - Prefs []users.NotifyPrefs - Watches []users.NotifyWatch - States []users.NotifyState - Nodes map[string]notifyNode // watched nodes, plus every foreign node when an admin chose foreign.new - Heard map[string]time.Time // newest packet involving the node in the packet store (node page "Last Heard") - Relayed map[string]time.Time // last relay hop, used for repeaters and rooms only (#1598) - Observers []notifyObserver // nil unless an admin chose observer.offline + Now time.Time + Health HealthThresholds // resolved, as Config.GetHealthThresholds returns it + LowMv int // Config.LowBatteryMv + Accounts map[int64]users.User + Prefs []users.NotifyPrefs + Watches []users.NotifyWatch + States []users.NotifyState + Nodes map[string]notifyNode // watched nodes, plus every foreign node when an admin chose foreign.new + Heard map[string]time.Time // newest packet certainly from the node (node page "Last Heard", #2260) + // SharedSeen is, per repeater and room, when its hop prefix (shared with + // other repeaters, last_relayed) was last in a path. Not evidence: it + // only explains an offline mail for a node that is "uncertain" in the UI. + SharedSeen map[string]time.Time + // OfflineAfter is the learned offline threshold per node (#2260, + // adaptiveOffline); a node without one, or with one beyond its role's + // silent window, uses that window. + OfflineAfter map[string]time.Duration + // NewestPacket is the newest packet in the store (zero: unknown). A + // node's silence is measured up to it, not up to Now: when the feed + // stalls, every node's clock stops with it, so a node with a short + // learned window is not mailed offline before IngestStale trips (#2260). + NewestPacket time.Time + Observers []notifyObserver // nil unless an admin chose observer.offline // IngestStale: the newest packet in the store is old, so a silent node // and a silent feed look the same. node.offline and observer.offline // are then not compared (states unchanged, no changes). @@ -66,9 +78,12 @@ type notifyChange struct { Name string // display name at evaluation time To string // users.NotifyGood, users.NotifyBad or users.NotifyTold BatteryMv *int // node.battery only - Text string // node.external only: the alert text ("" when resolved) - URL string // node.external only: the alert's link, "" for the node page - At time.Time + // SharedSeen: node.offline only, when the node's shared hop prefix was + // still in a path after its last certain packet (zero otherwise). + SharedSeen time.Time + Text string // node.external only: the alert text ("" when resolved) + URL string // node.external only: the alert's link, "" for the node page + At time.Time } // notifyResult: Changes per user, ordered by event then name; States are @@ -95,13 +110,6 @@ var notifyEventRank = map[string]int{ users.NotifyForeignNew: 3, users.NotifyObserverOffline: 4, } -// notifyInfra reports the roles whose relay activity counts as heard -// (GetHealthMs and roles.js use the same two). -func notifyInfra(role string) bool { - r := strings.ToLower(role) - return r == "repeater" || r == "room" -} - // nodeLabel is the node's name made safe for a mail line (mailSafeText), // or the first 8 hex characters of its key. func nodeLabel(n notifyNode, known bool, pk string) string { @@ -115,11 +123,19 @@ func nodeLabel(n notifyNode, known bool, pk string) string { } // nodeOnlineState is good when the node was heard within its role's silent -// window, like roles.js getNodeStatus: the latest of last_seen, the packet -// store's last heard and, for repeaters and rooms, the last relay hop. A -// node missing from the analyzer database is bad. While prev is good, the +// window, like roles.js getNodeStatus: the latest of last_seen and the packet +// store's certain last heard (its own packets and relay hops that can only be +// this node, #2260). last_relayed is not evidence: it also counts 1-byte +// prefix matches shared with other repeaters, which kept a silent repeater +// "online" for days. The window is the role's silent window, or the node's +// learned OfflineAfter when that is shorter (adaptiveOffline). A node +// missing from the analyzer database is bad. While prev is good, the // latest evidence is never older than in.IngestResumedAt (the outage grace). -func nodeOnlineState(pk string, n notifyNode, known bool, prev string, in *notifyInput) string { +// While prev is bad (entered at prevAt), the node comes back only on +// evidence from that moment on: a window that grows back (a learned +// threshold lost to retention or a restart) must not read a still-silent +// node as back online. +func nodeOnlineState(pk string, n notifyNode, known bool, prev string, prevAt time.Time, in *notifyInput) string { if !known { return users.NotifyBad } @@ -127,16 +143,22 @@ func nodeOnlineState(pk string, n notifyNode, known bool, prev string, in *notif if t, ok := in.Heard[pk]; ok && t.After(last) { last = t } - if notifyInfra(n.Role) { - if t, ok := in.Relayed[pk]; ok && t.After(last) { - last = t - } - } if prev == users.NotifyGood && last.Before(in.IngestResumedAt) { last = in.IngestResumedAt } + if prev == users.NotifyBad && !prevAt.IsZero() && last.Before(prevAt) { + return users.NotifyBad + } + now := in.Now + if !in.NewestPacket.IsZero() && in.NewestPacket.Before(now) { + now = in.NewestPacket + } _, silentMs := in.Health.GetHealthMs(strings.ToLower(n.Role)) - if last.IsZero() || in.Now.Sub(last) >= time.Duration(silentMs)*time.Millisecond { + silent := time.Duration(silentMs) * time.Millisecond + if a, ok := in.OfflineAfter[pk]; ok && a > 0 && a < silent { + silent = a + } + if last.IsZero() || now.Sub(last) >= silent { return users.NotifyBad } return users.NotifyGood @@ -224,10 +246,12 @@ func evalExternal(res *notifyResult, prev map[users.NotifyKey]string, bad []stri // rows are dropped, like opting out, so a re-promotion starts silently. func evaluateNotifications(in notifyInput) notifyResult { prev := make(map[users.NotifyKey]string, len(in.States)) + prevAt := make(map[users.NotifyKey]time.Time, len(in.States)) // when each stored state was entered // extBad: per user and node, the node.external keys stored as bad. extBad := map[int64]map[string][]string{} for _, s := range in.States { prev[s.NotifyKey] = s.State + prevAt[s.NotifyKey] = s.ChangedAt if s.Event == users.NotifyNodeExternal && s.State == users.NotifyBad { if pk, key, ok := strings.Cut(s.Subject, "/"); ok && key != externalBaselineKey { if extBad[s.UserID] == nil { @@ -298,7 +322,14 @@ func evaluateNotifications(in notifyInput) notifyResult { name := nodeLabel(n, known, pk) if p.Has(users.NotifyNodeOffline) && !in.IngestStale { k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyNodeOffline, Subject: pk} - compare(k, name, nodeOnlineState(pk, n, known, prev[k], &in), nil) + state := nodeOnlineState(pk, n, known, prev[k], prevAt[k], &in) + before := len(res.Changes[p.UserID]) + compare(k, name, state, nil) + if state == users.NotifyBad && len(res.Changes[p.UserID]) > before { + if seen, ok := in.SharedSeen[pk]; ok && seen.After(in.Heard[pk]) { + res.Changes[p.UserID][before].SharedSeen = seen + } + } } if p.Has(users.NotifyNodeBattery) && known && n.BatteryMv != nil { k := users.NotifyKey{UserID: p.UserID, Event: users.NotifyNodeBattery, Subject: pk} diff --git a/cmd/server/notify_eval_test.go b/cmd/server/notify_eval_test.go index b3704e353..d73a3dbb8 100644 --- a/cmd/server/notify_eval_test.go +++ b/cmd/server/notify_eval_test.go @@ -41,7 +41,6 @@ func evInput() notifyInput { Watches: []users.NotifyWatch{{UserID: evUser, Pubkey: evPkA}}, Nodes: map[string]notifyNode{}, Heard: map[string]time.Time{}, - Relayed: map[string]time.Time{}, } } @@ -110,24 +109,29 @@ func TestEvaluateOfflineAndBack(t *testing.T) { } } -func TestEvaluateRelayAwareInfra(t *testing.T) { +// Liveness is last_seen or the store's certain last heard (#2260): a +// repeater whose advert is days old but that certainly relayed an hour ago is +// online; one with nothing certain inside the silent window is offline. There +// is no separate relay input any more: last_relayed also counts 1-byte prefix +// matches of other repeaters and kept a silent repeater online for days. +func TestEvaluateCertainHeard(t *testing.T) { cases := []struct { name, role, lastSeen string - relayed time.Duration + heard time.Duration want string }{ - {"repeater relaying", "repeater", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, - {"room relaying, role case", "Room", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, + {"repeater certainly relaying", "repeater", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, + {"room certainly relaying, role case", "Room", evAgo(80 * time.Hour), time.Hour, users.NotifyGood}, {"repeater silent exactly 72 h", "repeater", evAgo(72 * time.Hour), 0, users.NotifyBad}, {"repeater 71 h", "repeater", evAgo(71 * time.Hour), 0, users.NotifyGood}, - {"companion: relay time ignored", "companion", evAgo(30 * time.Hour), time.Hour, users.NotifyBad}, + {"repeater last certain packet 73 h ago", "repeater", evAgo(80 * time.Hour), 73 * time.Hour, users.NotifyBad}, {"no timestamp at all", "repeater", "", 0, users.NotifyBad}, } for _, c := range cases { in := evInput() in.Nodes[evPkA] = notifyNode{Pubkey: evPkA, Role: c.role, LastSeen: c.lastSeen} - if c.relayed > 0 { - in.Relayed[evPkA] = evNow.Add(-c.relayed) + if c.heard > 0 { + in.Heard[evPkA] = evNow.Add(-c.heard) } if got, _ := evState(evaluateNotifications(in), evUser, users.NotifyNodeOffline, evPkA); got != c.want { t.Errorf("%s: %q; want %q", c.name, got, c.want) @@ -306,7 +310,7 @@ func TestEvaluateChangesAreOrderedPerUser(t *testing.T) { // in the PR. func BenchmarkEvaluateNotifications(b *testing.B) { in := notifyInput{Now: evNow, Health: (&Config{}).GetHealthThresholds(), LowMv: 3300, - Accounts: map[int64]users.User{}, Nodes: map[string]notifyNode{}, Heard: map[string]time.Time{}, Relayed: map[string]time.Time{}} + Accounts: map[int64]users.User{}, Nodes: map[string]notifyNode{}, Heard: map[string]time.Time{}} for i := 0; i < 2000; i++ { pk := fmt.Sprintf("%064x", i) in.Nodes[pk] = notifyNode{Pubkey: pk, Role: "repeater", LastSeen: evAgo(time.Duration(i) * time.Minute), BatteryMv: evMv(3200 + i%300)} diff --git a/cmd/server/notify_service.go b/cmd/server/notify_service.go index 211de6203..0cc0fbc02 100644 --- a/cmd/server/notify_service.go +++ b/cmd/server/notify_service.go @@ -29,6 +29,8 @@ type notifier struct { // externalErr is the last tick's reason the externalAlerts feed was not // used ("" when it was), so each new reason is logged once. externalErr string + // adaptive caches the learned offline thresholds (offlineWindows). + adaptive map[string]adaptiveEntry } func newNotifier(a *authService, src notifySource, now func() time.Time) *notifier { @@ -190,12 +192,6 @@ func (n *notifier) tick(ctx context.Context) { log.Printf("[notify] read nodes: %v", err) return } - var infra []string - for _, pk := range pubkeys { - if nd, ok := nodes[pk]; ok && notifyInfra(nd.Role) { - infra = append(infra, pk) - } - } var observers []notifyObserver if wantObservers { if observers, err = n.src.observers(); err != nil { @@ -208,9 +204,15 @@ func (n *notifier) tick(ctx context.Context) { external = n.readExternal(now) } heard, lock := n.src.lastHeard(pubkeys) + var infra []string + for _, pk := range pubkeys { + if r := strings.ToLower(nodes[pk].Role); r == "repeater" || r == "room" { + infra = append(infra, pk) + } + } res := evaluateNotifications(notifyInput{Now: now, Health: n.src.health(), LowMv: n.src.lowBatteryMv(), Accounts: accounts, Prefs: prefs, Watches: watches, States: states, Nodes: nodes, - Heard: heard, Relayed: n.src.lastRelayed(infra), Observers: observers, IngestStale: stale, IngestResumedAt: n.ingestResumedAt, External: external}) + Heard: heard, SharedSeen: n.src.sharedPrefixSeen(infra), OfflineAfter: n.offlineWindows(pubkeys, now), NewestPacket: n.src.newestPacket(), Observers: observers, IngestStale: stale, IngestResumedAt: n.ingestResumedAt, External: external}) if err := st.DeleteNotifyStates(res.Drop); err != nil { log.Printf("[notify] delete dropped states, nothing mailed this time: %v", err) return @@ -369,6 +371,11 @@ func notifyChangeText(c notifyChange, label string) string { what = "back online" if bad { what = "offline" + if !c.SharedSeen.IsZero() { + // #2260: the UI shows such a node as "uncertain". + what += " (nothing that can only be this node; its hop prefix, shared with other repeaters, was last in a path " + + c.SharedSeen.UTC().Format("2006-01-02 15:04 UTC") + ", possibly another node)" + } } case users.NotifyNodeBattery: what = "battery recovered" diff --git a/cmd/server/notify_service_test.go b/cmd/server/notify_service_test.go index 884876c88..07341bedd 100644 --- a/cmd/server/notify_service_test.go +++ b/cmd/server/notify_service_test.go @@ -20,7 +20,9 @@ type fakeNotifySource struct { low int nodeMap map[string]notifyNode heard map[string]time.Time - relayed map[string]time.Time + shared map[string]time.Time + acts map[string]certainActivity + actsAsked [][]string obs []notifyObserver newestAt func() time.Time // newest packet in the store; nil means none lock time.Duration // what lastHeard reports as its lock time @@ -57,11 +59,15 @@ func (f *fakeNotifySource) nodes(pubkeys []string, withForeign bool) (map[string } return out, nil } +func (f *fakeNotifySource) sharedPrefixSeen([]string) map[string]time.Time { return f.shared } +func (f *fakeNotifySource) certainActivity(pks []string, _ time.Time) map[string]certainActivity { + f.actsAsked = append(f.actsAsked, append([]string(nil), pks...)) + return f.acts +} func (f *fakeNotifySource) lastHeard(pks []string) (map[string]time.Time, time.Duration) { f.heardAsked = append([]string(nil), pks...) return f.heard, f.lock } -func (f *fakeNotifySource) lastRelayed([]string) map[string]time.Time { return f.relayed } func (f *fakeNotifySource) newestPacket() time.Time { if f.newestAt == nil { return time.Time{} @@ -103,7 +109,7 @@ func newNotifyFixture(t *testing.T, ns notifySettings) *notifyFixture { clk := ¬ifyClock{t: time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)} a.st.SetClock(clk.Now) src := &fakeNotifySource{isReady: true, hs: (&Config{}).GetHealthThresholds(), low: 3300, - nodeMap: map[string]notifyNode{}, heard: map[string]time.Time{}, relayed: map[string]time.Time{}, newestAt: clk.Now} + nodeMap: map[string]notifyNode{}, heard: map[string]time.Time{}, newestAt: clk.Now} if ns.enabled { a.notify = newNotifier(a, src, clk.Now) } diff --git a/cmd/server/notify_source.go b/cmd/server/notify_source.go index f95f8011c..89d712522 100644 --- a/cmd/server/notify_source.go +++ b/cmd/server/notify_source.go @@ -15,7 +15,8 @@ type notifySource interface { lowBatteryMv() int nodes(pubkeys []string, withForeign bool) (map[string]notifyNode, error) lastHeard(pubkeys []string) (map[string]time.Time, time.Duration) // and how long the store lock was held - lastRelayed(pubkeys []string) map[string]time.Time + sharedPrefixSeen(pubkeys []string) map[string]time.Time // last_relayed, for the mail text only + certainActivity(pubkeys []string, now time.Time) map[string]certainActivity observers() ([]notifyObserver, error) newestPacket() time.Time // zero when the store holds no packet externalFeed() ([]byte, error) // the externalAlerts feed body (#2177); only called when configured @@ -53,13 +54,30 @@ func (src serverNotifySource) lastHeard(pubkeys []string) (map[string]time.Time, return src.s.store.LastHeardMap(pubkeys) } -// lastRelayed reads the cached relay metrics (the map /api/nodes uses; the -// background recomputer refreshes it). Pass repeaters and rooms only. -func (src serverNotifySource) lastRelayed(pubkeys []string) map[string]time.Time { +// sharedPrefixSeen reads last_relayed from the cached relay metrics (the map +// /api/nodes uses). It counts 1-byte prefixes shared with other repeaters, +// so it only words the offline mail, never decides it (#2260). +func (src serverNotifySource) sharedPrefixSeen(pubkeys []string) map[string]time.Time { + out := map[string]time.Time{} if src.s.store == nil || len(pubkeys) == 0 { - return map[string]time.Time{} + return out } - return relayTimes(src.s.store.GetRepeaterRelayInfoMap(src.s.cfg.GetHealthThresholds().RelayActiveHours), pubkeys) + m := src.s.store.GetRepeaterRelayInfoMap(src.s.cfg.GetHealthThresholds().RelayActiveHours) + for _, pk := range pubkeys { + if info, ok := lookupRelayInfo(m, pk); ok { + if t, ok := parseRelayTS(info.LastRelayed); ok { + out[pk] = t + } + } + } + return out +} + +func (src serverNotifySource) certainActivity(pubkeys []string, now time.Time) map[string]certainActivity { + if src.s.store == nil { + return map[string]certainActivity{} + } + return src.s.store.CertainActivity(pubkeys, now) } func (src serverNotifySource) newestPacket() time.Time { @@ -83,22 +101,6 @@ func (src serverNotifySource) observers() ([]notifyObserver, error) { return notifyObserversFrom(list), nil } -// relayTimes parses LastRelayed for the given keys; keys without a relay -// time are absent. -func relayTimes(m map[string]RepeaterRelayInfo, pubkeys []string) map[string]time.Time { - out := make(map[string]time.Time, len(pubkeys)) - for _, pk := range pubkeys { - info, ok := lookupRelayInfo(m, pk) - if !ok || info.LastRelayed == "" { - continue - } - if t, ok := parseRelayTS(info.LastRelayed); ok { - out[pk] = t - } - } - return out -} - func notifyObserversFrom(list []Observer) []notifyObserver { out := make([]notifyObserver, 0, len(list)) for _, o := range list { @@ -171,20 +173,18 @@ func (db *DB) NotifyNodes(pubkeys []string, withForeign bool) (map[string]notify return out, nil } -// LastHeardMap returns, per pubkey with packets in the store, the newest -// FirstSeen among them: the node page's "Last Heard" (GetNodeHealth) in -// bulk, and how long the read lock was held (logged once per tick). One -// read lock, O(packets of the given nodes), string compares only; parsing +// LastHeardMap returns, per pubkey, the newest FirstSeen of a transmission +// certainly from that node (certainHeard, #2260): the node page's "Last +// Heard" (GetNodeHealth) in bulk, and how long the read lock was held +// (logged once per tick). One read lock, one map lookup per pubkey; parsing // happens after the lock is released. func (s *PacketStore) LastHeardMap(pubkeys []string) (map[string]time.Time, time.Duration) { latest := make(map[string]string, len(pubkeys)) s.mu.RLock() locked := time.Now() for _, pk := range pubkeys { - for _, tx := range s.byNode[pk] { - if tx != nil && tx.FirstSeen > latest[pk] { - latest[pk] = tx.FirstSeen - } + if ts := s.certainHeard[pk]; ts != "" { + latest[pk] = ts } } held := time.Since(locked) diff --git a/cmd/server/notify_source_test.go b/cmd/server/notify_source_test.go index 1d5f025a5..3c62f421d 100644 --- a/cmd/server/notify_source_test.go +++ b/cmd/server/notify_source_test.go @@ -56,9 +56,9 @@ func TestDBNotifyNodesChunksLongLists(t *testing.T) { } func TestPacketStoreLastHeardMap(t *testing.T) { - s := &PacketStore{byNode: map[string][]*StoreTx{ - evPkA: {{FirstSeen: "2026-10-07T10:00:00Z"}, {FirstSeen: "2026-10-07T11:30:00.000Z"}, nil, {FirstSeen: "2026-10-07T09:00:00Z"}}, - evPkB: {{FirstSeen: "not a time"}}, + s := &PacketStore{certainHeard: map[string]string{ + evPkA: "2026-10-07T11:30:00.000Z", + evPkB: "not a time", }} got, lock := s.LastHeardMap([]string{evPkA, evPkB, evPkF}) if len(got) != 1 || !got[evPkA].Equal(time.Date(2026, 10, 7, 11, 30, 0, 0, time.UTC)) { @@ -69,17 +69,15 @@ func TestPacketStoreLastHeardMap(t *testing.T) { } } -// 5,000 watched nodes with 20 packets each, the spec's upper figure. Quote -// the result in the PR: this runs under the store's read lock. +// 5,000 watched nodes, the spec's upper figure. Quote the result in the PR: +// this runs under the store's read lock (one map lookup per node, #2260). func BenchmarkLastHeardMap(b *testing.B) { - s := &PacketStore{byNode: map[string][]*StoreTx{}} + s := &PacketStore{certainHeard: map[string]string{}} keys := make([]string, 0, 5000) for i := 0; i < 5000; i++ { pk := fmt.Sprintf("%064x", i) keys = append(keys, pk) - for j := 0; j < 20; j++ { - s.byNode[pk] = append(s.byNode[pk], &StoreTx{FirstSeen: evNow.Add(-time.Duration(j) * time.Minute).Format(time.RFC3339)}) - } + s.certainHeard[pk] = evNow.Add(-time.Duration(i) * time.Minute).Format(time.RFC3339) } b.ResetTimer() for i := 0; i < b.N; i++ { @@ -87,17 +85,6 @@ func BenchmarkLastHeardMap(b *testing.B) { } } -func TestRelayTimes(t *testing.T) { - m := map[string]RepeaterRelayInfo{ - evPkA: {LastRelayed: "2026-10-07T11:00:00Z"}, - evPkB: {}, - } - got := relayTimes(m, []string{evPkA, evPkB, evPkF}) - if len(got) != 1 || !got[evPkA].Equal(time.Date(2026, 10, 7, 11, 0, 0, 0, time.UTC)) { - t.Fatalf("relayTimes = %v", got) - } -} - func TestNotifyObserversFrom(t *testing.T) { name, seen := "Roof", "2026-10-07T11:00:00Z" got := notifyObserversFrom([]Observer{{ID: "OBS1", Name: &name, LastSeen: &seen}, {ID: "OBS2"}}) @@ -108,7 +95,7 @@ func TestNotifyObserversFrom(t *testing.T) { func TestServerNotifySourceWithoutStore(t *testing.T) { src := serverNotifySource{s: &Server{cfg: &Config{}}} - if heard, lock := src.lastHeard([]string{evPkA}); len(heard) != 0 || lock != 0 || len(src.lastRelayed([]string{evPkA})) != 0 { + if heard, lock := src.lastHeard([]string{evPkA}); len(heard) != 0 || lock != 0 { t.Fatal("a server without a packet store reported times") } if src.lowBatteryMv() != 3300 || src.health().InfraSilentHours != 72 { diff --git a/cmd/server/pathhop_dedupe_2108_bench_test.go b/cmd/server/pathhop_dedupe_2108_bench_test.go index bd53037aa..bbffc5513 100644 --- a/cmd/server/pathhop_dedupe_2108_bench_test.go +++ b/cmd/server/pathhop_dedupe_2108_bench_test.go @@ -39,7 +39,7 @@ func phdBenchStore(n int) (store *PacketStore, resolve func(j int) []string) { store.byTxID[tx.ID] = tx store.byPayloadType[pt] = append(store.byPayloadType[pt], tx) addTxToPathHopIndex(store.byPathHop, tx) - store.indexResolvedPathHops(tx, resolve(i), hopsSeen) + store.indexResolvedPathHops(tx, resolve(i), hopsSeen, nil) } return store, resolve } @@ -78,7 +78,7 @@ func BenchmarkLateObservationIndex_2108(b *testing.B) { b.StartTimer() } store.mu.Lock() - store.indexResolvedPathHops(store.packets[j], pks[i%batch], hopsSeen) + store.indexResolvedPathHops(store.packets[j], pks[i%batch], hopsSeen, nil) store.mu.Unlock() } b.StopTimer() @@ -142,7 +142,7 @@ func BenchmarkTrafficShareScoreMap_2108(b *testing.B) { hopsSeen := map[string]bool{} for o := 1; o < c.obs; o++ { for j, tx := range store.packets { - store.indexResolvedPathHops(tx, resolve(j), hopsSeen) + store.indexResolvedPathHops(tx, resolve(j), hopsSeen, nil) } } if c.reversed { diff --git a/cmd/server/pathhop_eviction_1908_test.go b/cmd/server/pathhop_eviction_1908_test.go index a10bfd439..5a1060ce4 100644 --- a/cmd/server/pathhop_eviction_1908_test.go +++ b/cmd/server/pathhop_eviction_1908_test.go @@ -42,7 +42,7 @@ func TestPathHopEviction1908(t *testing.T) { tx.ScopeName = &expiredScope } addTxToPathHopIndex(store.byPathHop, tx) - store.indexResolvedPathHops(tx, []string{shared, pk}, hopsSeen) + store.indexResolvedPathHops(tx, []string{shared, pk}, hopsSeen, nil) } // Repeated observations no longer append the same resolved // association again (#2108), but eviction must still drop every diff --git a/cmd/server/pathhop_record_2108_test.go b/cmd/server/pathhop_record_2108_test.go index b10e549f7..5e5049b5b 100644 --- a/cmd/server/pathhop_record_2108_test.go +++ b/cmd/server/pathhop_record_2108_test.go @@ -77,7 +77,7 @@ func phdRecordStore(count int) (store *PacketStore, old, young []*StoreTx) { young = append(young, tx) } addTxToPathHopIndex(store.byPathHop, tx) - store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen, nil) } return store, old, young } @@ -107,7 +107,7 @@ func TestPathHopResolvedRecordBoundedAndEvicted_2108(t *testing.T) { hopsSeen := map[string]bool{} for _, tx := range store.packets { for k := 0; k < 10; k++ { - store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen, nil) } } if got := len(store.pathHopResolved); got != len(old)+len(young) { @@ -141,7 +141,7 @@ func TestPathHopResolvedRecordBoundedAndEvicted_2108(t *testing.T) { } // A survivor's next observation still adds nothing. for _, tx := range young { - store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen, nil) if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { t.Fatalf("tx %d has %d entries after eviction and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) } @@ -167,7 +167,7 @@ func TestPathHopResolvedRecordAcrossRebuild_2108(t *testing.T) { store, _, young := phdRecordStore(40) gone := store.packets[0] goneOnly := strings.Repeat("e", 64) // a relay only the removed transmission went through - store.indexResolvedPathHops(gone, []string{goneOnly}, map[string]bool{}) + store.indexResolvedPathHops(gone, []string{goneOnly}, map[string]bool{}, nil) store.packets = store.packets[1:] delete(store.byTxID, gone.ID) delete(store.byHash, gone.Hash) @@ -190,7 +190,7 @@ func TestPathHopResolvedRecordAcrossRebuild_2108(t *testing.T) { if _, ok := store.pathHopResolved[tx]; !ok { t.Fatalf("rebuild dropped the record of live tx %d", tx.ID) } - store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen, nil) if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { t.Fatalf("tx %d has %d entries after a rebuild and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) } @@ -216,7 +216,7 @@ func TestPathHopResolvedRecordClearedWithEmptyIndex_2108(t *testing.T) { } hopsSeen := map[string]bool{} for _, tx := range young { - store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen) + store.indexResolvedPathHops(tx, phdFreshKeys(), hopsSeen, nil) if n := phdEntriesOf(store, tx); n != phdRecordRefsPerTx { t.Fatalf("tx %d has %d entries after the rebuild and another observation, want %d", tx.ID, n, phdRecordRefsPerTx) } @@ -278,7 +278,7 @@ func BenchmarkPathHopRecordMemory_2108(b *testing.B) { r := (i*31 + j*997) % relays pks[j] = fmt.Sprintf("%02x%062x", r%256, r) } - store.indexResolvedPathHops(tx, pks, hopsSeen) + store.indexResolvedPathHops(tx, pks, hopsSeen, nil) } } with := heap() diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 13bd5016d..c17a871fe 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -1634,8 +1634,18 @@ func (s *Server) buildNodesResponse(r *http.Request) (NodeListResponse, error) { maxUseful := maxFloat(usefulMap) // #1879: which rule places each node for ?region=. Atomic snapshot. placements := s.store.GetNodeRegionPlacements() + // #2260: last_heard is the newer of last_seen and the certain last + // heard. One read lock and a map lookup per node. + pageKeys := make([]string, 0, len(nodes)) for _, node := range nodes { if pk, ok := node["public_key"].(string); ok { + pageKeys = append(pageKeys, pk) + } + } + heard, _ := s.store.LastHeardMap(pageKeys) + for _, node := range nodes { + if pk, ok := node["public_key"].(string); ok { + node["last_heard"] = lastHeardValue(node["last_seen"], heard[pk]) EnrichNodeWithHashSize(node, hashInfo[pk]) mbEntry, _ := s.store.GetMultibyteCapFor(pk) EnrichNodeWithMultiByte(node, mbEntry) @@ -1771,6 +1781,21 @@ func (s *Server) handleNodeSearch(w http.ResponseWriter, r *http.Request) { writeError(w, 500, err.Error()) return } + // #2260: same last_heard as /api/nodes, not a copy of last_seen. + if s.store != nil && len(nodes) > 0 { + keys := make([]string, 0, len(nodes)) + for _, node := range nodes { + if pk, ok := node["public_key"].(string); ok { + keys = append(keys, pk) + } + } + heard, _ := s.store.LastHeardMap(keys) + for _, node := range nodes { + if pk, ok := node["public_key"].(string); ok { + node["last_heard"] = lastHeardValue(node["last_seen"], heard[pk]) + } + } + } // Filter blacklisted nodes from search results if len(s.cfg.NodeBlacklist) > 0 { filtered := make([]map[string]interface{}, 0, len(nodes)) @@ -1847,6 +1872,8 @@ func (s *Server) handleNodeDetail(w http.ResponseWriter, r *http.Request) { mbEntry, _ := s.store.GetMultibyteCapFor(pubkey) EnrichNodeWithMultiByte(node, mbEntry) node["region_placement"] = RegionPlacementFor(s.store.GetNodeRegionPlacements(), pubkey) + heard, _ := s.store.LastHeardMap([]string{pubkey}) + node["last_heard"] = lastHeardValue(node["last_seen"], heard[pubkey]) if role, _ := node["role"].(string); role == "repeater" || role == "room" { ht := s.cfg.GetHealthThresholds() info := s.store.GetRepeaterRelayInfo(pubkey, ht.RelayActiveHours) diff --git a/cmd/server/routes_test.go b/cmd/server/routes_test.go index bac94ed0a..b33b92224 100644 --- a/cmd/server/routes_test.go +++ b/cmd/server/routes_test.go @@ -820,21 +820,15 @@ func TestNodeHealthNotFound(t *testing.T) { func TestNodeHealthPartialFromPackets(t *testing.T) { srv, router := setupTestServer(t) - // Inject a packet into byNode for a pubkey that doesn't exist in the nodes table + // Index the companion's own packet (srcPubKey) for a pubkey that doesn't + // exist in the nodes table. Going through indexByNode also records it as + // certain evidence, which is what stats.lastHeard reports (#2260). ghostPubkey := "ghost_companion_no_advert" now := time.Now().UTC().Format(time.RFC3339) snr := 5.0 srv.store.mu.Lock() - if srv.store.byNode == nil { - srv.store.byNode = make(map[string][]*StoreTx) - } - if srv.store.nodeHashes == nil { - srv.store.nodeHashes = newNodeSets() - } - srv.store.byNode[ghostPubkey] = []*StoreTx{ - {Hash: "abc123", FirstSeen: now, SNR: &snr, ObservationCount: 1}, - } - srv.store.nodeHashes[ghostPubkey] = map[int]struct{}{0: {}} + srv.store.indexByNode(&StoreTx{ID: 999999, Hash: "abc123", FirstSeen: now, SNR: &snr, ObservationCount: 1, + DecodedJSON: `{"srcPubKey":"` + ghostPubkey + `"}`}) srv.store.mu.Unlock() req := httptest.NewRequest("GET", "/api/nodes/"+ghostPubkey+"/health", nil) diff --git a/cmd/server/store.go b/cmd/server/store.go index 80cb8156e..81ed86e77 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -157,10 +157,11 @@ func parseObsTimestamp(s string) (time.Time, bool) { // decodedCacheDroppedKeys are payload fields no server code reads: the // ciphertext and MAC of encrypted payloads, advert signatures, the ISO copy -// of a timestamp, raw path data and control-packet details. +// of a timestamp, raw path data and control-packet details. signatureValid +// stays: liveness ignores adverts whose signature failed (#2260). // TestDecodedCacheDroppedKeysAreUnread fails if code starts naming one. var decodedCacheDroppedKeys = []string{ - "encryptedData", "mac", "signature", "signatureValid", "timestampISO", "pathData", + "encryptedData", "mac", "signature", "timestampISO", "pathData", "ctrlPubKey", "ctrlSubtype", "ctrlTag", "ctrlFlags", "ctrlZeroHop", "ctrlLength", } @@ -256,6 +257,7 @@ type PacketStore struct { byObserver map[string][]*StoreObs // observer_id → observations byNode map[string][]*StoreTx // pubkey → transmissions nodeHashes map[string]map[int]struct{} // pubkey → set of tx IDs listed in byNode[pubkey] + certainHeard map[string]string // pubkey → newest FirstSeen certainly from that node (certain_heard.go) byPathHop map[string][]*StoreTx // lowercase hop/pubkey → transmissions with that hop in path byPayloadType map[int][]*StoreTx // payload_type → transmissions loaded bool @@ -1095,7 +1097,7 @@ func (s *PacketStore) Load() error { if rpStr != "" { pks := resolvedPubkeysFromJSON(rpStr) // Single point of truth — see indexResolvedPathHops doc + #1558. - s.indexResolvedPathHops(tx, pks, hopsSeen) + s.indexResolvedPathHops(tx, pks, hopsSeen, relayPM) } else if relayPM != nil && obsPJ != "" && obsPJ != "[]" { // resolved_path not persisted — reconstruct relay hops from // path_json so relay-node analytics history survives a restart. @@ -1108,9 +1110,7 @@ func (s *PacketStore) Load() error { // PR #1643 R1 munger #1: gate on unique_prefix only — ambiguous // hops are silently dropped (skipped counter logged at end). rp := resolvePathForObsColdLoad(obsPJ, obsIDStr, tx, relayPM, &coldLoadAmbiguousHopsSkipped) - for _, pk := range extractResolvedPubkeys(rp) { - s.addToByNode(tx, pk) - } + s.indexColdLoadRelayHops(tx, extractResolvedPubkeys(rp), relayPM) } tx.addObservation(obs) @@ -1587,14 +1587,12 @@ func (s *PacketStore) loadChunk(from, to time.Time) error { // stats collapse after restart for relay-heavy nodes. The // pubkey union was pre-built outside the lock above. if pks := localResolvedPKsByTx[tx.ID]; len(pks) > 0 { - s.indexResolvedPathHops(tx, pks, hopsSeen) + s.indexResolvedPathHops(tx, pks, hopsSeen, relayPM) } // path_json fallback (resolved_path NULL): byNode ONLY, so relay // nodes keep their analytics history across a restart without // polluting the resolved_path/path-hop indexes (#1352). - for _, pk := range localByNodePKsByTx[tx.ID] { - s.addToByNode(tx, pk) - } + s.indexColdLoadRelayHops(tx, localByNodePKsByTx[tx.ID], relayPM) } s.mu.Unlock() runtime.Gosched() @@ -1975,17 +1973,32 @@ func skipPlainJSONString(s string, i int) (int, bool) { // idempotent per (transmission, pubkey) — byNode via nodeHashes, the // resolved index via its forward-list check, byPathHop via // pathHopResolved (#2108). -func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen map[string]bool) { +// - pm decides which hops are certain liveness evidence (#2260, +// noteCertainRelays); nil records none. +func (s *PacketStore) indexResolvedPathHops(tx *StoreTx, pks []string, hopsSeen map[string]bool, pm *prefixMap) { if len(pks) == 0 { return } for _, pk := range pks { s.addToByNode(tx, pk) } + s.noteCertainRelays(tx, pks, pm) s.addResolvedPubkeysToPathHopIndex(tx, pks, hopsSeen) s.addToResolvedPubkeyIndex(tx.ID, pks) } +// indexColdLoadRelayHops is the byNode-only counterpart of +// indexResolvedPathHops for relay hops re-resolved from path_json at cold +// load (resolved_path NULL, #1352): the path-hop and resolved indexes stay +// untouched, liveness still learns from the certain hops. Caller holds s.mu +// for writing. +func (s *PacketStore) indexColdLoadRelayHops(tx *StoreTx, pks []string, pm *prefixMap) { + for _, pk := range pks { + s.addToByNode(tx, pk) + } + s.noteCertainRelays(tx, pks, pm) +} + // indexByNode extracts pubkeys from decoded_json and indexes the transmission. // indexByNode indexes a transmission under all pubkeys found in its decoded // JSON. Resolved path pubkeys are handled separately via the decode-window. @@ -2004,6 +2017,12 @@ func (s *PacketStore) indexByNode(tx *StoreTx) bool { foundNew = true } indexed[v] = true + // The sender's own packet is certain evidence (#2260); + // a destination is not, the node need not be alive, and + // neither is an advert whose signature failed. + if field != "destPubKey" && !tx.signatureFailed() { + s.noteCertainHeard(v, tx) + } } } } @@ -3210,7 +3229,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac rpForBroadcast = resolvePathForObs(r.pathJSON, r.observerID, tx, cachedPM, cachedGraph) resolvedPubkeys = extractResolvedPubkeys(rpForBroadcast) // Single point of truth — see indexResolvedPathHops doc + #1558. - s.indexResolvedPathHops(tx, resolvedPubkeys, hopsSeen) + s.indexResolvedPathHops(tx, resolvedPubkeys, hopsSeen, cachedPM) } // Stash rpForBroadcast for later broadcast/persist (keyed by obs ID) if rpForBroadcast != nil { @@ -3524,7 +3543,7 @@ func (s *PacketStore) IngestNewObservations(sinceObsID, limit int) []map[string] obsResolvedPath = resolvePathForObs(r.pathJSON, r.observerID, tx, pm, graphRef) pks := extractResolvedPubkeys(obsResolvedPath) // Single point of truth — see indexResolvedPathHops doc + #1558. - s.indexResolvedPathHops(tx, pks, hopsSeen) + s.indexResolvedPathHops(tx, pks, hopsSeen, pm) } } // Stash for broadcast/persist @@ -5438,6 +5457,7 @@ func (s *PacketStore) evictStaleInternal(rpBatch map[int][]string) int { } if len(filtered) == 0 { delete(s.byNode, nodeKey) + delete(s.certainHeard, nodeKey) } else { s.byNode[nodeKey] = filtered } @@ -7293,6 +7313,11 @@ type prefixMap struct { // candidates without them; markNonRelay builds it so relayCandidates, // called per hop of every observation resolved, never copies. relay map[string][]nodeInfo + // retired holds the prefixes of path-capable nodes that retention moved + // to inactive_nodes. They are no resolver candidates, but such a + // repeater may still relay, so a hop on its prefix is not certain + // evidence for anyone (#2260, uniqueHopPrefix). + retired map[string]struct{} } // markNonRelay registers a set of lowercase pubkeys as listener-only. @@ -7415,6 +7440,9 @@ func (s *PacketStore) rebuildNodeCache() ([]nodeInfo, *prefixMap) { } else { s.logSchemaDegradationOnce("observers.can_relay read failed; path-hop disambiguator will not filter listener-only observers: " + err.Error()) } + if err := pm.markRetired(s.db); err != nil { + s.logSchemaDegradationOnce("inactive_nodes read failed; certain liveness will not see retired namesake prefixes: " + err.Error()) + } } s.cacheMu.Lock() @@ -9984,7 +10012,9 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string var packetsToday int var snrSum float64 var snrCount int - var lastHeard string + // Certain evidence only (#2260): packets merely attributed to the node + // (an ambiguous 1-byte hop) still count in the stats below. + lastHeard := s.certainHeard[n.pk] // See GetNodeHealth: this set is "saw traffic involving the node", // which is not "heard the node". Only the direct-RF rows below carry // signal numbers. @@ -10003,9 +10033,6 @@ func (s *PacketStore) GetBulkHealth(limit int, region, area string) []map[string snrSum += *pkt.SNR snrCount++ } - if lastHeard == "" || pkt.FirstSeen > lastHeard { - lastHeard = pkt.FirstSeen - } if pkt.ObserverID != "" { relayObservers[pkt.ObserverID] = struct{}{} } @@ -10086,7 +10113,8 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro var snrSum float64 var snrCount int var totalHops, hopCount int - var lastHeard string + // Certain evidence only (#2260): see GetBulkHealth. + lastHeard := s.certainHeard[pubkey] totalObservations := 0 // Observers that saw traffic involving this node — as originator, as a @@ -10105,9 +10133,6 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro snrSum += *pkt.SNR snrCount++ } - if lastHeard == "" || pkt.FirstSeen > lastHeard { - lastHeard = pkt.FirstSeen - } // Hop counting hops := txGetParsedPath(pkt) if len(hops) > 0 { @@ -10265,6 +10290,8 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR fromTime := now.Add(-time.Duration(days) * 24 * time.Hour) fromISO := fromTime.Format(time.RFC3339) toISO := now.Format(time.RFC3339) + // Before s.mu: its rare inline rebuild must not run under the read lock. + _, pm := s.getCachedNodesAndPM() s.mu.RLock() defer s.mu.RUnlock() @@ -10287,12 +10314,18 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR packets := s.nodeTxsSince(pubkey, coveredFrom.Format(time.RFC3339)) coveredHours := now.Sub(coveredFrom).Hours() - // Activity timeline (hourly buckets) + // Activity timeline (hourly buckets). Certain counts the packets that + // can only be this node (#2260); the rest were attributed to it through + // a hop prefix it shares, and may be another node's. timelineBuckets := map[string]int{} + certainBuckets := map[string]int{} for _, p := range packets { if len(p.FirstSeen) >= 13 { bucket := p.FirstSeen[:13] + ":00:00Z" timelineBuckets[bucket]++ + if certainFor(p, pubkey, pm) { + certainBuckets[bucket]++ + } } } bucketKeys := make([]string, 0, len(timelineBuckets)) @@ -10301,9 +10334,14 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR } sort.Strings(bucketKeys) activityTimeline := make([]TimeBucket, 0, len(bucketKeys)) + certainKeys := make([]string, 0, len(certainBuckets)) for _, k := range bucketKeys { b := k - activityTimeline = append(activityTimeline, TimeBucket{Bucket: &b, Count: timelineBuckets[k]}) + c := certainBuckets[k] + activityTimeline = append(activityTimeline, TimeBucket{Bucket: &b, Count: timelineBuckets[k], Certain: &c}) + if c > 0 { + certainKeys = append(certainKeys, k) + } } // SNR trend @@ -10503,9 +10541,11 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR uptimeHeatmap = append(uptimeHeatmap, *cell) } - // Computed stats + // Computed stats. Availability and the longest silence measure the node + // being heard, so only certain buckets count (#2260): a silent repeater + // must not look available through its namesake prefix. totalPackets := len(packets) - distinctHours := len(activityTimeline) + distinctHours := len(certainKeys) // Null under minCoveredHours: too short a window to extrapolate from. var availabilityPct, avgPacketsPerDay *float64 if coveredHours >= minCoveredHours { @@ -10514,29 +10554,7 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR availabilityPct, avgPacketsPerDay = &avail, &perDay } - // Longest silence - var longestSilenceMs int - var longestSilenceStart interface{} - if len(activityTimeline) >= 2 { - for i := 1; i < len(activityTimeline); i++ { - var t1Str, t2Str string - if activityTimeline[i-1].Bucket != nil { - t1Str = *activityTimeline[i-1].Bucket - } - if activityTimeline[i].Bucket != nil { - t2Str = *activityTimeline[i].Bucket - } - t1, e1 := time.Parse(time.RFC3339, t1Str) - t2, e2 := time.Parse(time.RFC3339, t2Str) - if e1 == nil && e2 == nil { - gap := int(t2.Sub(t1).Milliseconds()) - if gap > longestSilenceMs { - longestSilenceMs = gap - longestSilenceStart = t1Str - } - } - } - } + longestSilenceMs, longestSilenceStart := longestSilence(certainKeys, coveredFrom, now) // Signal grade & SNR stats var snrMean, snrStdDev float64 diff --git a/cmd/server/types.go b/cmd/server/types.go index d5fe34aeb..2914ef205 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -50,6 +50,9 @@ type TimeBucket struct { Label *string `json:"label,omitempty"` Count int `json:"count"` Bucket *string `json:"bucket,omitempty"` + // Certain is the part of Count that can only be the node (#2260), on + // the node analytics timeline; absent elsewhere. + Certain *int `json:"certain,omitempty"` } // ─── Stats ───────────────────────────────────────────────────────────────────── diff --git a/cmd/server/user_mgmt_config.go b/cmd/server/user_mgmt_config.go index aba0a0fe6..a5cabea13 100644 --- a/cmd/server/user_mgmt_config.go +++ b/cmd/server/user_mgmt_config.go @@ -92,6 +92,17 @@ type NotificationsConfig struct { // ExternalAlerts feeds the node.external event (#2177). Absent: the // event is not offered. ExternalAlerts *ExternalAlertsConfig `json:"externalAlerts,omitempty"` + // AdaptiveOffline learns node.offline's threshold per node from its own + // certain traffic (#2260, adaptive_offline.go). Absent or not enabled: + // the fixed healthThresholds silent windows only. + AdaptiveOffline *AdaptiveOfflineConfig `json:"adaptiveOffline,omitempty"` +} + +// AdaptiveOfflineConfig is userManagement.notifications.adaptiveOffline. +type AdaptiveOfflineConfig struct { + Enabled bool `json:"enabled"` + FloorMinutes int `json:"floorMinutes,omitempty"` // shortest learned threshold, default 30 + MinObservers int `json:"minObservers,omitempty"` // fewer distinct observers keep the fixed window, default 3 } // ExternalAlertsConfig is userManagement.notifications.externalAlerts: a @@ -110,6 +121,30 @@ type notifySettings struct { maxMailsPerDay int // notification mails per instance per 24 hours maxWatchesPerUser int external externalAlertSettings + adaptive adaptiveSettings +} + +// adaptiveSettings is the resolved adaptiveOffline block; the zero value is +// off. +type adaptiveSettings struct { + enabled bool + floor time.Duration + minObservers int +} + +const ( + defaultAdaptiveFloorMinutes = 30 + defaultAdaptiveMinObservers = 3 +) + +// resolveAdaptiveOffline fills the defaults; nil or not enabled is off. +func resolveAdaptiveOffline(c *AdaptiveOfflineConfig) adaptiveSettings { + if c == nil || !c.Enabled { + return adaptiveSettings{} + } + return adaptiveSettings{enabled: true, + floor: time.Duration(positiveOr(c.FloorMinutes, defaultAdaptiveFloorMinutes)) * time.Minute, + minObservers: positiveOr(c.MinObservers, defaultAdaptiveMinObservers)} } // externalAlertSettings is the resolved externalAlerts block; url "" means @@ -165,6 +200,7 @@ func resolveNotifications(c *NotificationsConfig) notifySettings { perUserPerDay: positiveOr(c.PerUserPerDay, defaultNotifyPerUserPerDay), maxMailsPerDay: positiveOr(c.MaxMailsPerDay, defaultNotifyMaxMailsPerDay), maxWatchesPerUser: positiveOr(c.MaxWatchesPerUser, defaultNotifyMaxWatches), + adaptive: resolveAdaptiveOffline(c.AdaptiveOffline), } } diff --git a/config.example.json b/config.example.json index b744c8495..11a4e1522 100644 --- a/config.example.json +++ b/config.example.json @@ -482,7 +482,12 @@ "intervalMinutes": 5, "perUserPerDay": 20, "maxMailsPerDay": 100, - "maxWatchesPerUser": 50 + "maxWatchesPerUser": 50, + "adaptiveOffline": { + "enabled": false, + "floorMinutes": 30, + "minObservers": 3 + } }, "backup": { "enabled": true, @@ -492,7 +497,7 @@ }, "_comment_userManagement": "Optional accounts (off by default; see docs/user-guide/accounts.md). When enabled: visitors can register with email + password and activate through a mailed link; addresses in adminEmails become admins; admins manage users at #/admin/users and can use the operator endpoints without the apiKey. The dashboard stays public. dbPath defaults to users.db next to the analyzer database: a separate file, the server never writes measurement data. publicBaseUrl is required and is used for every link in every mail. Mail goes through Brevo's transactional API (a free account is enough); brevoApiKey and webhookSecret may instead come from CORESCOPE_BREVO_API_KEY and CORESCOPE_BREVO_WEBHOOK_SECRET. Set webhookSecret (16+ chars) and point a Brevo transactional webhook at /api/mail/brevo/webhook with bearer auth to see delivery status per user. Alternatively set mail.provider to \"postal\" to send through your own Postal server: postalBaseUrl (https) and postalApiKey (or CORESCOPE_POSTAL_API_KEY) are then required, and the webhook is https://postal:@/api/mail/postal/webhook (secret: letters, digits, . _ - only; or CORESCOPE_POSTAL_WEBHOOK_SECRET). trustedProxies (CIDRs) lets login rate limits use X-Forwarded-For behind a reverse proxy. Startup fails if enabled with an incomplete mail setup.", "_comment_channelProposals": "Optional, needs userManagement.enabled (docs/user-guide/accounts.md, Channel proposals). Logged-in users propose hashtag channels; admins approve, reject or revoke them on the Proposals tab. An approved channel is decrypted by the ingestor from its next refresh (once a minute, no restart) and listed for every visitor on the Channels page. maxApproved bounds the decryption work (every approved key is tried on every group message); perUserPerDay and maxPending limit proposals. The ingestor reads users.db read-only from userManagement.dbPath, or next to the analyzer database; set dbPath explicitly when the server and the ingestor are not given the same analyzer database path.", - "_comment_notifications": "Optional, needs userManagement.enabled (docs/user-guide/accounts.md, Node notifications). Logged-in users watch nodes and get one mail per check when a watched node goes offline, comes back or reports a low battery; admins can also watch for new foreign nodes and observers going offline. Thresholds are healthThresholds and batteryThresholds. A check runs every intervalMinutes; at most perUserPerDay mails per user and maxMailsPerDay in total go out per 24 hours (over a limit the change is recorded and dropped, never queued). maxMailsPerDay counts notification mail only: keep it below the mail provider's daily quota so activation, password-reset and address-change mail still go out (the default 100 leaves 200 of Brevo's free 300 a day); maxWatchesPerUser bounds each watch list. Every mail has a one-click unsubscribe. Optional externalAlerts {url, maxAgeHours (48), label} offers the node.external event, fed by a JSON alerts feed computed outside CoreScope (docs/user-guide/accounts.md, External node alerts).", + "_comment_notifications": "Optional, needs userManagement.enabled (docs/user-guide/accounts.md, Node notifications). Logged-in users watch nodes and get one mail per check when a watched node goes offline, comes back or reports a low battery; admins can also watch for new foreign nodes and observers going offline. Thresholds are healthThresholds and batteryThresholds. A check runs every intervalMinutes; at most perUserPerDay mails per user and maxMailsPerDay in total go out per 24 hours (over a limit the change is recorded and dropped, never queued). maxMailsPerDay counts notification mail only: keep it below the mail provider's daily quota so activation, password-reset and address-change mail still go out (the default 100 leaves 200 of Brevo's free 300 a day); maxWatchesPerUser bounds each watch list. Every mail has a one-click unsubscribe. Optional externalAlerts {url, maxAgeHours (48), label} offers the node.external event, fed by a JSON alerts feed computed outside CoreScope (docs/user-guide/accounts.md, External node alerts). Optional adaptiveOffline (off by default): node.offline learns each watched node's threshold from its own certain traffic over the week before its last certain packet, so a repeater normally heard every minute is reported after floorMinutes (30) instead of infraSilentHours; empty stretches of up to 6 h that recur on another day are normal silence (the window is never shorter than the longest plus an hour), other empty stretches are outages and are left out of the learning; nodes with under one certain packet an hour at the 10th percentile, under a day of history or a 10th-percentile hour heard by fewer than minObservers (3) observers keep the fixed window; silence is measured up to the newest stored packet, so a stalling feed mails nobody (docs/user-guide/accounts.md, Adaptive offline threshold).", "_comment_usersBackup": "Part of userManagement (docs/user-guide/accounts.md, Backups). With accounts on, the server snapshots users.db at startup when the newest snapshot is older than 24 hours, then about every 24 hours, to users-.db (UTC, mode 0600) in dir (empty: backups/ next to users.db), and deletes the oldest beyond keep (default 7), never the one just written. On unless enabled is false. Snapshots hold password hashes, and deleted accounts stay in them for up to keep days: protect the directory like users.db. A dir that already exists is not tightened to 0700: set that yourself. Admins download a fresh snapshot at /api/admin/users-backup. A restore brings back accounts deleted after the snapshot (also self-deletions), old passwords, revoked sessions and used links: follow the Restore steps in the doc (note the deletions from the current audit log first, then disable those accounts and clear sessions and tokens before starting, and delete them again in the admin area).", "customizer": { "disabledTabs": [], diff --git a/docs/specs/2026-10-07-node-notifications-design.md b/docs/specs/2026-10-07-node-notifications-design.md index 31900ea70..e45aa822f 100644 --- a/docs/specs/2026-10-07-node-notifications-design.md +++ b/docs/specs/2026-10-07-node-notifications-design.md @@ -202,6 +202,9 @@ except where a ruling below changes one. 1. "Per day" is a rolling 24 hours counted from `mail_log`, like the proposal limit. 2. Offline uses the node page's three timestamps: `nodes.last_seen`, the packet store's newest packet involving the node, and for repeaters and rooms `last_relayed`. + Superseded by #2260: `nodes.last_seen` and the packet store's newest packet that can + only be this node. `last_relayed` and packets merely attributed through a shared + 1-byte hop prefix kept a silent repeater online for days. 3. No evaluation runs while the packet store is still loading after a restart; the first one runs one interval after startup. 4. `foreign.new` keeps a per-admin baseline row (subject `*`): the first evaluation after diff --git a/docs/user-guide/accounts.md b/docs/user-guide/accounts.md index 73bdc77c8..9828db0fe 100644 --- a/docs/user-guide/accounts.md +++ b/docs/user-guide/accounts.md @@ -207,11 +207,34 @@ watched node changes state. Admins can also watch the instance. - Events: a watched node goes offline or comes back, using the node page's thresholds: offline once it has not been heard for the `healthThresholds` silent hours of its role - (heard is the latest of its last advert, the packet store's newest packet involving - it, and for repeaters and rooms its last relay hop). A watched node also reports a + (heard is the latest of its last advert and the newest packet that can only be this + node: its own packets, and relay hops whose prefix no other repeater shares. A 1-byte + hop shared with other repeaters is not counted, so a silent repeater is not kept + online by a namesake prefix; the node page shows such a repeater as *Uncertain*, and + its offline mail says that only its shared prefix was still seen). A watched node also reports a battery below `batteryThresholds.lowMv`, and recovers at `lowMv + 100` mV. Admins can add a new foreign node (once per node) and an observer going offline (`observerStaleMinutes`) or back (`observerOnlineMinutes`). +- Adaptive offline threshold (optional, off by default): with + `"adaptiveOffline": { "enabled": true, "floorMinutes": 30, "minObservers": 3 }` in + `notifications`, a watched node's offline window is learned from its own traffic + instead of the fixed silent hours. The server counts the node's certain packets per + hour over the week before its last certain packet (so an outage does not lower the + rate), takes the 10th percentile as the rate λ so quiet nights do not alert, and + reports the node offline after 14/λ hours of silence: the point where silence at that + rate has less than a one-in-a-million chance. The result is at least `floorMinutes` + and at most the role's silent hours. A repeater heard about 35 times an hour is thus + reported after 30 minutes instead of 72 hours. Empty stretches are sorted first: one of + at most 6 hours that comes back on another day at the same hours is the node's normal + silence, and the window is never shorter than the longest such stretch plus an hour; + any other empty stretch is an outage and is left out of the learning, so one outage + neither stretches the window for a week nor stops the node from being learned. Nodes + with less than one certain packet an hour at the 10th percentile, under a day of + history in memory, or whose 10th-percentile hour was heard by fewer than + `minObservers` observers keep the fixed window: when one observer carries a node, that + observer's outage would look like the node's. Silence is measured up to the newest + packet in the store, so a stalling feed stops every node's clock instead of mailing + the nodes with short windows. The learned windows are recomputed every 10 minutes. - A check runs every `intervalMinutes` (the first one an interval after startup, and only after the startup load). All changes for one user in one check go into one mail. The first check of a newly watched node stores its state without a mail; states diff --git a/public/live.js b/public/live.js index 8a35e393a..f2fa69554 100644 --- a/public/live.js +++ b/public/live.js @@ -3168,8 +3168,8 @@ // #1598: infra nodes (repeater/room) are ALWAYS dimmed, never deleted — // a dimmed marker still tells operators "infrastructure exists here, // advert stale", whereas deletion silently rewrites the map. Staleness - // itself is relay-aware via getNodeStatus(node) (max of advert-based - // freshness and last_relayed for infra). + // itself comes from getNodeStatus(node), whose last_heard includes the + // relays that can only be this node (#2260). // Remaining WS-only non-infra nodes are removed to prevent memory leaks. function pruneStaleNodes() { var now = Date.now(); @@ -3179,7 +3179,10 @@ if (!n) continue; var lastSeen = n._liveSeen || (n.last_heard ? new Date(n.last_heard).getTime() : null) || (n.last_seen ? new Date(n.last_seen).getTime() : null); if (lastSeen == null) continue; + // #2260: 'uncertain' (only a shared hop prefix seen) dims like stale; + // it only occurs for repeaters and rooms, which are never removed. var status = window.getNodeStatus ? getNodeStatus(n) : 'active'; + if (status === 'uncertain') status = 'stale'; var isInfra = n.role === 'repeater' || n.role === 'room'; var marker = nodeMarkers[key]; if (status === 'stale') { diff --git a/public/map.js b/public/map.js index be3c3cfed..a8d5e8451 100644 --- a/public/map.js +++ b/public/map.js @@ -530,6 +530,7 @@
+
@@ -1843,11 +1844,11 @@ // Count active/stale per role from loaded nodes const roleCounts = {}; for (const role of roles) { - roleCounts[role] = { active: 0, stale: 0 }; + roleCounts[role] = { active: 0, uncertain: 0, stale: 0 }; } for (const n of nodes) { const role = (n.role || 'companion').toLowerCase(); - if (!roleCounts[role]) roleCounts[role] = { active: 0, stale: 0 }; + if (!roleCounts[role]) roleCounts[role] = { active: 0, uncertain: 0, stale: 0 }; const status = getNodeStatus(n); // #1598: relay-aware for infra roleCounts[role][status]++; } @@ -1861,12 +1862,15 @@ if (role === 'observer') { countStr = `(${obsCount})`; } else { - const rc = roleCounts[role] || { active: 0, stale: 0 }; + const rc = roleCounts[role] || { active: 0, uncertain: 0, stale: 0 }; const isInfra = role === 'repeater' || role === 'room'; const thresh = isInfra ? '72h' : '24h'; const activeTip = 'Active \u2014 heard within the last ' + thresh; const staleTip = 'Stale \u2014 not heard for over ' + thresh; - countStr = `(${rc.active} active, ${rc.stale} stale)`; + const uncertainTip = 'Uncertain \u2014 only a hop prefix shared with other repeaters still appears in paths'; + countStr = `(${rc.active} active, ` + + (rc.uncertain ? `${rc.uncertain} uncertain, ` : '') + + `${rc.stale} stale)`; } lbl.innerHTML = ` ${ROLE_LABELS[role]} ${countStr}`; lbl.querySelector('input').addEventListener('change', e => { @@ -2109,7 +2113,8 @@ for (const node of filtered) { const lastSeenTime = node.last_heard || node.last_seen; - const isStale = getNodeStatus(node) === 'stale'; // #1598: relay-aware for infra + // #2260: uncertain is not proof of life either, so it dims like stale. + const isStale = getNodeStatus(node) !== 'active'; const pk = (node.public_key || '').toLowerCase(); const isAlsoObserver = _observerByPubkey.has(pk); const useLabel = node.role === 'repeater' && filters.hashLabels; diff --git a/public/node-analytics.js b/public/node-analytics.js index 67eda90de..c7eb9fffe 100644 --- a/public/node-analytics.js +++ b/public/node-analytics.js @@ -60,6 +60,22 @@ } window.NodeAnalyticsCoverage = { notice: coverageNotice, rateText: rateText }; + // #2260: split each timeline bucket into packets that can only be this + // node (certain) and packets attributed to it through a hop prefix it + // shares with other nodes (uncertain). A server without the certain field + // counts everything as certain, as before. + function timelineSeries(tl) { + const certain = [], uncertain = []; + (tl || []).forEach(b => { + const count = Number(b.count) || 0; + const c = b.certain == null ? count : Math.min(Number(b.certain) || 0, count); + certain.push(c); + uncertain.push(count - c); + }); + return { certain, uncertain }; + } + window.NodeAnalyticsTimeline = { series: timelineSeries }; + async function loadAnalytics(container, pubkey, days) { const gen = viewGeneration(); const seq = ++loadSeq; @@ -105,7 +121,7 @@
Availability
${rateText(s.availabilityPct, '%')}
-
% of time windows with at least one packet
+
% of hours in which this node was certainly heard
Signal Grade
@@ -130,14 +146,14 @@
Longest Silence
${formatSilence(s.longestSilenceMs)}
-
Longest gap between consecutive packets
+
Longest gap without a packet certainly from this node, including one still running

Activity Timeline

-
Packet count per time bucket — shows when this node is most active
+
Packet count per time bucket. Certain: can only be this node. Uncertain: a hop prefix this node shares with others, so possibly another node.
@@ -214,6 +230,10 @@ const ctx = document.getElementById('activityChart'); if (!ctx) return; const tl = data.activityTimeline; + const series = timelineSeries(tl); + const style = getComputedStyle(document.documentElement); + const certainColor = style.getPropertyValue('--accent').trim(); + const uncertainColor = style.getPropertyValue('--text-muted').trim(); const c = new Chart(ctx, { type: 'bar', data: { @@ -221,9 +241,12 @@ const d = new Date(b.bucket); return (typeof formatChartAxisLabel === 'function') ? formatChartAxisLabel(d, currentDays <= 3) : (currentDays <= 3 ? d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }) : d.toLocaleDateString([], { month: 'short', day: 'numeric' })); }), - datasets: [{ label: 'Packets', data: tl.map(b => b.count), backgroundColor: 'rgba(74,158,255,0.5)', borderColor: '#4a9eff', borderWidth: 1 }] + datasets: [ + { label: 'Certain', data: series.certain, backgroundColor: certainColor, borderWidth: 0, stack: 'p' }, + { label: 'Uncertain', data: series.uncertain, backgroundColor: uncertainColor, borderWidth: 0, stack: 'p' } + ] }, - options: { responsive: true, plugins: { legend: { display: false } }, scales: { x: { ticks: { maxTicksAutoSkip: true, maxRotation: 45 } }, y: { beginAtZero: true } } } + options: { responsive: true, plugins: { legend: { display: true, position: 'bottom', labels: { boxWidth: 12 } } }, scales: { x: { stacked: true, ticks: { maxTicksAutoSkip: true, maxRotation: 45 } }, y: { stacked: true, beginAtZero: true } } } }); charts.push(c); } diff --git a/public/nodes.js b/public/nodes.js index e6fc5cc32..7bf528638 100644 --- a/public/nodes.js +++ b/public/nodes.js @@ -83,8 +83,9 @@ let lastHeard = localStorage.getItem('meshcore-nodes-last-heard') || ''; let statusFilter = localStorage.getItem('meshcore-nodes-status-filter') || 'all'; // #1845: "silent longer than N", the inverse of the Last Heard filter. Empty - // means off. Ages are relay-aware (getEffectiveHeardMs), so a repeater that - // forwards traffic is never counted as silent just because its ADVERT is old. + // means off. Ages come from getEffectiveHeardMs, whose last_heard includes + // certain relays (#2260), so a repeater that forwards traffic is never + // counted as silent just because its ADVERT is old. let silentFor = localStorage.getItem('meshcore-nodes-silent-for') || ''; const SILENT_MS = { '1d': 86400000, '3d': 259200000, '7d': 604800000, '14d': 1209600000, '30d': 2592000000 }; // #1268 M3: '' or 'differs' (LoRa settings differ from nearby same-band nodes). @@ -199,6 +200,9 @@ if (status === 'active') { return 'Active \u2014 heard within the last ' + threshold + '.' + (isInfra ? ' Repeaters typically advertise every 12-24h.' : ''); } + if (status === 'uncertain') { + return 'Uncertain \u2014 nothing that can only be this ' + role + ' for over ' + threshold + '. Its hop prefix, shared with other repeaters, still appears in paths: that may be this node or another one.'; + } if (role === 'companion') { return 'Stale \u2014 not heard for over ' + threshold + '. Companions only advertise when the user initiates \u2014 this may be normal.'; } @@ -221,25 +225,27 @@ // Single source of truth for all status-related info const role = (n.role || '').toLowerCase(); const roleColor = ROLE_COLORS[n.role] || '#6b7280'; - // Prefer last_heard (from in-memory packets) > _lastHeard (health API) > last_seen (DB) - const lastHeardTime = n._lastHeard || n.last_heard || n.last_seen; - const lastHeardMs = lastHeardTime ? new Date(lastHeardTime).getTime() : 0; - // #1598: pass the full node so infra staleness also considers last_relayed + // The newest of _lastHeard (health API), last_heard and last_seen, the + // same timestamp the status uses (getEffectiveHeardMs, #2260). + const effMs = window.getEffectiveHeardMs(n); + const lastHeardMs = isNaN(effMs) ? 0 : effMs; + const lastHeardTime = lastHeardMs ? new Date(lastHeardMs).toISOString() : null; + // #1598/#2260: certain relays are already in last_heard const status = getNodeStatus(n); const statusTooltip = getStatusTooltip(role, status); - const statusLabel = status === 'active' ? ' Active' : ' Stale'; + const statusLabel = status === 'active' ? ' Active' + : status === 'uncertain' ? ' Uncertain' + : ' Stale'; const statusAge = lastHeardMs ? (Date.now() - lastHeardMs) : Infinity; let explanation = ''; if (status === 'active') { - // #1598: an infra node can be active purely via relay participation — - // surface that instead of a misleading old "last heard". - const relayedMs = n.last_relayed ? new Date(n.last_relayed).getTime() : 0; - if (relayedMs > lastHeardMs) { - explanation = 'Last relayed ' + renderNodeTimestampText(n.last_relayed); - } else { - explanation = 'Last heard ' + (lastHeardTime ? renderNodeTimestampText(lastHeardTime) : 'unknown'); - } + // #2260: no "Last relayed" here. last_relayed also counts 1-byte prefix + // matches of other repeaters; the certain relays are in last heard. + explanation = 'Last heard ' + (lastHeardTime ? renderNodeTimestampText(lastHeardTime) : 'unknown'); + } else if (status === 'uncertain') { + explanation = 'Last certainly heard ' + (lastHeardTime ? renderNodeTimestampText(lastHeardTime) : 'never') + + ' \u2014 its shared hop prefix was last in a path ' + renderNodeTimestampText(n.last_relayed) + ', possibly another node'; } else { const ageDays = Math.floor(statusAge / 86400000); const ageHours = Math.floor(statusAge / 3600000); @@ -728,10 +734,12 @@ const observers = h.observers || []; const relayObserverCount = Number(h.relayObserverCount) || 0; const recent = h.recentPackets || []; - const lastHeard = stats.lastHeard; - - // Attach health lastHeard for shared helpers - n._lastHeard = lastHeard || n.last_seen; + // Attach health lastHeard for shared helpers. Displayed: the newest of + // it, the node's last_heard and last_seen (#2260); either fetch may be + // the fresher one. + n._lastHeard = stats.lastHeard || n.last_seen; + const effMs = window.getEffectiveHeardMs(n); + const lastHeard = isNaN(effMs) ? null : new Date(effMs).toISOString(); const si = getStatusInfo(n); const roleColor = si.roleColor; const statusLabel = si.statusLabel; @@ -771,7 +779,7 @@ - ${(n.role === 'repeater' || n.role === 'room') ? `` : ''} + ${(n.role === 'repeater' || n.role === 'room') ? `` : ''} ${(n.role === 'repeater' || n.role === 'room') && (n.traffic_share_score != null || n.usefulness_score != null) ? (() => { // #1456: prefer the new traffic_share_score field; fall back // to legacy usefulness_score for graceful degradation @@ -1416,7 +1424,7 @@ }); } // Status filter (active/stale) - if (statusFilter === 'active' || statusFilter === 'stale') { + if (statusFilter === 'active' || statusFilter === 'stale' || statusFilter === 'uncertain') { filtered = filtered.filter(n => getNodeStatus(n) === statusFilter); // #1598 } // #1845: counts are taken BEFORE the silence filter is applied, so the @@ -1518,6 +1526,7 @@
+
Status${statusLabel} ${statusExplanation}
Last Heard${renderNodeTimestampHtml(lastHeard || n.last_seen)}
Last Relayed${n.last_relayed ? renderNodeTimestampHtml(n.last_relayed) + ' ' + (n.relay_active ? ' actively relaying' : ' alive (idle)') : 'never observed as relay hop alive (idle)'}${(n.relay_count_1h != null || n.relay_count_24h != null) ? ` (${n.relay_count_1h || 0} relays/hr, ${n.relay_count_24h || 0} relays/24h)` : ''}
Last Relayed${n.last_relayed ? renderNodeTimestampHtml(n.last_relayed) + ' ' + (n.relay_active ? ' actively relaying' : ' alive (idle)') : 'never observed as relay hop alive (idle)'}${(n.relay_count_1h != null || n.relay_count_24h != null) ? ` (${n.relay_count_1h || 0} relays/hr, ${n.relay_count_24h || 0} relays/24h)` : ''}