diff --git a/.eslintrc.json b/.eslintrc.json index a469ec57..deb317e8 100644 --- a/.eslintrc.json +++ b/.eslintrc.json @@ -38,6 +38,7 @@ "MeshAudio": "readonly", "MeshConfigReady": "readonly", "NodeHopAnalytics": "readonly", + "ObserverAirtime": "readonly", "PAYLOAD_COLORS": "readonly", "PAYLOAD_TYPES": "readonly", "PERF_SLOW_MS": "readonly", diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 2c41202d..5796328a 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -583,6 +583,7 @@ jobs: suite 2 test-map-nodes-pagination-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B suite 3 test-observer-iata-1188-e2e.js BASE_URL=$B suite 2 test-issue-1639-observers-sort-e2e.js BASE_URL=$B + suite 2 test-issue-661-airtime-e2e.js BASE_URL=$B suite 2 test-issue-1758-ng-filter-rerenders-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B suite 2 test-nav-fluid-1055-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B suite 3 test-nav-priority-1102-e2e.js CHROMIUM_REQUIRE=1 BASE_URL=$B diff --git a/cmd/server/config.go b/cmd/server/config.go index 4fa6f543..acc3315b 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -209,6 +209,10 @@ type Config struct { // BatteryThresholds: voltage cutoffs for low/critical alerts (#663). BatteryThresholds *BatteryThresholdsConfig `json:"batteryThresholds,omitempty"` + // DutyCycleWarnPct: rolling 1 h TX airtime share (percent) at which the + // RF Health view and observer list warn. 0/missing => 10 (EU g3, #661). + DutyCycleWarnPct float64 `json:"dutyCycleWarnPct,omitempty"` + // Customizer controls operator-side knobs for the in-app customizer modal // (theme/branding/etc.). See CustomizerConfig and issue #1508. Customizer *CustomizerConfig `json:"customizer,omitempty"` diff --git a/cmd/server/db.go b/cmd/server/db.go index 747a4f3d..68ff6155 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -3038,12 +3038,16 @@ type MetricsSummaryRow struct { CurrentBattMv *int `json:"battery_mv"` SampleCount int `json:"sample_count"` Sparkline []*float64 `json:"sparkline"` + // Airtime share over the hour before now, whatever the window (#661). + TxAirtimePct1h *float64 `json:"tx_airtime_pct_1h"` + RxAirtimePct1h *float64 `json:"rx_airtime_pct_1h"` } // GetMetricsSummary returns a fleet summary of observer metrics within a time window. // Uses a CTE with ROW_NUMBER to get latest values in a single pass (no correlated subqueries). -// Also returns sparkline data (noise_floor time series) per observer. -func (db *DB) GetMetricsSummary(since string) ([]MetricsSummaryRow, error) { +// Also returns sparkline data (noise_floor time series) per observer, and the +// TX/RX airtime share over the hour before now. +func (db *DB) GetMetricsSummary(since string, now time.Time) ([]MetricsSummaryRow, error) { query := ` WITH ranked AS ( SELECT observer_id, noise_floor, battery_mv, @@ -3110,11 +3114,51 @@ func (db *DB) GetMetricsSummary(since string) ([]MetricsSummaryRow, error) { result[i].Sparkline = s } } + + if err := db.fillAirtime1h(result, now); err != nil { + return nil, err + } } return result, nil } +// fillAirtime1h sets the rolling 1 h TX/RX airtime share on each row from +// the cumulative counters sampled in the hour before now, in one query. +func (db *DB) fillAirtime1h(rows []MetricsSummaryRow, now time.Time) error { + since := now.Add(-airtimeWindow).Format(time.RFC3339) + q, err := db.conn.Query(`SELECT observer_id, timestamp, tx_air_secs, rx_air_secs + FROM observer_metrics WHERE timestamp >= ? ORDER BY observer_id, timestamp ASC`, since) + if err != nil { + return err + } + defer q.Close() + + tx := make(map[string][]airtimePoint) + rx := make(map[string][]airtimePoint) + for q.Next() { + var oid, ts string + var txSecs, rxSecs *int + if err := q.Scan(&oid, &ts, &txSecs, &rxSecs); err != nil { + return err + } + at, err := time.Parse(time.RFC3339, ts) + if err != nil { + continue + } + tx[oid] = append(tx[oid], airtimePoint{At: at, Secs: txSecs}) + rx[oid] = append(rx[oid], airtimePoint{At: at, Secs: rxSecs}) + } + if err := q.Err(); err != nil { + return err + } + for i := range rows { + rows[i].TxAirtimePct1h = rollingAirtimePct(tx[rows[i].ObserverID], now) + rows[i].RxAirtimePct1h = rollingAirtimePct(rx[rows[i].ObserverID], now) + } + return nil +} + // (PruneOldMetrics / RemoveStaleObservers removed in #1283 â see note // above the MetricsSample type. Ingestor owns these writes now.) diff --git a/cmd/server/db_test.go b/cmd/server/db_test.go index 97782764..1e788e74 100644 --- a/cmd/server/db_test.go +++ b/cmd/server/db_test.go @@ -1918,7 +1918,7 @@ func TestGetMetricsSummary(t *testing.T) { "obs2", t1, -115.0) since := now.Add(-24 * time.Hour).Format(time.RFC3339) - summary, err := db.GetMetricsSummary(since) + summary, err := db.GetMetricsSummary(since, now) if err != nil { t.Fatal(err) } diff --git a/cmd/server/observer_airtime.go b/cmd/server/observer_airtime.go new file mode 100644 index 00000000..d92b1f42 --- /dev/null +++ b/cmd/server/observer_airtime.go @@ -0,0 +1,81 @@ +package main + +import ( + "math" + "time" +) + +// Rolling 1 h airtime per observer (#661). The firmware enforces its own +// duty-cycle budget over a 1 h window (Dispatcher.cpp, duty_cycle_window_ms), +// but repeaters default to airtime_factor 1.0, i.e. a 50% budget, so the EU +// 10% limit is not enforced on the device. The summary therefore reports the +// TX share of the last hour and the threshold the UI warns at. + +// airtimeWindow is the span the rolling figure covers, matching the +// firmware's duty-cycle window. +const airtimeWindow = time.Hour + +// airtimeMinSpan is the shortest stretch of samples that yields a figure; +// shorter (one sample, or a reboot late in the hour) reads as unknown. The +// ingestor rounds sample times to its sample interval (5 min by default), so +// the span can be off by up to one interval: about 17% at 30 min, 8% at 1 h. +const airtimeMinSpan = 30 * time.Minute + +// airtimeMaxAge is how old the newest sample may be; an observer that has +// gone quiet longer than this reads as unknown rather than "last hour". +const airtimeMaxAge = 15 * time.Minute + +// defaultDutyCycleWarnPct is the EU 868 MHz g3 sub-band duty-cycle limit. +const defaultDutyCycleWarnPct = 10.0 + +// airtimePoint is one sample of a cumulative airtime counter (seconds since +// boot); Secs is nil when the observer did not report it. +type airtimePoint struct { + At time.Time + Secs *int +} + +// rollingAirtimePct returns the airtime share, in percent, over the points +// given in time order. A counter that goes down marks a reboot: only the +// stretch after the last reset counts. Returns nil when that stretch is +// shorter than airtimeMinSpan, or the newest sample is older than +// airtimeMaxAge at now. +func rollingAirtimePct(pts []airtimePoint, now time.Time) *float64 { + var start, last *airtimePoint + for i := range pts { + p := &pts[i] + if p.Secs == nil { + continue + } + if start == nil || *p.Secs < *last.Secs { + start = p + } + last = p + } + if start == nil || now.Sub(last.At) > airtimeMaxAge { + return nil + } + span := last.At.Sub(start.At) + if span < airtimeMinSpan { + return nil + } + pct := float64(*last.Secs-*start.Secs) / span.Seconds() * 100 + pct = math.Min(math.Max(pct, 0), 100) + pct = math.Round(pct*100) / 100 + return &pct +} + +// DutyCycleWarnPctOrDefault returns the configured duty-cycle warning +// threshold in percent, or the EU 10% default. +func (c *Config) DutyCycleWarnPctOrDefault() float64 { + if c.DutyCycleWarnPct > 0 { + return c.DutyCycleWarnPct + } + return defaultDutyCycleWarnPct +} + +// MetricsSummaryResponse is the body of GET /api/observers/metrics/summary. +type MetricsSummaryResponse struct { + Observers []MetricsSummaryRow `json:"observers"` + DutyCycleWarnPct float64 `json:"duty_cycle_warn_pct"` +} diff --git a/cmd/server/observer_airtime_test.go b/cmd/server/observer_airtime_test.go new file mode 100644 index 00000000..a0c15154 --- /dev/null +++ b/cmd/server/observer_airtime_test.go @@ -0,0 +1,124 @@ +package main + +import ( + "encoding/json" + "net/http/httptest" + "testing" + "time" +) + +func airPt(base time.Time, minAgo int, secs int) airtimePoint { + v := secs + return airtimePoint{At: base.Add(-time.Duration(minAgo) * time.Minute), Secs: &v} +} + +func pctOrNil(p *float64) interface{} { + if p == nil { + return nil + } + return *p +} + +func TestRollingAirtimePct(t *testing.T) { + now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC) + cases := []struct { + name string + pts []airtimePoint + want interface{} + }{ + {"steady 10%", []airtimePoint{airPt(now, 60, 1000), airPt(now, 30, 1180), airPt(now, 0, 1360)}, 10.0}, + {"reboot mid-hour uses the segment after the reset", + []airtimePoint{airPt(now, 60, 0), airPt(now, 40, 100), airPt(now, 30, 5), airPt(now, 0, 185)}, 10.0}, + {"single sample", []airtimePoint{airPt(now, 0, 50)}, nil}, + {"no samples", nil, nil}, + {"span below 30 minutes", []airtimePoint{airPt(now, 25, 0), airPt(now, 0, 150)}, nil}, + {"span of exactly 30 minutes", []airtimePoint{airPt(now, 30, 0), airPt(now, 0, 180)}, 10.0}, + {"reboot leaves less than 30 minutes", []airtimePoint{airPt(now, 60, 0), airPt(now, 30, 400), airPt(now, 25, 2), airPt(now, 0, 20)}, nil}, + {"newest sample older than 15 minutes", []airtimePoint{airPt(now, 60, 0), airPt(now, 20, 240)}, nil}, + {"newest sample exactly 15 minutes old", []airtimePoint{airPt(now, 60, 0), airPt(now, 15, 270)}, 10.0}, + {"nil counters are skipped", + []airtimePoint{airPt(now, 60, 0), {At: now.Add(-30 * time.Minute)}, airPt(now, 0, 36)}, 1.0}, + {"clamped to 100", []airtimePoint{airPt(now, 30, 0), airPt(now, 0, 5000)}, 100.0}, + {"rounded to 2 decimals", []airtimePoint{airPt(now, 60, 0), airPt(now, 0, 1)}, 0.03}, + } + for _, c := range cases { + got := pctOrNil(rollingAirtimePct(c.pts, now)) + if got != c.want { + t.Errorf("%s: got %v, want %v", c.name, got, c.want) + } + } +} + +func TestGetMetricsSummaryAirtime1h(t *testing.T) { + db := setupTestDB(t) + seedTestData(t, db) + now := time.Now().UTC().Truncate(time.Second) + ins := func(obs string, minAgo, tx, rx int) { + ts := now.Add(-time.Duration(minAgo) * time.Minute).Format(time.RFC3339) + if _, err := db.conn.Exec("INSERT INTO observer_metrics (observer_id, timestamp, noise_floor, tx_air_secs, rx_air_secs) VALUES (?, ?, ?, ?, ?)", + obs, ts, -110.0, tx, rx); err != nil { + t.Fatal(err) + } + } + // obs1: a sample 3 h ago with a far lower counter must not count; inside + // the hour it transmits 180 s in 30 min (10%) and receives 360 s (20%). + ins("obs1", 180, 0, 0) + ins("obs1", 30, 5000, 9000) + ins("obs1", 0, 5180, 9360) + // obs2: only one sample inside the hour, so no rolling figure. + ins("obs2", 120, 0, 0) + ins("obs2", 5, 900, 900) + // obs3: a full hour at 10%, but silent for the last 40 minutes. + ins("obs3", 60, 0, 0) + ins("obs3", 40, 120, 120) + + summary, err := db.GetMetricsSummary(now.Add(-24*time.Hour).Format(time.RFC3339), now) + if err != nil { + t.Fatal(err) + } + byID := map[string]MetricsSummaryRow{} + for _, r := range summary { + byID[r.ObserverID] = r + } + if got := pctOrNil(byID["obs1"].TxAirtimePct1h); got != 10.0 { + t.Errorf("obs1 tx 1h = %v, want 10", got) + } + if got := pctOrNil(byID["obs1"].RxAirtimePct1h); got != 20.0 { + t.Errorf("obs1 rx 1h = %v, want 20", got) + } + if got := pctOrNil(byID["obs2"].TxAirtimePct1h); got != nil { + t.Errorf("obs2 tx 1h = %v, want nil", got) + } + if _, ok := byID["obs3"]; !ok { + t.Fatal("obs3 missing from the summary") + } + if got := pctOrNil(byID["obs3"].TxAirtimePct1h); got != nil { + t.Errorf("obs3 (stale) tx 1h = %v, want nil", got) + } +} + +func TestMetricsSummaryDutyCycleWarnPct(t *testing.T) { + srv, router := setupTestServer(t) + get := func() float64 { + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/api/observers/metrics/summary?window=1h", nil)) + if w.Code != 200 { + t.Fatalf("status %d: %s", w.Code, w.Body.String()) + } + var resp MetricsSummaryResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if resp.Observers == nil { + t.Fatal("observers must be [] not null") + } + return resp.DutyCycleWarnPct + } + if got := get(); got != 10 { + t.Errorf("default duty-cycle warn = %v, want 10", got) + } + srv.cfg.DutyCycleWarnPct = 1 + if got := get(); got != 1 { + t.Errorf("configured duty-cycle warn = %v, want 1", got) + } +} diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 3a6b1e7b..401a9d61 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -205,7 +205,7 @@ func routeDescriptions() map[string]routeMeta { "GET /api/observers/{id}": {Summary: "Get observer detail", Tag: "observers"}, "GET /api/observers/{id}/metrics": {Summary: "Get observer metrics", Description: "Packet rates, uptime, and performance metrics.", Tag: "observers"}, "GET /api/observers/{id}/analytics": {Summary: "Get observer analytics", Tag: "observers"}, - "GET /api/observers/metrics/summary": {Summary: "Observer metrics summary", Description: "Aggregate metrics across all observers.", Tag: "observers"}, + "GET /api/observers/metrics/summary": {Summary: "Observer metrics summary", Description: "Aggregate metrics across all observers within window (noise floor, battery, sparkline). Each row also carries tx_airtime_pct_1h / rx_airtime_pct_1h: the airtime share over the last hour regardless of window, null with under 30 minutes of samples since the last reboot or no sample in the last 15 minutes. duty_cycle_warn_pct is the configured warning threshold (dutyCycleWarnPct, default 10).", Tag: "observers", QueryParams: []paramMeta{{Name: "window", Description: "Lookback such as 1h, 24h, 7d (default 24h)", Type: "string"}, {Name: "region", Description: "IATA code to filter observers", Type: "string"}}}, // Misc "GET /api/resolve-hops": {Summary: "Resolve hop path", Description: "Resolves hash prefixes in a hop path to node names. Returns affinity scores and best candidates.", Tag: "nodes", QueryParams: []paramMeta{{Name: "hops", Description: "Comma-separated hop hash prefixes", Type: "string", Required: true}}}, diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 4c16bad2..608e95d5 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -3692,8 +3692,9 @@ func (s *Server) handleMetricsSummary(w http.ResponseWriter, r *http.Request) { return } - since := time.Now().UTC().Add(-dur).Format(time.RFC3339) - summary, err := s.db.GetMetricsSummary(since) + now := time.Now().UTC() + since := now.Add(-dur).Format(time.RFC3339) + summary, err := s.db.GetMetricsSummary(since, now) if err != nil { writeError(w, 500, err.Error()) return @@ -3713,8 +3714,9 @@ func (s *Server) handleMetricsSummary(w http.ResponseWriter, r *http.Request) { summary = filtered } - writeJSON(w, map[string]interface{}{ - "observers": summary, + writeJSON(w, MetricsSummaryResponse{ + Observers: summary, + DutyCycleWarnPct: s.cfg.DutyCycleWarnPctOrDefault(), }) } diff --git a/config.example.json b/config.example.json index daa676f1..74809dee 100644 --- a/config.example.json +++ b/config.example.json @@ -369,6 +369,8 @@ "criticalMv": 3000, "_comment": "Voltage cutoffs (millivolts) for the per-node battery trend chart on /node-analytics. Latest sample below lowMv shows the node as â ī¸ Low; below criticalMv shows đĒĢ Critical. Both default to 3300 / 3000 if omitted. Source data: observer_metrics.battery_mv populated from observer status messages; only nodes that are themselves observers (matching pubkey â observer id) yield a series. Issue #663." }, + "dutyCycleWarnPct": 10, + "_comment_dutyCycleWarnPct": "TX airtime share (percent) over the last hour at which RF Health and the observer list flag an observer. Default 10, the EU 868 MHz g3 sub-band limit. Repeater firmware defaults to a 50% budget (airtime_factor 1.0), so the device does not enforce this itself. Issue #661.", "_comment_mqttSources": "Each source connects to an MQTT broker. Supported schemes: mqtt:// (plain TCP), mqtts:// (TLS), ws:// (WebSocket), wss:// (WebSocket TLS). topics: what to subscribe to. iataFilter: only ingest packets from these regions (optional). region: default IATA region for this source â used when packet/topic doesn't specify one (optional, priority: payload > topic > this field).", "compression": { "gzip": false, diff --git a/public/analytics.js b/public/analytics.js index 7851bdb2..1bfb12b8 100644 --- a/public/analytics.js +++ b/public/analytics.js @@ -4330,6 +4330,7 @@ function destroy() { _stopRolesRefresh(); _stopScopesRefresh(); _stopDistanceRet avg: ${avgNf} max: ${maxNf} ${obs.sample_count} samples + TX 1h: ${ObserverAirtime.badge(obs.tx_airtime_pct_1h, summaryData.duty_cycle_warn_pct)} `; }).join(''); diff --git a/public/index.html b/public/index.html index a6c417e2..b845beaf 100644 --- a/public/index.html +++ b/public/index.html @@ -225,6 +225,7 @@ + diff --git a/public/observer-airtime.js b/public/observer-airtime.js new file mode 100644 index 00000000..90eaf004 --- /dev/null +++ b/public/observer-airtime.js @@ -0,0 +1,36 @@ +/* #661 â rolling 1 h TX airtime share per observer, shared by the RF Health + * tiles (analytics.js) and the observer list (observers.js). The value and + * the warning threshold come from /api/observers/metrics/summary + * (tx_airtime_pct_1h, duty_cycle_warn_pct). */ +window.ObserverAirtime = (function () { + 'use strict'; + var DEFAULT_WARN_PCT = 10; + + function warnPct(w) { + return w > 0 ? w : DEFAULT_WARN_PCT; + } + + function format(pct) { + if (pct == null) return 'â'; + var n = Number(pct); + if (n > 0 && n < 0.1) return '<0.1%'; + return n.toFixed(1) + '%'; + } + + // Compared at the one decimal format() shows, so "10.0%" is never unflagged. + function isOver(pct, w) { + return pct != null && Math.round(Number(pct) * 10) / 10 >= warnPct(w); + } + + function badge(pct, w) { + if (pct == null) { + return 'â'; + } + var over = isOver(pct, w); + var title = 'TX airtime over the last hour. Duty-cycle warning at ' + warnPct(w) + '%'; + return '' + + format(pct).replace('<', '<') + ''; + } + + return { format: format, isOver: isOver, badge: badge }; +})(); diff --git a/public/observers.js b/public/observers.js index 90f3c607..7caf780a 100644 --- a/public/observers.js +++ b/public/observers.js @@ -140,6 +140,8 @@ window.preserveCompareSelection = function preserveCompareSelection(prevIds, tbo let _fetchedAt = 0; // #1562: ms epoch when the current `observers` payload was received let _loadObserversReqId = 0; // #1563: monotonic id; resolutions older than the latest are discarded let obsSkewMap = {}; // observerID â {offsetSec, samples} + let obsAirtimeMap = {}; // #661 observerID â tx_airtime_pct_1h + let obsAirtimeWarnPct; // #661 duty_cycle_warn_pct from the summary let wsHandler = null; let refreshTimer = null; let regionChangeHandler = null; @@ -255,6 +257,7 @@ window.preserveCompareSelection = function preserveCompareSelection(prevIds, tbo _mqttPanelTeardown = null; observers = []; obsSkewMap = {}; + obsAirtimeMap = {}; } async function loadObservers(opts) { @@ -265,13 +268,20 @@ window.preserveCompareSelection = function preserveCompareSelection(prevIds, tbo // (or vice versa) with stale data + a misleading "0s ago" pill. var myId = ++_loadObserversReqId; try { - const [data, skewData] = await Promise.all([ + const [data, skewData, airData] = await Promise.all([ api('/observers', { ttl: CLIENT_TTL.observers, bust: bust }), - api('/observers/clock-skew', { ttl: 30000 }).catch(function() { return []; }) + api('/observers/clock-skew', { ttl: 30000 }).catch(function() { return []; }), + // #661 â rolling 1 h TX airtime; one bulk call, not per observer. + api('/observers/metrics/summary?window=1h', { ttl: 30000 }).catch(function() { return {}; }) ]); if (myId !== _loadObserversReqId) return; // stale resolve, newer in-flight observers = data.observers || []; _fetchedAt = Date.now(); // #1562: stamp freshness for the header label + obsAirtimeMap = {}; + obsAirtimeWarnPct = airData && airData.duty_cycle_warn_pct; + ((airData && airData.observers) || []).forEach(function(a) { + if (a && a.observer_id) obsAirtimeMap[a.observer_id] = a.tx_airtime_pct_1h; + }); obsSkewMap = {}; (Array.isArray(skewData) ? skewData : []).forEach(function(s) { if (s && s.observerID) obsSkewMap[s.observerID] = s; @@ -384,7 +394,7 @@ window.preserveCompareSelection = function preserveCompareSelection(prevIds, tbo