From 2bb000f2aaf7d83f76dcbf33521ddd21180960e3 Mon Sep 17 00:00:00 2001 From: efiten Date: Thu, 22 Oct 2026 21:20:42 +0200 Subject: [PATCH] feat(notify): adaptive offline threshold and certain node analytics (#2260) (#2262) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 2 of #2260. **Builds on #2261** (certain liveness): review/merge that first; until then this diff also shows its commit. ## Adaptive offline threshold `userManagement.notifications.adaptiveOffline` — **off by default**: ```json "notifications": { "enabled": true, "adaptiveOffline": { "enabled": true, "floorMinutes": 30, "minObservers": 3 } } ``` The fixed `infraSilentHours` (72 h) treats every repeater the same. With this on, a watched node's `node.offline` window is learned from its own **certain** traffic (the #2261 rule): 1. Certain packets per hour over the week **before its last certain packet**. Ending there, not at now, keeps an outage from dragging the rate down until the fixed window returns and the node reads "back online". 2. Empty runs are classified: ≤ 6 h and recurring on another day at the same hours = normal silence (kept; the window is never shorter than the longest + 1 h); any other run = outage, left out of the learning so it neither stretches the window nor blocks learning. 3. λ = 10th percentile of the remaining hourly counts, so quiet nights don't alert. 4. Window = 14/λ hours: silence at rate λ has a < 10⁻⁶ chance (Poisson, ln 10⁶ ≈ 13.8). 5. Clamped to [`floorMinutes`, the role's silent window]. Kept on the fixed window: λ < 1/hour, under 24 h of history in memory, or a 10th-percentile hour heard by fewer than `minObservers` observers (one observer dropping out must not read as the node going offline). Silence is measured up to the newest packet in the store, so a stalling feed stops every node's clock. A node stored offline comes back only on evidence from the moment it went offline on. The notifier caches each learned window for 10 minutes and drops unwatched nodes. For the #2260 repeater (~840 packets/day) the mail would have gone out about 30 minutes into the outage, instead of never (or 72 h with #2261 alone). ## Certain node analytics - Each `activityTimeline` bucket carries `certain`: the part that can only be this node (`certainFor`, the packet-level form of the `certainHeard` rule). `TimeBucket.Certain` is `omitempty`, so other endpoints are unchanged. - Availability and longest silence use certain buckets only; the longest silence now includes one **still running** (an outage in progress is the silence worth seeing). - The chart stacks certain (`--accent`) and uncertain (`--text-muted`) — existing theme variables, so no new customizer entry. Descriptions updated. ## Performance `CertainActivity` walks a week of packets per watched node, at most once per node per 10 minutes, with the read lock taken **per node**: ``` BenchmarkCertainActivity-16 12.1 ms/op (50 watched repeaters × 6,000 packets; ~0.25 ms per lock hold) BenchmarkNoteCertainRelays-16 121 ns/op 0 allocs ``` Node analytics adds one `certainFor` per packet of the requested node (a prefix-map lookup, or the cached decoded JSON for its own packets). ## Tests - `adaptiveOfflineAfter`: busy repeater hits the floor; 5/hour; quiet nights set the rate; > 10 % empty hours, short history, too few observers → fixed. - `CertainActivity` ends at the last certain packet and counts only certain packets — verified red by mutating the window to end at now. - Notifier end to end: learned window passed → bad; inside → good; too few observers / setting off → fixed 72 h; cache refresh and prune. - `certainFor`, `longestSilence` (incl. running silence), config defaults, `/api/nodes/{pk}/analytics` splits a shared-prefix bucket. - Frontend: `NodeAnalyticsTimeline.series` (new `tests/unit/test-node-analytics-certain-timeline.js`). - `cmd/server` `go test ./...` and `test-all.sh` pass locally, also on a trial merge into a downstream fork. ## Validation Server on the freshened e2e fixture: analytics buckets carry `certain` (e.g. 89 of 90 packets certain for one repeater). **No browser screenshot**: Playwright isn't installed on this machine. ## Later (rule 8) `floorMinutes` / `minObservers` are server config, not theme settings; an admin UI for notification settings would be the place, not the customizer. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 (1M context) --- cmd/server/adaptive_offline.go | 264 ++++++++++ cmd/server/adaptive_offline_test.go | 453 ++++++++++++++++++ cmd/server/certain_heard.go | 62 +++ cmd/server/node_analytics_coverage_test.go | 7 +- cmd/server/notify_eval.go | 38 +- cmd/server/notify_service.go | 4 +- cmd/server/notify_service_test.go | 6 + cmd/server/notify_source.go | 8 + cmd/server/store.go | 47 +- cmd/server/types.go | 3 + cmd/server/user_mgmt_config.go | 36 ++ config.example.json | 9 +- docs/user-guide/accounts.md | 20 + public/node-analytics.js | 33 +- test-all.sh | 1 + .../test-node-analytics-certain-timeline.js | 61 +++ 16 files changed, 1011 insertions(+), 41 deletions(-) create mode 100644 cmd/server/adaptive_offline.go create mode 100644 cmd/server/adaptive_offline_test.go create mode 100644 tests/unit/test-node-analytics-certain-timeline.js diff --git a/cmd/server/adaptive_offline.go b/cmd/server/adaptive_offline.go new file mode 100644 index 00000000..4dd16fd6 --- /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 00000000..d5fd2867 --- /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 index 9c3da21d..0f10aa13 100644 --- a/cmd/server/certain_heard.go +++ b/cmd/server/certain_heard.go @@ -139,6 +139,27 @@ func (pm *prefixMap) markRetired(db *DB) error { 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 @@ -154,3 +175,44 @@ func lastHeardValue(lastSeen interface{}, heard time.Time) interface{} { } 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/node_analytics_coverage_test.go b/cmd/server/node_analytics_coverage_test.go index 850edaba..f0baf5c4 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 58f00578..525e21bd 100644 --- a/cmd/server/notify_eval.go +++ b/cmd/server/notify_eval.go @@ -46,7 +46,16 @@ type notifyInput struct { // 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 - Observers []notifyObserver // nil unless an admin chose observer.offline + // 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). @@ -118,9 +127,15 @@ func nodeLabel(n notifyNode, known bool, pk string) string { // 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. A node missing from the analyzer database is bad. While prev is good, the +// "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 } @@ -131,8 +146,19 @@ func nodeOnlineState(pk string, n notifyNode, known bool, prev string, in *notif 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 @@ -220,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 { @@ -294,7 +322,7 @@ 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} - state := nodeOnlineState(pk, n, known, prev[k], &in) + 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 { diff --git a/cmd/server/notify_service.go b/cmd/server/notify_service.go index b33031ee..0cc0fbc0 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 { @@ -210,7 +212,7 @@ func (n *notifier) tick(ctx context.Context) { } 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, SharedSeen: n.src.sharedPrefixSeen(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 diff --git a/cmd/server/notify_service_test.go b/cmd/server/notify_service_test.go index bab0d543..07341bed 100644 --- a/cmd/server/notify_service_test.go +++ b/cmd/server/notify_service_test.go @@ -21,6 +21,8 @@ type fakeNotifySource struct { nodeMap map[string]notifyNode heard 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 @@ -58,6 +60,10 @@ 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 diff --git a/cmd/server/notify_source.go b/cmd/server/notify_source.go index 93b8dbb0..89d71252 100644 --- a/cmd/server/notify_source.go +++ b/cmd/server/notify_source.go @@ -16,6 +16,7 @@ type notifySource interface { 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 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 @@ -72,6 +73,13 @@ func (src serverNotifySource) sharedPrefixSeen(pubkeys []string) map[string]time 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 { if src.s.store == nil { return time.Time{} diff --git a/cmd/server/store.go b/cmd/server/store.go index 68a61a3e..81ed86e7 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -10290,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() @@ -10312,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)) @@ -10326,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 @@ -10528,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 { @@ -10539,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 d5fe34ae..2914ef20 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 aba0a0fe..a5cabea1 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 b744c849..11a4e152 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/user-guide/accounts.md b/docs/user-guide/accounts.md index b991dfda..9828db0f 100644 --- a/docs/user-guide/accounts.md +++ b/docs/user-guide/accounts.md @@ -215,6 +215,26 @@ watched node changes state. Admins can also watch the instance. 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/node-analytics.js b/public/node-analytics.js index 67eda90d..c7eb9fff 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/test-all.sh b/test-all.sh index 4a50b662..d84bca08 100755 --- a/test-all.sh +++ b/test-all.sh @@ -107,6 +107,7 @@ node tests/unit/test-issue-1518-home-url.js node tests/unit/test-issue-1532-live-fullscreen.js node tests/unit/test-issue-1562-observers-summary.js node tests/unit/test-issue-748-node-analytics-coverage.js +node tests/unit/test-node-analytics-certain-timeline.js node tests/unit/test-issue-661-observer-airtime.js node tests/unit/test-issue-1268-rf-params.js node tests/unit/test-issue-1563-aggregate-and-inflight.js diff --git a/tests/unit/test-node-analytics-certain-timeline.js b/tests/unit/test-node-analytics-certain-timeline.js new file mode 100644 index 00000000..f595bf29 --- /dev/null +++ b/tests/unit/test-node-analytics-certain-timeline.js @@ -0,0 +1,61 @@ +/** + * #2260 — node analytics timeline: each bucket splits into packets that can + * only be this node (certain) and packets attributed through a hop prefix it + * shares (uncertain). Loads the real public/node-analytics.js in a vm sandbox. + */ +'use strict'; +const fs = require('fs'); +const path = require('path'); +const vm = require('vm'); +const assert = require('assert'); + +const REPO_ROOT = path.resolve(__dirname, '..', '..'); +let passed = 0, failed = 0; +function t(name, fn) { + try { fn(); passed++; console.log(' ✓ ' + name); } + catch (e) { failed++; console.error(' ✗ ' + name + ': ' + e.message); } +} + +const ctx = { window: {}, registerPage: () => {}, Date, Math, Number, String, isNaN, console }; +vm.createContext(ctx); +vm.runInContext(fs.readFileSync(path.join(REPO_ROOT, 'public', 'node-analytics.js'), 'utf8'), ctx); +const series = ctx.window.NodeAnalyticsTimeline && ctx.window.NodeAnalyticsTimeline.series; + +console.log('node-analytics.js certain timeline (#2260)'); + +t('exposes NodeAnalyticsTimeline.series', () => { + assert.strictEqual(typeof series, 'function'); +}); + +t('splits each bucket into certain and uncertain', () => { + const s = series([{ count: 80, certain: 80 }, { count: 3, certain: 0 }, { count: 10, certain: 4 }]); + assert.deepStrictEqual(Array.from(s.certain), [80, 0, 4]); + assert.deepStrictEqual(Array.from(s.uncertain), [0, 3, 6]); +}); + +t('the silent repeater of #2260: after the outage only uncertain bars remain', () => { + const s = series([{ count: 94, certain: 94 }, { count: 2, certain: 0 }, { count: 1, certain: 0 }]); + assert.deepStrictEqual(Array.from(s.certain), [94, 0, 0]); + assert.strictEqual(s.uncertain.reduce((a, b) => a + b, 0), 3); +}); + +t('a server without the certain field counts everything as certain', () => { + const s = series([{ count: 5 }, { count: 0 }]); + assert.deepStrictEqual(Array.from(s.certain), [5, 0]); + assert.deepStrictEqual(Array.from(s.uncertain), [0, 0]); +}); + +t('certain never exceeds count, strings are cast', () => { + const s = series([{ count: '4', certain: '9' }]); + assert.deepStrictEqual(Array.from(s.certain), [4]); + assert.deepStrictEqual(Array.from(s.uncertain), [0]); +}); + +t('missing timeline yields empty series', () => { + const s = series(undefined); + assert.strictEqual(s.certain.length, 0); + assert.strictEqual(s.uncertain.length, 0); +}); + +console.log(`\n${passed} passed, ${failed} failed`); +if (failed) process.exit(1);