mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-08-08 18:02:46 +02:00
data,web: mesh activity reporting & announcements
This commit is contained in:
l5y
+154
@@ -3979,3 +3979,157 @@ survive the swatch-shape change), **FU-A2** (the region toggle) and **UX-A9**
|
||||
(chat hanging indent — the `overflow-wrap` addition rides the same D-033 rule),
|
||||
and the `buildRoleButtons` filter specs (swatch/dataset/compound-key behaviour
|
||||
unchanged). No Ruby/Python/Rust/Flutter surface is touched.
|
||||
|
||||
---
|
||||
|
||||
## Feature: Mesh activity reporting & announcements
|
||||
|
||||
Maps to SPEC decisions **MA1–MA10**. Each ingestor counts **every** frame it
|
||||
handles (all RX, incl. ignored/errored/unimplemented, plus its own TX) as one
|
||||
merged `packets` figure, appends the per-interval delta to its hourly
|
||||
`POST /api/ingestors` heartbeat (MA1–MA2); the web app persists a per-ingestor
|
||||
`ingestor_activity` time-series (MA3) from which `GET /api/stats` derives a
|
||||
`MAX`-per-protocol 24 h packets/hour moving average (MA4–MA5). Each ingestor then
|
||||
periodically broadcasts a one-line activity summary — numbers **dogfed from the
|
||||
target instance's own API** — on its protocol's default channel (MA6), gated by
|
||||
`RX_ONLY` (reused as the transmit gate), the target's `/version` privacy flag
|
||||
(fail-closed), and a ≥ 24 h post-start delay (MA7–MA8), via an **optional**
|
||||
duck-typed provider send
|
||||
that leaves `MeshProtocol` conformance intact (MA9). Unless a check says
|
||||
otherwise, start the server in **public** mode
|
||||
(`API_TOKEN=acctest PRIVATE=0 FEDERATION=0 bundle exec ruby app.rb`).
|
||||
|
||||
### MA-A1 — Every frame is counted, including drops; TX too — MA1
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_activity_unit.py -k "count" )
|
||||
```
|
||||
**Expected:** pass. The merged `packets` counter increments once per received
|
||||
frame at the earliest seam (`handlers/_state._mark_packet_seen`) — verified for a
|
||||
**stored** packet, an **ignored** packet (`unsupported-port` / `no-message-payload`),
|
||||
and an **errored** packet (one that raises inside `store_packet_dict`) — and once
|
||||
per ingestor **transmission** (the announcement send and a MeshCore
|
||||
telemetry/status poll). The count is taken *before* any drop/dispatch decision, so
|
||||
no receive or transmit path bypasses it ("we don't want to under-report").
|
||||
|
||||
### MA-A2 — Heartbeat carries a per-interval delta that resets — MA2
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_activity_unit.py -k "heartbeat_delta" )
|
||||
( . .venv/bin/activate && pytest -q tests/test_mesh.py -k "ingestor" )
|
||||
```
|
||||
**Expected:** pass. `queue_ingestor_heartbeat` includes `packets` = frames counted
|
||||
since the previous heartbeat and **zeroes** the running counter afterward: two
|
||||
heartbeats bracketing N then M frames report N then M (never N then N+M). A
|
||||
heartbeat with no traffic sends `0` (or omits the field). A pre-feature payload
|
||||
without `packets` is still accepted by `POST /api/ingestors` (additive, D8).
|
||||
|
||||
### MA-A3 — Activity time-series: append per heartbeat, pruned by retention — MA3
|
||||
```bash
|
||||
( cd web && bundle exec rspec spec/activity_spec.rb )
|
||||
( cd web && bundle exec rspec spec/retention_spec.rb -e "ingestor_activity" )
|
||||
git grep -nE 'ingestor_activity' -- data/ingestor_activity.sql web/lib
|
||||
```
|
||||
**Expected:** pass, and the grep shows a new **append-only** `ingestor_activity`
|
||||
table (`ingestor_id`, `at`, `packets`, `protocol`, index on `at`). Each
|
||||
`POST /api/ingestors` carrying `packets` appends exactly one row (the `ingestors`
|
||||
snapshot row is still upserted, one per node); two ingestors of one protocol write
|
||||
**independent** rows (not pre-summed). The retention worker deletes activity rows
|
||||
older than the configured window (≥ 24 h), so the table cannot grow unbounded.
|
||||
|
||||
### MA-A4 — MAX-per-protocol packets/hour over 24 h — MA4
|
||||
```bash
|
||||
( cd web && bundle exec rspec spec/queries_spec.rb -e "packets_per_hour" )
|
||||
```
|
||||
**Expected:** pass. With two `meshcore` ingestors reporting different 24 h packet
|
||||
totals, `packets_per_hour.meshcore` = `MAX(total_A, total_B) ÷ 24` — the busiest
|
||||
single vantage, so the quieter ingestor and any overlap never inflate it. `total`
|
||||
is the `MAX` across **all** ingestors regardless of protocol; a protocol with no
|
||||
active ingestor reads `0`. Rows older than 24 h do not contribute.
|
||||
|
||||
### MA-A5 — `/api/stats` exposes `packets_per_hour` additively — MA5
|
||||
```bash
|
||||
curl -s http://127.0.0.1:41447/api/stats \
|
||||
| python3 -c 'import sys,json; d=json.load(sys.stdin); p=d["packets_per_hour"]; \
|
||||
SC=("total","meshcore","meshtastic","reticulum"); \
|
||||
print(all(isinstance(p[s],(int,float)) for s in SC) and p["reticulum"]==0 \
|
||||
and set(SC).issubset(d) and all(m in d["total"] for m in ("nodes","messages","telemetry")))'
|
||||
```
|
||||
**Expected:** prints `True` — the response carries an additive top-level
|
||||
`packets_per_hour` map keyed `{total, meshcore, meshtastic, reticulum}`
|
||||
(`reticulum` a `0` stub), **and** the pre-existing scope × metric × window tree
|
||||
(S1: each scope still carrying `nodes`/`messages`/`telemetry`) is unchanged and
|
||||
still present. No version bump — **S-A1** still passes.
|
||||
|
||||
### MA-A6 — Announcement content is dogfed from the instance API — MA6
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_announce_unit.py -k "message or dogfeed" )
|
||||
```
|
||||
**Expected:** pass. The announcement string is exactly
|
||||
`"<Protocol> activity in the last 24h: <N> active nodes, <M> packets/hour. https://<domain>"`,
|
||||
where `<N>` = the target's `GET /api/stats` `<protocol>.nodes.day` and `<M>` =
|
||||
`GET /api/stats` `packets_per_hour.<protocol>` — both fetched over HTTP from
|
||||
`<domain>`, never computed from the ingestor's local counters — and the rendered
|
||||
line is truncated to the protocol's character limit. `<domain>` = the configured
|
||||
`INSTANCE_DOMAIN`.
|
||||
|
||||
### MA-A7 — Announcement gates: RX_ONLY, privacy fail-closed, 24 h — MA7
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \
|
||||
-k "gate or private or rx_only or elapsed" )
|
||||
```
|
||||
**Expected:** pass. **No** announcement is sent when any gate fails: `RX_ONLY=1`
|
||||
(the **reused** receive-only flag — its existing MeshCore-poll TX suppression now
|
||||
also covers the announcement, so it is the sole opt-out; no separate `ENABLE_TX`
|
||||
env exists); the target `/version` reports `private_mode: true`; the `/version`
|
||||
fetch **errors or is unparseable** (fail-closed — treated as private/skip); or
|
||||
`< 24 h` have elapsed since ingestor start. With `RX_ONLY` unset (the default
|
||||
`0`), `private_mode: false`, and `≥ 24 h` elapsed, exactly one announcement per
|
||||
24 h per domain is transmitted.
|
||||
|
||||
### MA-A8 — Default channel/scope + 24 h cadence — MA8
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \
|
||||
-k "channel or cadence or interval or domains" )
|
||||
```
|
||||
**Expected:** pass. The announcement is sent on Meshtastic channel `CHANNEL_INDEX`
|
||||
(default `0`) / MeshCore's public channel; the first fires no earlier than 24 h
|
||||
post-start and subsequent ones no more often than every 24 h; an ingestor with
|
||||
several `INSTANCE_DOMAIN` targets announces each once per cycle with that domain's
|
||||
own numbers and link.
|
||||
|
||||
### MA-A9 — Send is optional and duck-typed; MeshProtocol conformance intact — MA9
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/test_provider_unit.py \
|
||||
-k "send_channel_announcement or MeshProtocol" )
|
||||
```
|
||||
**Expected:** pass. Both `MeshtasticProvider` and `MeshcoreProvider` expose
|
||||
`send_channel_announcement(...)`; the announce scheduler resolves it via
|
||||
`getattr(provider, "send_channel_announcement", None)` and **no-ops when absent**
|
||||
(e.g. a receive-only transport). The `@runtime_checkable MeshProtocol` interface is
|
||||
unchanged — **A4b**'s `isinstance(provider, MeshProtocol)` conformance still passes
|
||||
and `send_channel_announcement` is **not** a required member.
|
||||
|
||||
### MA-A10 — Additive contract is documented — MA10 / D8
|
||||
```bash
|
||||
git grep -nE 'packets|ingestor_activity|packets_per_hour' -- data/mesh_ingestor/CONTRACTS.md
|
||||
```
|
||||
**Expected:** the additive heartbeat `packets` field, the `ingestor_activity`
|
||||
schema, and the `GET /api/stats` `packets_per_hour` addition are all documented in
|
||||
`CONTRACTS.md` (Layer C source of truth). The engineering bar (100 % tests/docs/
|
||||
headers/lint) is enforced by Layer **B** (B1–B5); behavior is covered by
|
||||
MA-A1…MA-A9.
|
||||
|
||||
### MA-R1 — Regression: prior acceptance still holds
|
||||
```bash
|
||||
( . .venv/bin/activate && pytest -q tests/ )
|
||||
( cd web && bundle exec rspec ) && ( cd web && npm test )
|
||||
```
|
||||
**Expected:** every prior check still passes. At risk and explicitly required to
|
||||
remain green: **A1a/A1b** (apex — the new ingestor→instance GET and the LoRa
|
||||
announcement add no broker term or dependency); **A4b** (MeshProtocol isinstance
|
||||
conformance — send stays optional/duck-typed, MA9); **S-A1** (the `/api/stats`
|
||||
scope × metric × window tree is unchanged; `packets_per_hour` is an additive
|
||||
sibling — no version bump, so `test_version_sync.py` is unaffected); **A2a/A2b**
|
||||
(privacy — the message API still 404s under `PRIVATE`, and the announcement
|
||||
fail-closes on the same flag, MA7); **C2** (`tests/test_mesh.py` — the
|
||||
`POST /api/ingestors` `packets` field is additive and old payloads still validate);
|
||||
and **B1** (all suites). No `/api/*` response shape is broken by construction.
|
||||
|
||||
@@ -1045,3 +1045,61 @@ contradicted.
|
||||
| **LC2** | **Legend pressed state paints (specificity fix).** A role chip is a `<button>`; the reset `button:not(.chat-tab):not(.sort-button)` (0,2,1) outranked the bare `.legend-item[aria-pressed="true"]` (0,2,0), so every chip computed to `#333` and only `font-weight` distinguished selected from filtered. The selector is now `button.legend-item[aria-pressed="true"]` (0,2,1), which wins on source order — the selected blue paints. | review |
|
||||
| **LC3** | **Chat log never scrolls horizontally (extends UX11/D-033).** `.chat-tabpanel` is `overflow-y: auto`, so its x-axis is a scroll container; nothing broke long tokens. `overflow-wrap: anywhere` on `.chat-entry-msg, .chat-entry-node` wraps unbreakable tokens under the 19ch hang and lowers the entry's min-content width so it can never exceed the panel. `anywhere`, not `break-word`; no `overflow-x: hidden` (which would only mask the next regression). | review |
|
||||
| **LC4** | **Engineering bar (D9).** Each fix shipped with a fail-first check (Phase 2), full JSDoc/comments, the exact Apache header, clean linters; all prior suites stay green; the `buildRoleButtons` filter specs gain the swatch-shape assertion, none removed. | CLAUDE.md |
|
||||
|
||||
---
|
||||
|
||||
## Feature: Mesh activity reporting & announcements
|
||||
|
||||
Turns each ingestor from a pure listener into a reporter **and** an announcer.
|
||||
(1) **Reporting:** every ingestor counts *every* frame it handles — all received
|
||||
frames (including ignored / errored / unimplemented) plus its own transmissions —
|
||||
as one merged `packets` figure and appends the per-interval delta to its hourly
|
||||
`POST /api/ingestors` heartbeat; the web app persists a per-ingestor activity
|
||||
time-series so a **packets/hour moving average** is computable across
|
||||
time × protocol × multiple ingestors. (2) **Announcing** ("we stop listening and
|
||||
start talking"): each ingestor periodically broadcasts a one-line activity summary
|
||||
on its protocol's default channel, drawing the numbers **back from the target
|
||||
instance's own API** (dogfeeding, so one radio's partial view never
|
||||
under-represents the mesh). Integrates with
|
||||
`data/mesh_ingestor/handlers/_state.py` (RX/TX counting seam), `ingestors.py`
|
||||
(heartbeat payload), `config.py` (reuse `RX_ONLY`), `daemon.py`
|
||||
(announce schedule), `mesh_protocol.py` + `protocols/{meshtastic,meshcore,meshtastic_udp}`
|
||||
(optional send), a new announce + instance-stats-GET module, `data/ingestors.sql`
|
||||
plus a new `data/ingestor_activity.sql` table,
|
||||
`web/lib/potato_mesh/application/routes/{ingest,api}.rb`, the query + retention
|
||||
layers, and `data/mesh_ingestor/CONTRACTS.md`.
|
||||
|
||||
**Conflict check against existing decisions.** *Invariant I (apex / local-LoRa)*
|
||||
— **consistent, extends §4.2**: the announcement is a LoRa transmission on the
|
||||
local aether and the dogfeed is an HTTP GET to the ingestor's *own* instance; no
|
||||
MQTT/broker/cloud dependency or connection is added (A1 stays clean), and ingestor
|
||||
TX already exists (`RX_ONLY`, MeshCore telemetry polls), so this widens an existing
|
||||
capability rather than crossing the apex line. *Invariant II (privacy)* —
|
||||
**extends**: a new gate suppresses the announcement unless the target instance
|
||||
reports non-private, read authoritatively from its `/version` and **fail-closed**
|
||||
on any fetch error; only public aggregates (active nodes, packet rate) are ever
|
||||
announced. *Invariant III (federation) & §5 (no analytics / phone-home)* —
|
||||
**consistent, new authorized behavior**: the announcement publishes an *unsigned
|
||||
public activity line on the community's local mesh* — not the signed federation
|
||||
wire, not cloud egress, not third-party analytics — and is opt-out via
|
||||
`RX_ONLY=1`. *Invariant IV (parity & pluggability)* — **extends**: sending is an
|
||||
*optional, duck-typed* provider capability, not a new formal `MeshProtocol` member,
|
||||
so the `A4b` conformance contract and every existing provider are untouched; both
|
||||
protocols announce identically. *D8 (stable contract)* — **extends**: the heartbeat
|
||||
`packets` field, the `/api/stats` `packets_per_hour` map, and the activity table
|
||||
are all additive; no version bump. *§3.3 (web is POST-only intake)* —
|
||||
**consistent**: the dogfeed is a read; no new ingest path. No invariant is
|
||||
contradicted.
|
||||
|
||||
| # | Decision | Source |
|
||||
| --- | --- | --- |
|
||||
| **MA1** | **Merged packet counter — count everything, at the earliest seam.** Each ingestor maintains one merged `packets` counter incremented on **every received frame** — *all of them, including ignored / errored / unimplemented / unsupported-port packets* — counted at the earliest common receive seam (`handlers/_state._mark_packet_seen`, already invoked by both the Meshtastic `on_receive` path and every MeshCore handler/telemetry path *before* any dispatch or filtering), **plus every ingestor-initiated transmission** (the announcement itself and the existing MeshCore telemetry/status polls). RX and TX are deliberately **merged** into a single figure (no separate breakdown). "We don't want to under-report" is satisfied at source: counting precedes every drop/ignore decision. A build-phase check confirms each protocol's RX entry funnels through the seam so no family is missed. | interview |
|
||||
| **MA2** | **Heartbeat carries the per-interval delta.** `POST /api/ingestors` gains an additive optional `packets` field = frames counted since the previous heartbeat (a per-interval **delta**, not a since-boot cumulative), reset to zero each time a heartbeat is queued. Per-interval deltas map one-to-one onto time-series rows and are restart-safe (a reboot simply starts a fresh interval). Absent ⇒ treated as `0`, so pre-feature ingestors and the existing `tests/test_mesh.py` fixtures are unaffected (D8). | interview |
|
||||
| **MA3** | **Per-ingestor activity time-series (the moving-average schema).** A new append-only table `ingestor_activity` (`ingestor_id TEXT`, `at INTEGER`, `packets INTEGER`, `protocol TEXT`, indexed on `at`) records one row per heartbeat delta. This is the schema that makes a packets/hour moving average computable while distinguishing **multiple protocols and multiple ingestors per protocol** (each ingestor's contribution is kept separate rather than pre-summed). The `ingestors` snapshot table (one row per node) is unchanged. Rows are pruned by the existing retention worker (window ≥ the 24 h the announcement needs; sized with the other activity floors). | proposed |
|
||||
| **MA4** | **Aggregation = MAX per protocol (dedup-free "in the air").** The mesh-wide packets/hour for a protocol is `MAX` over that protocol's ingestors of *(that ingestor's total `packets` reported in the last 24 h ÷ 24)*. A single radio can only hear ≤ what is actually transmitted, so the busiest single vantage is the best dedup-free estimate of unique air traffic and can never double-count a frame heard by two radios (true per-frame dedup is impossible anyway — ignored/errored frames carry no id). The fixed `÷ 24` denominator (not ÷ elapsed) keeps the rate stable and avoids divide-by-small spikes; an ingestor with < 24 h of data simply reads lower and ramps up (moot in practice — the announcement fires only ≥ 24 h after start, MA7). *Accepted limitation:* MAX under-estimates the true union when different radios are busiest in different hours. | interview |
|
||||
| **MA5** | **Exposure via an additive `/api/stats` field.** `GET /api/stats` gains a top-level additive `packets_per_hour` map — `{ total, meshcore, meshtastic, reticulum }` — carrying the MA4 24 h MAX-aggregated moving average (`reticulum` a forward-looking `0` stub, consistent with S6; `total` is the MAX across all ingestors regardless of protocol). The existing scope × metric × window tree (S1) is untouched, so this needs **no** version bump. The announcement's active-node figure reuses the existing `<protocol>.nodes.day` count (already deduped by `node_id`); only packets/hour is new. | proposed |
|
||||
| **MA6** | **Announcement content, drawn from the instance's own API (dogfeeding).** Each ingestor broadcasts, for its own configured protocol, a single line formatted to that protocol's character limit: `"<Protocol> activity in the last 24h: <N> active nodes, <M> packets/hour. https://<domain>"`. `<N>` and `<M>` are fetched **from the target instance** (`GET /api/stats` → `<protocol>.nodes.day` and `packets_per_hour.<protocol>`), never computed from the ingestor's local view — because one ingestor may not see the whole mesh. `<domain>` is the configured `INSTANCE_DOMAIN`. Reporting (MA1–MA4) is independent and continues regardless of announcement state. | interview |
|
||||
| **MA7** | **Announcement is triple-gated.** An announcement is transmitted only when **all** hold: (a) `RX_ONLY` is unset — the **reused** receive-only flag (default `0`) is the single transmit gate; `RX_ONLY=1` forbids *every* ingestor TX (the MeshCore polls and the announcement alike), so it is the sole opt-out and **no separate `ENABLE_TX` env is added**; (b) the target instance reports **non-private** — the ingestor GETs `<domain>/version` and honors `config.private_mode`, re-checked every cycle, and **fails closed** (skips) on any fetch/parse error, so a privacy signal is never missed (Invariant II); (c) **≥ 24 h have elapsed since ingestor start** — so the first numbers are accurate over a full window and restarts cannot spam the channel. | interview |
|
||||
| **MA8** | **Default channel/scope + 24 h cadence.** The announcement is broadcast on the protocol's **default channel and scope** — Meshtastic channel `CHANNEL_INDEX` (default `0`) via the interface text-send; MeshCore its public/default channel — honoring existing `ALLOWED_CHANNELS`/`HIDDEN_CHANNELS` intent. After the initial ≥ 24 h wait it repeats **every 24 h**, per configured instance domain (an ingestor with several `INSTANCE_DOMAIN` targets announces each with its own numbers and link). TX volume is negligible (~1 frame/day/domain). | proposed |
|
||||
| **MA9** | **Optional duck-typed provider send.** Transmission is exposed as a new **optional** provider method (e.g. `send_channel_announcement(iface, text)`) accessed via `getattr(provider, …, None)` — mirroring the existing optional `self_node_item` extension — so the `@runtime_checkable MeshProtocol` interface and its `A4b` isinstance conformance are unchanged, and a provider that cannot transmit (or a receive-only transport) simply omits it. Both `meshtastic` and `meshcore` providers implement it; neither protocol is privileged. | interview |
|
||||
| **MA10** | **Engineering bar (D9).** Every new/changed unit ships with 100 % unit tests (counter increment across RX families + TX; heartbeat delta + reset; activity-table write + retention; MAX-aggregation math incl. multi-ingestor; the `/api/stats` field; the four announcement gates incl. fail-closed privacy and the 24 h wait; per-protocol send), full PDoc/RDoc, Apache headers, `black`/`rufo` clean; `pytest`/`rspec`/`npm test` stay green; `CONTRACTS.md` documents the additive heartbeat field, the activity schema, and the `/api/stats` addition. | CLAUDE.md |
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
-- Copyright © 2025-26 l5yth & contributors
|
||||
--
|
||||
-- Licensed under the Apache License, Version 2.0 (the "License");
|
||||
-- you may not use this file except in compliance with the License.
|
||||
-- You may obtain a copy of the License at
|
||||
--
|
||||
-- http://www.apache.org/licenses/LICENSE-2.0
|
||||
--
|
||||
-- Unless required by applicable law or agreed to in writing, software
|
||||
-- distributed under the License is distributed on an "AS IS" BASIS,
|
||||
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
-- See the License for the specific language governing permissions and
|
||||
-- limitations under the License.
|
||||
|
||||
PRAGMA journal_mode=WAL;
|
||||
|
||||
-- Per-heartbeat mesh-activity time-series (SPEC MA3).
|
||||
--
|
||||
-- Append-only: one row per ingestor heartbeat that carries a `packets` count
|
||||
-- (the merged RX+TX frames observed since the previous heartbeat, MA1/MA2).
|
||||
-- Each ingestor's contribution is kept as its own rows rather than pre-summed,
|
||||
-- so a packets/hour moving average can be computed across time, protocol, and
|
||||
-- multiple ingestors per protocol (the read-side MAX aggregation, MA4). Rows
|
||||
-- are pruned by the retention worker on `at`.
|
||||
CREATE TABLE IF NOT EXISTS ingestor_activity (
|
||||
id INTEGER PRIMARY KEY,
|
||||
ingestor_id TEXT NOT NULL,
|
||||
at INTEGER NOT NULL,
|
||||
packets INTEGER NOT NULL,
|
||||
protocol TEXT NOT NULL DEFAULT 'meshtastic'
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_ingestor_activity_at ON ingestor_activity(at);
|
||||
CREATE INDEX IF NOT EXISTS idx_ingestor_activity_protocol_at
|
||||
ON ingestor_activity(protocol, at);
|
||||
+3
-1
@@ -32,4 +32,6 @@ if ! .venv/bin/python -c "import sys; exit(0 if '.venv' in sys.prefix else 1)" 2
|
||||
fi
|
||||
.venv/bin/python -m pip install -U pip
|
||||
.venv/bin/python -m pip install -r "$(dirname "$0")/requirements.txt"
|
||||
exec .venv/bin/python mesh.py
|
||||
|
||||
# -u keeps stdout unbuffered
|
||||
exec .venv/bin/python -u mesh.py
|
||||
|
||||
@@ -218,6 +218,9 @@ Heartbeat payload:
|
||||
- `version` (string)
|
||||
- Optional: `lora_freq`, `modem_preset`
|
||||
- Optional: `protocol` (string; e.g. `"meshtastic"`, `"meshcore"`) — declares the mesh backend for this ingestor; defaults to `"meshtastic"` when absent
|
||||
- Optional: `packets` (int ≥ 0) — **mesh-activity delta (SPEC MA1/MA2).** The merged count of *every* frame this ingestor handled since its previous heartbeat: all received frames (including ignored / errored / unimplemented) **plus** its own transmissions (announcement + MeshCore telemetry polls), counted at the earliest receive/transmit seam so nothing is under-reported. It is a **per-interval delta** (reset on each send), **not** a since-boot cumulative. Additive and backward-compatible: an absent or negative value records no activity, so pre-feature ingestors are unaffected.
|
||||
|
||||
**Mesh-activity time-series (SPEC MA3).** Each heartbeat carrying a non-negative `packets` value appends one **append-only** row to the `ingestor_activity` table (`ingestor_id`, `at`, `packets`, `protocol`; `data/ingestor_activity.sql`); the `ingestors` snapshot row is upserted as before. Each ingestor's contribution is stored separately (never pre-summed) so a packets/hour moving average is computable across time × protocol × multiple ingestors. The row is best-effort — a failed activity insert never sinks the liveness heartbeat (still `201`). Rows are pruned by the retention worker on `at`. The read-side aggregate is served by `GET /api/stats` (`packets_per_hour`, below).
|
||||
|
||||
**Protocol propagation**: all event records (`messages`, `positions`, `telemetry`, `traces`, `neighbors`) that reference this ingestor via their `ingestor` field inherit its `protocol` value at write time when no explicit per-record `protocol` stamp is present. Per-record stamps take precedence — the ingestor heartbeat default only kicks in when the per-record field is absent or malformed.
|
||||
|
||||
@@ -303,6 +306,7 @@ do **not** accept `before`.
|
||||
"meshcore": { "nodes": {…}, "messages": {…}, "telemetry": {…} },
|
||||
"meshtastic": { "nodes": {…}, "messages": {…}, "telemetry": {…} },
|
||||
"reticulum": { "nodes": {…}, "messages": {…}, "telemetry": {…} }, // stub: always 0
|
||||
"packets_per_hour": { "total": 50, "meshcore": 50, "meshtastic": 30, "reticulum": 0 },
|
||||
"sampled": false
|
||||
}
|
||||
```
|
||||
@@ -320,6 +324,17 @@ do **not** accept `before`.
|
||||
- **Privacy.** Every metric honors the node opt-out marker. When `PRIVATE=1`, all
|
||||
`messages` counts are forced to `0` (mirroring the disabled message API);
|
||||
`nodes`/`telemetry` counts remain.
|
||||
- **`packets_per_hour`** (additive, SPEC MA4/MA5) is a flat top-level map keyed
|
||||
`{ total, meshcore, meshtastic, reticulum }` carrying the 24-hour packets/hour
|
||||
moving average as a rounded integer. It is aggregated **MAX-per-protocol**:
|
||||
`MAX` over that protocol's ingestors of *(the ingestor's `packets` total in the
|
||||
last 24 h ÷ 24)* — a single radio hears ≤ what is actually transmitted, so the
|
||||
busiest vantage is the best dedup-free estimate of air traffic and never
|
||||
double-counts a frame heard by two radios. `total` is the same MAX over **every**
|
||||
ingestor regardless of protocol; `reticulum` is the always-zero forward-looking
|
||||
stub. Unlike `messages`, it is **not** privacy-gated (packets are a public
|
||||
aggregate, no message content). Additive to the 0.7.x `/api/stats` tree — no
|
||||
version bump; the ingestor dogfeeds it for the activity announcement (MA6).
|
||||
- **`sampled`** is unchanged: always `false` (the counts are exact, not sampled).
|
||||
|
||||
### GET /api/events live-update stream (SSE)
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Merged mesh-activity packet counter shared across the ingestor.
|
||||
|
||||
SPEC decision **MA1** (``## Feature: Mesh activity reporting & announcements``):
|
||||
the ingestor counts **every** frame it handles — all received frames (including
|
||||
ignored / errored / unimplemented) *plus* its own transmissions — as one merged
|
||||
figure so the dashboard can report "what is in the air" without under-reporting.
|
||||
|
||||
This module owns that single counter so the receive seam
|
||||
(:func:`data.mesh_ingestor.handlers._state._mark_packet_seen`), the transmit
|
||||
sites (the MeshCore telemetry/status polls and, later, the activity
|
||||
announcement), and the heartbeat sender
|
||||
(:func:`data.mesh_ingestor.ingestors.queue_ingestor_heartbeat`) all share one
|
||||
consistent view. It deliberately depends on nothing but :mod:`threading`, so it
|
||||
can be imported from any layer (handlers, protocols, ingestors) without risking
|
||||
an import cycle.
|
||||
|
||||
The counter is a *per-interval delta* (**MA2**): :func:`take_packet_count`
|
||||
returns the running total and atomically resets it to zero, so each heartbeat
|
||||
carries only the frames observed since the previous heartbeat. Access is
|
||||
serialised by a lock because frames are recorded on the mesh library's receive
|
||||
thread (or the MeshCore asyncio loop) while the daemon thread drains the count
|
||||
on its heartbeat cycle.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
_lock = threading.Lock()
|
||||
"""Serialises access to :data:`_packet_count` across the record/drain threads."""
|
||||
|
||||
_packet_count: int = 0
|
||||
"""Merged count of frames handled (RX + TX) since the last :func:`take_packet_count`."""
|
||||
|
||||
|
||||
def _add(count: int) -> None:
|
||||
"""Add *count* to the merged activity counter under the shared lock.
|
||||
|
||||
Parameters:
|
||||
count: Number of frames to add to the running total.
|
||||
"""
|
||||
|
||||
global _packet_count
|
||||
with _lock:
|
||||
_packet_count += count
|
||||
|
||||
|
||||
def record_packet(count: int = 1) -> None:
|
||||
"""Count received frame(s) toward the merged activity total (MA1).
|
||||
|
||||
Invoked from the earliest common receive seam
|
||||
(:func:`~data.mesh_ingestor.handlers._state._mark_packet_seen`) so *every*
|
||||
received frame is counted before any dispatch, filter, or drop decision —
|
||||
including ignored, errored, and unimplemented packets.
|
||||
|
||||
Parameters:
|
||||
count: Number of received frames to record; defaults to ``1``.
|
||||
"""
|
||||
|
||||
_add(count)
|
||||
|
||||
|
||||
def record_tx(count: int = 1) -> None:
|
||||
"""Count ingestor transmission(s) toward the merged activity total (MA1).
|
||||
|
||||
Invoked at each transmit site (the MeshCore telemetry/status polls and the
|
||||
activity announcement) so the ingestor's own airtime is reflected in the
|
||||
reported figure, merged with the received count.
|
||||
|
||||
Parameters:
|
||||
count: Number of transmitted frames to record; defaults to ``1``.
|
||||
"""
|
||||
|
||||
_add(count)
|
||||
|
||||
|
||||
def take_packet_count() -> int:
|
||||
"""Return the merged activity count and atomically reset it to zero.
|
||||
|
||||
This yields the *per-interval delta* (MA2): the number of frames recorded
|
||||
since the previous call. The read and the reset happen under the shared
|
||||
lock so a frame recorded concurrently is never double-counted or lost.
|
||||
|
||||
Returns:
|
||||
The number of frames (RX + TX) recorded since the last call.
|
||||
"""
|
||||
|
||||
global _packet_count
|
||||
with _lock:
|
||||
value = _packet_count
|
||||
_packet_count = 0
|
||||
return value
|
||||
|
||||
|
||||
def reset() -> None:
|
||||
"""Reset the merged activity counter to zero.
|
||||
|
||||
Provided for test isolation and for any future "clear on reconnect" need;
|
||||
unlike :func:`take_packet_count` it discards the running total without
|
||||
returning it.
|
||||
"""
|
||||
|
||||
global _packet_count
|
||||
with _lock:
|
||||
_packet_count = 0
|
||||
|
||||
|
||||
__all__ = [
|
||||
"record_packet",
|
||||
"record_tx",
|
||||
"take_packet_count",
|
||||
"reset",
|
||||
]
|
||||
@@ -0,0 +1,404 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Activity announcement: dogfeed the instance API and build the broadcast text.
|
||||
|
||||
SPEC decision **MA6** — the ingestor periodically broadcasts a one-line activity
|
||||
summary on its protocol's default channel, drawing the numbers **back from the
|
||||
target instance's own API** (dogfeeding: one radio may not see the whole mesh).
|
||||
|
||||
This module owns the read-only *dogfeed* HTTP client (GET ``/version`` and
|
||||
``/api/stats`` — both public, no auth) and the character-limited message builder.
|
||||
The transmit primitive lives on each provider (``send_channel_announcement``,
|
||||
MA9); the scheduling and the ``RX_ONLY`` / privacy / 24-hour gates live in the
|
||||
daemon (MA7/MA8). Every fetch fails **soft** — any network or
|
||||
shape error returns ``None`` so the caller can fail closed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
from . import config
|
||||
|
||||
#: Conservative single-frame text limits per protocol (SPEC MA6). ASCII bytes ≈
|
||||
#: characters; the announcement template is short, so truncation is only a
|
||||
#: safety net for an unusually long instance domain.
|
||||
ANNOUNCE_CHAR_LIMITS = {"meshtastic": 200, "meshcore": 140}
|
||||
|
||||
#: Fallback character limit for an unrecognised protocol.
|
||||
_DEFAULT_CHAR_LIMIT = 140
|
||||
|
||||
#: Human-facing protocol labels used in the announcement line.
|
||||
_PROTOCOL_DISPLAY_NAMES = {"meshtastic": "Meshtastic", "meshcore": "MeshCore"}
|
||||
|
||||
#: Timeout (seconds) for each dogfeed GET.
|
||||
_DOGFEED_TIMEOUT_SECS = 10
|
||||
|
||||
#: Browser-like headers mirroring :mod:`~data.mesh_ingestor.queue` so an instance
|
||||
#: behind Cloudflare does not block the dogfeed request.
|
||||
_HTTP_HEADERS = {
|
||||
"User-Agent": (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||||
),
|
||||
"Accept": "application/json",
|
||||
"Accept-Language": "en-US,en;q=0.9",
|
||||
}
|
||||
|
||||
|
||||
def _get_json(url: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS) -> dict | None:
|
||||
"""GET *url* and parse a JSON object.
|
||||
|
||||
Parameters:
|
||||
url: Absolute URL to fetch.
|
||||
timeout: Socket timeout in seconds.
|
||||
|
||||
Returns:
|
||||
The decoded mapping, or ``None`` on any network / decode error or when
|
||||
the payload is not a JSON object.
|
||||
"""
|
||||
|
||||
req = urllib.request.Request(url, headers=dict(_HTTP_HEADERS))
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
payload = json.loads(resp.read().decode("utf-8"))
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"dogfeed GET failed",
|
||||
context="announce.get",
|
||||
severity="warn",
|
||||
url=url,
|
||||
error_class=exc.__class__.__name__,
|
||||
error_message=str(exc),
|
||||
)
|
||||
return None
|
||||
return payload if isinstance(payload, dict) else None
|
||||
|
||||
|
||||
def fetch_private_mode(
|
||||
instance_url: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS
|
||||
) -> bool | None:
|
||||
"""Return the target instance's ``private_mode`` flag from ``GET /version``.
|
||||
|
||||
Parameters:
|
||||
instance_url: Base URL of the PotatoMesh instance (e.g. ``https://mesh``).
|
||||
timeout: Socket timeout in seconds.
|
||||
|
||||
Returns:
|
||||
``True`` / ``False`` for a well-formed response, or ``None`` when the
|
||||
fetch fails or the flag is absent/malformed. The caller treats ``None``
|
||||
as "assume private, skip" (fail-closed — SPEC MA7 / Invariant II).
|
||||
"""
|
||||
|
||||
data = _get_json(f"{instance_url}/version", timeout=timeout)
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
config_block = data.get("config")
|
||||
if not isinstance(config_block, dict):
|
||||
return None
|
||||
value = config_block.get("private_mode")
|
||||
return value if isinstance(value, bool) else None
|
||||
|
||||
|
||||
def fetch_activity(
|
||||
instance_url: str, protocol: str, *, timeout: float = _DOGFEED_TIMEOUT_SECS
|
||||
) -> tuple[int, float | int] | None:
|
||||
"""Return ``(active_nodes, packets_per_hour)`` for *protocol* from ``/api/stats``.
|
||||
|
||||
These are the mesh-wide numbers the announcement quotes (SPEC MA6):
|
||||
``active_nodes`` = ``<protocol>.nodes.day`` and ``packets_per_hour`` =
|
||||
``packets_per_hour.<protocol>``.
|
||||
|
||||
Parameters:
|
||||
instance_url: Base URL of the PotatoMesh instance.
|
||||
protocol: ``"meshtastic"`` or ``"meshcore"``.
|
||||
timeout: Socket timeout in seconds.
|
||||
|
||||
Returns:
|
||||
A ``(active_nodes, packets_per_hour)`` tuple, or ``None`` on any
|
||||
fetch/parse error or malformed shape.
|
||||
"""
|
||||
|
||||
data = _get_json(f"{instance_url}/api/stats", timeout=timeout)
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
scope = data.get(protocol)
|
||||
packets_map = data.get("packets_per_hour")
|
||||
if not isinstance(scope, dict) or not isinstance(packets_map, dict):
|
||||
return None
|
||||
nodes = scope.get("nodes")
|
||||
if not isinstance(nodes, dict):
|
||||
return None
|
||||
active_nodes = nodes.get("day")
|
||||
packets = packets_map.get(protocol)
|
||||
if not isinstance(active_nodes, int) or not isinstance(packets, (int, float)):
|
||||
return None
|
||||
return int(active_nodes), packets
|
||||
|
||||
|
||||
def protocol_display_name(protocol: str | None) -> str:
|
||||
"""Return the human-facing protocol label used in the announcement.
|
||||
|
||||
Parameters:
|
||||
protocol: Protocol key (e.g. ``"meshcore"``).
|
||||
|
||||
Returns:
|
||||
A display label (``"MeshCore"``), the capitalised key for an unknown
|
||||
protocol, or ``""`` when *protocol* is empty.
|
||||
"""
|
||||
|
||||
if not protocol:
|
||||
return ""
|
||||
return _PROTOCOL_DISPLAY_NAMES.get(protocol, protocol.capitalize())
|
||||
|
||||
|
||||
def build_announcement(
|
||||
protocol: str,
|
||||
active_nodes: int,
|
||||
packets_per_hour: float | int,
|
||||
instance_url: str,
|
||||
*,
|
||||
char_limit: int | None = None,
|
||||
) -> str:
|
||||
"""Build the activity announcement line (SPEC MA6).
|
||||
|
||||
Format: ``"<Protocol> activity in the last 24h: <N> active nodes, <M>
|
||||
packets/hour. <instance_url>"``, truncated to the protocol's character limit.
|
||||
|
||||
Parameters:
|
||||
protocol: ``"meshtastic"`` or ``"meshcore"``.
|
||||
active_nodes: 24-hour active-node count for *protocol*.
|
||||
packets_per_hour: 24-hour packets/hour moving average for *protocol*.
|
||||
instance_url: Base URL of the instance (already carries the scheme).
|
||||
char_limit: Optional explicit limit; defaults to the protocol's entry in
|
||||
:data:`ANNOUNCE_CHAR_LIMITS`.
|
||||
|
||||
Returns:
|
||||
The announcement string, truncated to fit the character limit.
|
||||
"""
|
||||
|
||||
text = (
|
||||
f"{protocol_display_name(protocol)} activity in the last 24h: "
|
||||
f"{active_nodes} active nodes, {packets_per_hour} packets/hour. "
|
||||
f"{instance_url}"
|
||||
)
|
||||
limit = (
|
||||
char_limit
|
||||
if char_limit is not None
|
||||
else ANNOUNCE_CHAR_LIMITS.get(protocol, _DEFAULT_CHAR_LIMIT)
|
||||
)
|
||||
if limit is not None and len(text) > limit:
|
||||
text = text[:limit]
|
||||
return text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scheduling & gating (SPEC MA7/MA8)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: The first announcement is withheld until the ingestor has been running this
|
||||
#: long, so the dogfed 24-hour numbers are accurate over a full window and a
|
||||
#: restart storm cannot spam the channel (SPEC MA7 d).
|
||||
ANNOUNCE_INITIAL_DELAY_SECS = 24 * 60 * 60
|
||||
|
||||
#: Minimum spacing between announcement cycles once eligible (SPEC MA8).
|
||||
ANNOUNCE_INTERVAL_SECS = 24 * 60 * 60
|
||||
|
||||
|
||||
def announcements_enabled() -> bool:
|
||||
"""Return whether announcements may be transmitted at all (SPEC MA7 a).
|
||||
|
||||
``True`` only when :data:`~data.mesh_ingestor.config.RX_ONLY` is off — the
|
||||
reused receive-only flag (default off) is the single transmit gate; there is
|
||||
no separate enable switch.
|
||||
|
||||
Returns:
|
||||
``True`` when the transmit gate permits an announcement.
|
||||
"""
|
||||
|
||||
return not getattr(config, "RX_ONLY", False)
|
||||
|
||||
|
||||
def announce_due(*, start_time: float, last_announce: float | None, now: float) -> bool:
|
||||
"""Return whether an announcement cycle is due (SPEC MA7 d / MA8).
|
||||
|
||||
Parameters:
|
||||
start_time: Unix time the ingestor started.
|
||||
last_announce: Unix time of the previous cycle, or ``None`` if none yet.
|
||||
now: Current unix time.
|
||||
|
||||
Returns:
|
||||
``True`` when at least :data:`ANNOUNCE_INITIAL_DELAY_SECS` have elapsed
|
||||
since *start_time* **and** at least :data:`ANNOUNCE_INTERVAL_SECS` since
|
||||
*last_announce*.
|
||||
"""
|
||||
|
||||
if now - start_time < ANNOUNCE_INITIAL_DELAY_SECS:
|
||||
return False
|
||||
if last_announce is not None and now - last_announce < ANNOUNCE_INTERVAL_SECS:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def send_announcement_to_instance(
|
||||
provider: object, iface: object, instance_url: str, protocol: str
|
||||
) -> bool:
|
||||
"""Dogfeed one instance and, unless it is private, build and send the line.
|
||||
|
||||
The privacy gate is **fail-closed** (SPEC MA7 c / Invariant II): the
|
||||
announcement is sent only when ``/version`` explicitly reports
|
||||
``private_mode == False``; a ``True`` flag or any fetch/parse error skips it.
|
||||
|
||||
Parameters:
|
||||
provider: Active mesh provider exposing ``send_channel_announcement``.
|
||||
iface: Live mesh interface to transmit through.
|
||||
instance_url: Base URL of the target PotatoMesh instance.
|
||||
protocol: The ingestor's protocol (``"meshtastic"`` / ``"meshcore"``).
|
||||
|
||||
Returns:
|
||||
``True`` when an announcement was transmitted, ``False`` otherwise.
|
||||
"""
|
||||
|
||||
if fetch_private_mode(instance_url) is not False:
|
||||
config._debug_log(
|
||||
"Activity announcement skipped: instance private or unreachable",
|
||||
context="announce.tx",
|
||||
url=instance_url,
|
||||
)
|
||||
return False
|
||||
numbers = fetch_activity(instance_url, protocol)
|
||||
if numbers is None:
|
||||
config._debug_log(
|
||||
"Activity announcement skipped: no activity numbers from instance",
|
||||
context="announce.tx",
|
||||
url=instance_url,
|
||||
)
|
||||
return False
|
||||
active_nodes, packets_per_hour = numbers
|
||||
send = getattr(provider, "send_channel_announcement", None)
|
||||
if not callable(send):
|
||||
config._debug_log(
|
||||
"Activity announcement skipped: provider cannot transmit",
|
||||
context="announce.tx",
|
||||
url=instance_url,
|
||||
protocol=protocol,
|
||||
)
|
||||
return False
|
||||
text = build_announcement(protocol, active_nodes, packets_per_hour, instance_url)
|
||||
# Emitted for every outbound announcement so each TX is traceable end-to-end
|
||||
# (the provider then logs the mesh-layer send).
|
||||
config._debug_log(
|
||||
"Activity announcement transmitting",
|
||||
context="announce.tx",
|
||||
url=instance_url,
|
||||
protocol=protocol,
|
||||
active_nodes=active_nodes,
|
||||
packets_per_hour=packets_per_hour,
|
||||
text=text,
|
||||
)
|
||||
send(iface, text)
|
||||
return True
|
||||
|
||||
|
||||
def run_announcement_cycle(
|
||||
provider: object, iface: object, *, protocol: str | None = None
|
||||
) -> bool:
|
||||
"""Announce to **every** configured instance domain (SPEC MA6/MA8 per-domain).
|
||||
|
||||
Each instance is dogfed and sent its own numbers/link independently; a
|
||||
failure against one instance never aborts the others.
|
||||
|
||||
Parameters:
|
||||
provider: Active mesh provider.
|
||||
iface: Live mesh interface.
|
||||
protocol: Override protocol; defaults to
|
||||
:data:`~data.mesh_ingestor.config.PROTOCOL`.
|
||||
|
||||
Returns:
|
||||
``True`` when at least one announcement was transmitted.
|
||||
"""
|
||||
|
||||
protocol = protocol or getattr(config, "PROTOCOL", "meshtastic")
|
||||
sent_any = False
|
||||
for instance_url, _api_token in getattr(config, "INSTANCES", ()):
|
||||
try:
|
||||
if send_announcement_to_instance(provider, iface, instance_url, protocol):
|
||||
sent_any = True
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
"activity announcement failed",
|
||||
context="announce.cycle",
|
||||
severity="warn",
|
||||
url=instance_url,
|
||||
error_class=exc.__class__.__name__,
|
||||
error_message=str(exc),
|
||||
)
|
||||
return sent_any
|
||||
|
||||
|
||||
def maybe_run_announcements(
|
||||
provider: object,
|
||||
iface: object,
|
||||
*,
|
||||
start_time: float,
|
||||
last_announce: float | None,
|
||||
now: float | None = None,
|
||||
) -> float | None:
|
||||
"""Daemon entry point: run an announcement cycle when scheduled (SPEC MA7/MA8).
|
||||
|
||||
Applies the enable / RX-only gate and the 24-hour schedule, then dogfeeds and
|
||||
announces to each configured instance. The cycle timestamp advances whenever
|
||||
the schedule fires (whether or not any instance was actually announced to),
|
||||
so a private or unreachable instance is retried on the next 24-hour tick
|
||||
rather than on every loop iteration.
|
||||
|
||||
Parameters:
|
||||
provider: Active mesh provider.
|
||||
iface: Live mesh interface.
|
||||
start_time: Unix time the ingestor started.
|
||||
last_announce: Unix time of the previous cycle, or ``None``.
|
||||
now: Current unix time; defaults to :func:`time.time`.
|
||||
|
||||
Returns:
|
||||
The updated ``last_announce`` timestamp — *now* when a cycle ran, else
|
||||
the unchanged *last_announce*.
|
||||
"""
|
||||
|
||||
if now is None:
|
||||
now = time.time()
|
||||
if not announcements_enabled():
|
||||
return last_announce
|
||||
if not announce_due(start_time=start_time, last_announce=last_announce, now=now):
|
||||
return last_announce
|
||||
run_announcement_cycle(provider, iface)
|
||||
return now
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ANNOUNCE_CHAR_LIMITS",
|
||||
"ANNOUNCE_INITIAL_DELAY_SECS",
|
||||
"ANNOUNCE_INTERVAL_SECS",
|
||||
"announce_due",
|
||||
"announcements_enabled",
|
||||
"build_announcement",
|
||||
"fetch_activity",
|
||||
"fetch_private_mode",
|
||||
"maybe_run_announcements",
|
||||
"protocol_display_name",
|
||||
"run_announcement_cycle",
|
||||
"send_announcement_to_instance",
|
||||
]
|
||||
@@ -144,9 +144,9 @@ RX_ONLY = os.environ.get("RX_ONLY") == "1"
|
||||
"""Receive-only mode: forbid every ingestor-initiated mesh transmission.
|
||||
|
||||
Some operators run listening posts where any TX is undesired. When set, the
|
||||
ingestor never transmits on the mesh: currently this disables the MeshCore
|
||||
contact telemetry/status polls (the only ingestor-initiated RF traffic), and
|
||||
any future TX feature must honour it too. Local companion-link reads (host
|
||||
ingestor never transmits on the mesh: this disables the MeshCore contact
|
||||
telemetry/status polls and the periodic activity announcement (SPEC MA7), the
|
||||
only ingestor-initiated RF traffic. Local companion-link reads (host
|
||||
self-telemetry, contact roster, channel queries) are not transmissions and
|
||||
continue to work."""
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ import time
|
||||
|
||||
from pubsub import pub
|
||||
|
||||
from . import config, handlers, ingestors, interfaces, queue
|
||||
from . import announce, config, handlers, ingestors, interfaces, queue
|
||||
from .mesh_protocol import MeshProtocol
|
||||
from .utils import _retry_dict_snapshot
|
||||
|
||||
@@ -265,6 +265,7 @@ class _DaemonState:
|
||||
ingestor_announcement_sent: bool = False
|
||||
announced_target: bool = False
|
||||
last_self_node_report: float | None = None
|
||||
last_announce: float | None = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -573,6 +574,39 @@ def _try_send_self_node(state: _DaemonState) -> None:
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Periodic activity-announcement helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _process_announcements(state: _DaemonState) -> float | None:
|
||||
"""Run the periodic activity-announcement cycle when scheduled (SPEC MA6-MA8).
|
||||
|
||||
Delegates the transmit gate (``RX_ONLY``), the per-instance fail-closed
|
||||
privacy check, the >=24h post-start delay, and the 24h cadence to
|
||||
:func:`~data.mesh_ingestor.announce.maybe_run_announcements`, which dogfeeds
|
||||
each configured instance's own API for the numbers it announces. A no-op
|
||||
until the interface is connected (the announcement needs a live radio to
|
||||
transmit).
|
||||
|
||||
Parameters:
|
||||
state: Current daemon loop state (provides the provider, interface, and
|
||||
the last-announce timestamp).
|
||||
|
||||
Returns:
|
||||
The updated ``last_announce`` timestamp for :class:`_DaemonState`.
|
||||
"""
|
||||
|
||||
if state.iface is None:
|
||||
return state.last_announce
|
||||
return announce.maybe_run_announcements(
|
||||
state.provider,
|
||||
state.iface,
|
||||
start_time=ingestors.ingestor_start_time(),
|
||||
last_announce=state.last_announce,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Loop iteration helper
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -610,6 +644,7 @@ def _loop_iteration(state: _DaemonState) -> bool:
|
||||
or _now - state.last_self_node_report >= config._SELF_NODE_REPORT_INTERVAL_SECS
|
||||
):
|
||||
_try_send_self_node(state)
|
||||
state.last_announce = _process_announcements(state)
|
||||
state.retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS)
|
||||
return False
|
||||
|
||||
@@ -727,6 +762,7 @@ __all__ = [
|
||||
"_event_wait_allows_default_timeout",
|
||||
"_is_ble_interface",
|
||||
"_node_items_snapshot",
|
||||
"_process_announcements",
|
||||
"_process_ingestor_heartbeat",
|
||||
"_subscribe_receive_topics",
|
||||
"_try_connect",
|
||||
|
||||
@@ -24,7 +24,7 @@ from __future__ import annotations
|
||||
import math
|
||||
import time
|
||||
|
||||
from .. import config
|
||||
from .. import activity, config
|
||||
from ..serialization import _canonical_node_id
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -182,10 +182,17 @@ def last_packet_monotonic() -> float | None:
|
||||
|
||||
|
||||
def _mark_packet_seen() -> None:
|
||||
"""Record that a packet has been processed by updating the monotonic clock."""
|
||||
"""Record that a packet has been processed.
|
||||
|
||||
Updates the monotonic activity clock (used for inactivity-reconnect) and
|
||||
increments the merged mesh-activity counter (SPEC MA1) so that *every*
|
||||
received frame — including one a downstream handler later ignores or fails
|
||||
to store — is counted here, before any dispatch or drop decision.
|
||||
"""
|
||||
|
||||
global _last_packet_monotonic
|
||||
_last_packet_monotonic = time.monotonic()
|
||||
activity.record_packet()
|
||||
|
||||
|
||||
__all__ = [
|
||||
|
||||
@@ -21,7 +21,7 @@ from dataclasses import dataclass, field
|
||||
from typing import Callable
|
||||
|
||||
from .. import VERSION as INGESTOR_VERSION
|
||||
from . import config, queue
|
||||
from . import activity, config, queue
|
||||
from .serialization import _canonical_node_id
|
||||
|
||||
HEARTBEAT_INTERVAL_SECS = 60 * 60
|
||||
@@ -108,12 +108,17 @@ def queue_ingestor_heartbeat(
|
||||
if not force and last is not None and now - last < interval:
|
||||
return False
|
||||
|
||||
# Drain the merged activity counter only once we are committed to sending
|
||||
# (past every early-return guard above), so a suppressed heartbeat never
|
||||
# discards an interval's packet count. The value is the per-interval delta
|
||||
# since the previous heartbeat (SPEC MA2); ``take_packet_count`` resets it.
|
||||
payload = {
|
||||
"node_id": canonical,
|
||||
"start_time": STATE.start_time,
|
||||
"last_seen_time": now,
|
||||
"version": INGESTOR_VERSION,
|
||||
"protocol": getattr(config, "PROTOCOL", "meshtastic") or "meshtastic",
|
||||
"packets": activity.take_packet_count(),
|
||||
}
|
||||
if getattr(config, "LORA_FREQ", None) is not None:
|
||||
payload["lora_freq"] = config.LORA_FREQ
|
||||
|
||||
@@ -35,6 +35,15 @@ deferred until contacts have been fetched so that the daemon's first
|
||||
(issue #788).
|
||||
"""
|
||||
|
||||
_ANNOUNCE_SEND_TIMEOUT_SECS: float = 15.0
|
||||
"""Seconds to wait for an activity announcement (``send_chan_msg``) to complete.
|
||||
|
||||
The announcement is scheduled from the daemon's synchronous thread onto the
|
||||
MeshCore asyncio loop via ``run_coroutine_threadsafe``; this bounds the blocking
|
||||
wait on that cross-thread future so a wedged radio cannot stall the daemon loop
|
||||
(SPEC MA6/MA9).
|
||||
"""
|
||||
|
||||
_DEFAULT_BAUDRATE: int = 115200
|
||||
"""Default baud rate for MeshCore serial connections."""
|
||||
|
||||
|
||||
@@ -20,8 +20,8 @@ import asyncio
|
||||
import sys
|
||||
import threading
|
||||
|
||||
from ... import config
|
||||
from ._constants import _CONNECT_TIMEOUT_SECS
|
||||
from ... import activity, config
|
||||
from ._constants import _ANNOUNCE_SEND_TIMEOUT_SECS, _CONNECT_TIMEOUT_SECS
|
||||
from .decode import _self_info_to_node_dict
|
||||
from .identity import _meshcore_node_id
|
||||
from .interface import _MeshcoreInterface
|
||||
@@ -200,3 +200,40 @@ class MeshcoreProvider:
|
||||
if self_item is not None:
|
||||
items.append(self_item)
|
||||
return items
|
||||
|
||||
def send_channel_announcement(self, iface: object, text: str) -> None:
|
||||
"""Broadcast an activity announcement on the default channel (SPEC MA6/MA9).
|
||||
|
||||
Schedules ``send_chan_msg`` on channel
|
||||
:data:`~data.mesh_ingestor.config.CHANNEL_INDEX` onto the MeshCore
|
||||
asyncio loop (which the provider runs in a background thread) via
|
||||
:func:`asyncio.run_coroutine_threadsafe`, blocking up to
|
||||
:data:`_ANNOUNCE_SEND_TIMEOUT_SECS` for completion, and counts the
|
||||
transmission toward the merged activity total (MA1). This is an
|
||||
**optional**, duck-typed provider method (not a formal
|
||||
:class:`~data.mesh_ingestor.mesh_protocol.MeshProtocol` member); the
|
||||
daemon resolves it via ``getattr`` and skips it when absent. It is a
|
||||
no-op when the interface has no live event loop / handle.
|
||||
|
||||
Parameters:
|
||||
iface: Active :class:`_MeshcoreInterface`.
|
||||
text: Announcement string to transmit.
|
||||
"""
|
||||
|
||||
if not isinstance(iface, _MeshcoreInterface):
|
||||
return
|
||||
mc = getattr(iface, "_mc", None)
|
||||
loop = getattr(iface, "_loop", None)
|
||||
if mc is None or loop is None or loop.is_closed():
|
||||
return
|
||||
activity.record_tx()
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
mc.commands.send_chan_msg(config.CHANNEL_INDEX, text), loop
|
||||
)
|
||||
future.result(timeout=_ANNOUNCE_SEND_TIMEOUT_SECS)
|
||||
config._debug_log(
|
||||
"MeshCore activity announcement transmitted",
|
||||
context="meshcore.tx",
|
||||
channel=config.CHANNEL_INDEX,
|
||||
chars=len(text),
|
||||
)
|
||||
|
||||
@@ -35,7 +35,7 @@ import asyncio
|
||||
import time
|
||||
from collections.abc import Mapping
|
||||
|
||||
from ... import config
|
||||
from ... import activity, config
|
||||
from .interface import _MeshcoreInterface
|
||||
from .messages import _derive_message_id
|
||||
|
||||
@@ -401,7 +401,12 @@ async def _poll_contact_telemetry(
|
||||
Falls back to a status request when the telemetry pull yields nothing, so
|
||||
sensor-less nodes still report battery/uptime. One contact per call keeps
|
||||
airtime bounded to a single request per poll interval regardless of roster
|
||||
size; the meshcore library serialises mesh requests internally.
|
||||
size; the meshcore library serialises mesh requests internally. Both ends
|
||||
of the attempt emit a ``meshcore.telemetry.poll`` debug line — one when the
|
||||
request is initiated, one when neither the telemetry nor the status request
|
||||
returned usable data — because ``req_*_sync`` return ``None`` on timeout
|
||||
without raising, which would otherwise leave an unanswered poll
|
||||
indistinguishable from a disabled poll loop.
|
||||
|
||||
Parameters:
|
||||
mc: Connected MeshCore instance.
|
||||
@@ -415,7 +420,16 @@ async def _poll_contact_telemetry(
|
||||
node_id = iface.lookup_node_id((contact.get("public_key") or "")[:12])
|
||||
if node_id is None:
|
||||
return
|
||||
# Log before the request goes out: req_*_sync return None on timeout without
|
||||
# raising, so an unanswered poll is otherwise silent.
|
||||
config._debug_log(
|
||||
"MeshCore contact telemetry poll initiated",
|
||||
context="meshcore.telemetry.poll",
|
||||
node_id=node_id,
|
||||
)
|
||||
try:
|
||||
# An on-air pull is an ingestor transmission — count it (SPEC MA1).
|
||||
activity.record_tx()
|
||||
lpp = await mc.commands.req_telemetry_sync(contact)
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
@@ -430,6 +444,8 @@ async def _poll_contact_telemetry(
|
||||
):
|
||||
return
|
||||
try:
|
||||
# The status fallback is a second on-air pull — count it too (MA1).
|
||||
activity.record_tx()
|
||||
status = await mc.commands.req_status_sync(contact)
|
||||
except Exception as exc:
|
||||
config._debug_log(
|
||||
@@ -439,8 +455,14 @@ async def _poll_contact_telemetry(
|
||||
error=str(exc),
|
||||
)
|
||||
return
|
||||
_queue_meshcore_telemetry(
|
||||
if _queue_meshcore_telemetry(
|
||||
handlers, node_id, _status_to_telemetry_section(status), "status"
|
||||
):
|
||||
return
|
||||
config._debug_log(
|
||||
"MeshCore contact telemetry poll returned no data",
|
||||
context="meshcore.telemetry.poll",
|
||||
node_id=node_id,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ from __future__ import annotations
|
||||
|
||||
from pubsub import pub
|
||||
|
||||
from .. import config, daemon as _daemon, handlers, interfaces
|
||||
from .. import activity, config, daemon as _daemon, handlers, interfaces
|
||||
from ..utils import _retry_dict_snapshot
|
||||
|
||||
|
||||
@@ -96,5 +96,31 @@ class MeshtasticProvider:
|
||||
return []
|
||||
return result
|
||||
|
||||
def send_channel_announcement(self, iface: object, text: str) -> None:
|
||||
"""Broadcast an activity announcement on the default channel (SPEC MA6/MA9).
|
||||
|
||||
Sends *text* on channel :data:`~data.mesh_ingestor.config.CHANNEL_INDEX`
|
||||
via the Meshtastic interface and counts the transmission toward the
|
||||
merged activity total (MA1). This is an **optional**, duck-typed provider
|
||||
method (not a formal :class:`MeshProtocol` member); the daemon resolves
|
||||
it via ``getattr`` and skips it when absent.
|
||||
|
||||
Parameters:
|
||||
iface: Active Meshtastic interface exposing ``sendText``.
|
||||
text: Announcement string to transmit.
|
||||
"""
|
||||
|
||||
send_text = getattr(iface, "sendText", None)
|
||||
if not callable(send_text):
|
||||
return
|
||||
activity.record_tx()
|
||||
send_text(text, channelIndex=config.CHANNEL_INDEX)
|
||||
config._debug_log(
|
||||
"Meshtastic activity announcement transmitted",
|
||||
context="meshtastic.tx",
|
||||
channel=config.CHANNEL_INDEX,
|
||||
chars=len(text),
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["MeshtasticProvider"]
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
"""Unit tests for the merged mesh-activity counter (SPEC MA1/MA2).
|
||||
|
||||
Covers :mod:`data.mesh_ingestor.activity` (the counter API), the receive-seam
|
||||
wiring in :func:`data.mesh_ingestor.handlers._state._mark_packet_seen` (so every
|
||||
received frame — stored, ignored, or errored — is counted), and the
|
||||
per-interval ``packets`` delta carried by
|
||||
:func:`data.mesh_ingestor.ingestors.queue_ingestor_heartbeat`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
import data.mesh_ingestor.activity as activity
|
||||
import data.mesh_ingestor.ingestors as ingestors_mod
|
||||
from data.mesh_ingestor.handlers import _state
|
||||
from data.mesh_ingestor.ingestors import (
|
||||
_IngestorState,
|
||||
queue_ingestor_heartbeat,
|
||||
set_ingestor_node_id,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_activity_counter():
|
||||
"""Zero the shared activity counter around every test for isolation."""
|
||||
activity.reset()
|
||||
yield
|
||||
activity.reset()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_ingestor_state():
|
||||
"""Swap in fresh ingestor identity state around every test."""
|
||||
original = ingestors_mod.STATE
|
||||
ingestors_mod.STATE = _IngestorState()
|
||||
yield
|
||||
ingestors_mod.STATE = original
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Counter API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCounterApi:
|
||||
"""Tests for the :mod:`data.mesh_ingestor.activity` counter primitives."""
|
||||
|
||||
def test_record_packet_counts_one_by_default(self):
|
||||
"""A bare ``record_packet()`` adds a single frame."""
|
||||
activity.record_packet()
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
def test_record_packet_counts_a_custom_amount(self):
|
||||
"""``record_packet(n)`` adds ``n`` frames at once."""
|
||||
activity.record_packet(4)
|
||||
assert activity.take_packet_count() == 4
|
||||
|
||||
def test_record_tx_counts_transmissions(self):
|
||||
"""``record_tx()`` counts an ingestor transmission (MA1)."""
|
||||
activity.record_tx()
|
||||
activity.record_tx(2)
|
||||
assert activity.take_packet_count() == 3
|
||||
|
||||
def test_rx_and_tx_share_one_merged_count(self):
|
||||
"""RX and TX accumulate into a single merged figure (MA1)."""
|
||||
activity.record_packet(5)
|
||||
activity.record_tx(2)
|
||||
assert activity.take_packet_count() == 7
|
||||
|
||||
def test_take_packet_count_resets_to_zero(self):
|
||||
"""Reading the count via ``take_packet_count`` clears it (MA2)."""
|
||||
activity.record_packet(3)
|
||||
assert activity.take_packet_count() == 3
|
||||
assert activity.take_packet_count() == 0
|
||||
|
||||
def test_reset_discards_the_running_total(self):
|
||||
"""``reset()`` zeroes the counter without returning it."""
|
||||
activity.record_packet(9)
|
||||
activity.reset()
|
||||
assert activity.take_packet_count() == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Receive seam — every frame counted, before any dispatch/drop
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestReceiveSeam:
|
||||
"""The RX seam counts every received frame regardless of its fate."""
|
||||
|
||||
def test_mark_packet_seen_counts_the_frame(self):
|
||||
"""``_mark_packet_seen`` increments the merged counter (MA1)."""
|
||||
_state._mark_packet_seen()
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
def test_on_receive_counts_stored_ignored_and_errored(self, monkeypatch):
|
||||
"""Every ``on_receive`` outcome is counted at the seam (MA1).
|
||||
|
||||
The count is taken before dispatch, so a packet the downstream handler
|
||||
stores, ignores (returns without queuing), or errors on (raises) is
|
||||
counted identically.
|
||||
"""
|
||||
from data.mesh_ingestor.handlers import generic
|
||||
|
||||
monkeypatch.setattr(generic, "_pkt_to_dict", lambda p: dict(p))
|
||||
|
||||
# Stored / ignored: store returns without raising.
|
||||
monkeypatch.setattr(generic, "store_packet_dict", lambda pkt: None)
|
||||
generic.on_receive({"id": 1}, object())
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
# Errored: store raises; on_receive swallows it, but the frame was
|
||||
# already counted at the seam.
|
||||
def _boom(pkt):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
monkeypatch.setattr(generic, "store_packet_dict", _boom)
|
||||
generic.on_receive({"id": 2}, object())
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
def test_on_receive_dedup_counts_once(self, monkeypatch):
|
||||
"""A packet already flagged ``_potatomesh_seen`` is not re-counted."""
|
||||
from data.mesh_ingestor.handlers import generic
|
||||
|
||||
monkeypatch.setattr(generic, "_pkt_to_dict", lambda p: dict(p))
|
||||
monkeypatch.setattr(generic, "store_packet_dict", lambda pkt: None)
|
||||
|
||||
packet = {"id": 1}
|
||||
generic.on_receive(packet, object())
|
||||
generic.on_receive(packet, object()) # same dict → deduped
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Heartbeat per-interval delta (MA2)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestHeartbeatDelta:
|
||||
"""The heartbeat carries — and resets — the per-interval packet delta."""
|
||||
|
||||
def test_heartbeat_delta_included_in_payload(self):
|
||||
"""The heartbeat payload carries the accumulated ``packets`` count."""
|
||||
set_ingestor_node_id("!aabbccdd")
|
||||
activity.record_packet(7)
|
||||
sent = []
|
||||
queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
|
||||
assert sent[0]["packets"] == 7
|
||||
|
||||
def test_heartbeat_delta_resets_between_sends(self):
|
||||
"""Consecutive heartbeats report their own interval, not the cumulative."""
|
||||
set_ingestor_node_id("!aabbccdd")
|
||||
sent = []
|
||||
activity.record_packet(3)
|
||||
queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
|
||||
activity.record_packet(2)
|
||||
queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
|
||||
assert [pl["packets"] for pl in sent] == [3, 2]
|
||||
|
||||
def test_heartbeat_delta_zero_when_no_traffic(self):
|
||||
"""A heartbeat with no observed frames reports ``0``."""
|
||||
set_ingestor_node_id("!aabbccdd")
|
||||
sent = []
|
||||
queue_ingestor_heartbeat(force=True, send=lambda p, pl: sent.append(pl))
|
||||
assert sent[0]["packets"] == 0
|
||||
|
||||
def test_heartbeat_delta_not_drained_when_suppressed(self):
|
||||
"""A suppressed (interval-guarded) heartbeat must not discard the count."""
|
||||
set_ingestor_node_id("!aabbccdd")
|
||||
ingestors_mod.STATE.last_heartbeat = int(time.time())
|
||||
activity.record_packet(5)
|
||||
sent = []
|
||||
result = queue_ingestor_heartbeat(send=lambda p, pl: sent.append(pl))
|
||||
assert result is False
|
||||
assert sent == []
|
||||
# The interval-guarded heartbeat returned before building the payload,
|
||||
# so the 5 frames survive for the next real heartbeat.
|
||||
assert activity.take_packet_count() == 5
|
||||
@@ -0,0 +1,506 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
"""Unit tests for :mod:`data.mesh_ingestor.announce` (SPEC MA6).
|
||||
|
||||
Covers the character-limited announcement message builder and the read-only
|
||||
*dogfeed* HTTP client (fetch the instance's own ``/version`` and ``/api/stats``),
|
||||
including the fail-soft behaviour on network / shape errors.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import urllib.error
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(REPO_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
import data.mesh_ingestor.announce as announce
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
"""Minimal context-manager stand-in for an ``http.client.HTTPResponse``."""
|
||||
|
||||
def __init__(self, body: str):
|
||||
self._body = body.encode("utf-8")
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_exc):
|
||||
return False
|
||||
|
||||
def read(self):
|
||||
return self._body
|
||||
|
||||
|
||||
def _install_fake_http(monkeypatch, url_to_payload):
|
||||
"""Patch ``announce``'s ``urlopen`` to serve *url_to_payload* by URL."""
|
||||
|
||||
def _fake_urlopen(req, timeout=None):
|
||||
url = getattr(req, "full_url", req)
|
||||
if url not in url_to_payload:
|
||||
raise urllib.error.URLError(f"no fake response for {url}")
|
||||
return _FakeResponse(json.dumps(url_to_payload[url]))
|
||||
|
||||
monkeypatch.setattr(announce.urllib.request, "urlopen", _fake_urlopen)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Message builder
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildAnnouncementMessage:
|
||||
"""Tests for :func:`announce.build_announcement` (MA6)."""
|
||||
|
||||
def test_message_format_matches_spec(self):
|
||||
"""The announcement line matches the SPEC MA6 template exactly."""
|
||||
text = announce.build_announcement("meshcore", 12, 50, "https://mesh.example")
|
||||
assert text == (
|
||||
"MeshCore activity in the last 24h: 12 active nodes, "
|
||||
"50 packets/hour. https://mesh.example"
|
||||
)
|
||||
|
||||
def test_message_uses_protocol_display_name(self):
|
||||
"""Meshtastic renders its capitalised display name."""
|
||||
text = announce.build_announcement("meshtastic", 3, 7, "https://m")
|
||||
assert text.startswith("Meshtastic activity in the last 24h:")
|
||||
|
||||
def test_message_truncates_to_protocol_char_limit(self):
|
||||
"""An over-long line is clipped to the protocol's character limit."""
|
||||
long_url = "https://" + ("a" * 300) + ".example"
|
||||
text = announce.build_announcement("meshtastic", 5, 10, long_url)
|
||||
assert len(text) == announce.ANNOUNCE_CHAR_LIMITS["meshtastic"]
|
||||
|
||||
def test_message_honours_explicit_char_limit(self):
|
||||
"""An explicit ``char_limit`` overrides the per-protocol default."""
|
||||
text = announce.build_announcement(
|
||||
"meshcore", 5, 10, "https://x", char_limit=10
|
||||
)
|
||||
assert len(text) == 10
|
||||
|
||||
def test_message_unknown_protocol_uses_fallback_limit(self):
|
||||
"""An unrecognised protocol falls back to the default char limit."""
|
||||
long_url = "https://" + ("z" * 300)
|
||||
text = announce.build_announcement("reticulum", 1, 2, long_url)
|
||||
assert text.startswith("Reticulum activity")
|
||||
assert len(text) == announce._DEFAULT_CHAR_LIMIT
|
||||
|
||||
|
||||
class TestProtocolDisplayName:
|
||||
"""Tests for :func:`announce.protocol_display_name`."""
|
||||
|
||||
def test_known_protocols(self):
|
||||
"""Known keys map to their branded labels."""
|
||||
assert announce.protocol_display_name("meshcore") == "MeshCore"
|
||||
assert announce.protocol_display_name("meshtastic") == "Meshtastic"
|
||||
|
||||
def test_unknown_protocol_is_capitalised(self):
|
||||
"""An unknown key is capitalised as a best effort."""
|
||||
assert announce.protocol_display_name("reticulum") == "Reticulum"
|
||||
|
||||
def test_empty_protocol_is_blank(self):
|
||||
"""An empty/``None`` protocol yields an empty label."""
|
||||
assert announce.protocol_display_name("") == ""
|
||||
assert announce.protocol_display_name(None) == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dogfeed HTTP client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDogfeedFetchActivity:
|
||||
"""Tests for :func:`announce.fetch_activity` (MA6)."""
|
||||
|
||||
def test_dogfeed_reads_nodes_and_packets_per_hour(self, monkeypatch):
|
||||
"""Returns ``(<protocol>.nodes.day, packets_per_hour.<protocol>)``."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{
|
||||
"https://mesh.example/api/stats": {
|
||||
"meshcore": {"nodes": {"day": 12, "hour": 3}},
|
||||
"meshtastic": {"nodes": {"day": 99}},
|
||||
"packets_per_hour": {"meshcore": 50, "meshtastic": 30},
|
||||
},
|
||||
},
|
||||
)
|
||||
assert announce.fetch_activity("https://mesh.example", "meshcore") == (12, 50)
|
||||
|
||||
def test_dogfeed_returns_none_on_malformed_shape(self, monkeypatch):
|
||||
"""A response missing the packets_per_hour map yields ``None``."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{"https://mesh.example/api/stats": {"meshcore": {"nodes": {"day": 1}}}},
|
||||
)
|
||||
assert announce.fetch_activity("https://mesh.example", "meshcore") is None
|
||||
|
||||
def test_dogfeed_returns_none_when_nodes_object_missing(self, monkeypatch):
|
||||
"""A scope present but lacking a ``nodes`` sub-object yields ``None``."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{
|
||||
"https://mesh.example/api/stats": {
|
||||
"meshcore": {},
|
||||
"packets_per_hour": {"meshcore": 50},
|
||||
},
|
||||
},
|
||||
)
|
||||
assert announce.fetch_activity("https://mesh.example", "meshcore") is None
|
||||
|
||||
def test_dogfeed_returns_none_on_non_integer_counts(self, monkeypatch):
|
||||
"""Non-numeric counts are rejected as malformed."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{
|
||||
"https://mesh.example/api/stats": {
|
||||
"meshcore": {"nodes": {"day": "lots"}},
|
||||
"packets_per_hour": {"meshcore": 50},
|
||||
},
|
||||
},
|
||||
)
|
||||
assert announce.fetch_activity("https://mesh.example", "meshcore") is None
|
||||
|
||||
def test_dogfeed_returns_none_on_network_error(self, monkeypatch):
|
||||
"""A transport error fails soft to ``None``."""
|
||||
|
||||
def _boom(req, timeout=None):
|
||||
raise urllib.error.URLError("boom")
|
||||
|
||||
monkeypatch.setattr(announce.urllib.request, "urlopen", _boom)
|
||||
assert announce.fetch_activity("https://mesh.example", "meshtastic") is None
|
||||
|
||||
|
||||
class TestDogfeedFetchPrivateMode:
|
||||
"""Tests for :func:`announce.fetch_private_mode` (MA7 privacy gate)."""
|
||||
|
||||
@pytest.mark.parametrize("flag", [True, False])
|
||||
def test_dogfeed_reads_private_mode_flag(self, monkeypatch, flag):
|
||||
"""Reads ``config.private_mode`` from ``/version``."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{"https://mesh.example/version": {"config": {"private_mode": flag}}},
|
||||
)
|
||||
assert announce.fetch_private_mode("https://mesh.example") is flag
|
||||
|
||||
def test_dogfeed_missing_flag_returns_none(self, monkeypatch):
|
||||
"""A response without the flag yields ``None`` (caller fails closed)."""
|
||||
_install_fake_http(
|
||||
monkeypatch,
|
||||
{"https://mesh.example/version": {"config": {"site_name": "x"}}},
|
||||
)
|
||||
assert announce.fetch_private_mode("https://mesh.example") is None
|
||||
|
||||
def test_dogfeed_missing_config_block_returns_none(self, monkeypatch):
|
||||
"""A response without a ``config`` block yields ``None``."""
|
||||
_install_fake_http(
|
||||
monkeypatch, {"https://mesh.example/version": {"version": "0.7.3"}}
|
||||
)
|
||||
assert announce.fetch_private_mode("https://mesh.example") is None
|
||||
|
||||
def test_dogfeed_network_error_returns_none(self, monkeypatch):
|
||||
"""A transport error fails soft to ``None`` (fail-closed input)."""
|
||||
|
||||
def _boom(req, timeout=None):
|
||||
raise urllib.error.URLError("boom")
|
||||
|
||||
monkeypatch.setattr(announce.urllib.request, "urlopen", _boom)
|
||||
assert announce.fetch_private_mode("https://mesh.example") is None
|
||||
|
||||
|
||||
class TestDogfeedGetJson:
|
||||
"""Tests for the low-level :func:`announce._get_json` helper."""
|
||||
|
||||
def test_dogfeed_non_object_json_returns_none(self, monkeypatch):
|
||||
"""A JSON array (not an object) is rejected."""
|
||||
_install_fake_http(monkeypatch, {"https://mesh.example/x": [1, 2, 3]})
|
||||
assert announce._get_json("https://mesh.example/x") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scheduling & gating (SPEC MA7/MA8)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _RecordingProvider:
|
||||
"""Fake provider that records ``send_channel_announcement`` calls."""
|
||||
|
||||
def __init__(self):
|
||||
self.sent = []
|
||||
|
||||
def send_channel_announcement(self, iface, text):
|
||||
self.sent.append((iface, text))
|
||||
|
||||
|
||||
def _stub_dogfeed(monkeypatch, *, private=False, numbers=(12, 50)):
|
||||
"""Patch the dogfeed fetchers to controlled return values (no HTTP)."""
|
||||
monkeypatch.setattr(announce, "fetch_private_mode", lambda url, **_kw: private)
|
||||
monkeypatch.setattr(announce, "fetch_activity", lambda url, proto, **_kw: numbers)
|
||||
|
||||
|
||||
class TestAnnouncementGates:
|
||||
"""Tests for :func:`announce.announcements_enabled` (SPEC MA7 a)."""
|
||||
|
||||
def test_gate_enabled_by_default(self, monkeypatch):
|
||||
"""Enabled when RX_ONLY is off (the default)."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", False)
|
||||
assert announce.announcements_enabled() is True
|
||||
|
||||
def test_gate_disabled_when_rx_only(self, monkeypatch):
|
||||
"""RX_ONLY (reused as the single transmit gate) forbids the announcement."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", True)
|
||||
assert announce.announcements_enabled() is False
|
||||
|
||||
|
||||
class TestAnnounceDue:
|
||||
"""Tests for :func:`announce.announce_due` (SPEC MA7 d / MA8)."""
|
||||
|
||||
def test_gate_not_due_before_24h_elapsed(self):
|
||||
"""Withheld until 24 h after start."""
|
||||
now = 1_000_000
|
||||
assert (
|
||||
announce.announce_due(start_time=now - 1000, last_announce=None, now=now)
|
||||
is False
|
||||
)
|
||||
|
||||
def test_due_after_24h_elapsed(self):
|
||||
"""Eligible once 24 h have elapsed since start."""
|
||||
now = 1_000_000
|
||||
assert (
|
||||
announce.announce_due(start_time=now - 90_000, last_announce=None, now=now)
|
||||
is True
|
||||
)
|
||||
|
||||
def test_cadence_interval_blocks_second_within_24h(self):
|
||||
"""A recent cycle blocks the next until the 24 h interval passes."""
|
||||
now = 1_000_000
|
||||
assert (
|
||||
announce.announce_due(
|
||||
start_time=now - 200_000, last_announce=now - 3600, now=now
|
||||
)
|
||||
is False
|
||||
)
|
||||
|
||||
def test_cadence_due_again_after_24h_interval(self):
|
||||
"""Due again once 24 h have elapsed since the previous cycle."""
|
||||
now = 1_000_000
|
||||
assert (
|
||||
announce.announce_due(
|
||||
start_time=now - 200_000, last_announce=now - 90_000, now=now
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
|
||||
class TestSendAnnouncementToInstance:
|
||||
"""Tests for :func:`announce.send_announcement_to_instance` (MA6/MA7)."""
|
||||
|
||||
def test_sends_the_built_line_when_public(self, monkeypatch):
|
||||
"""A public instance receives the dogfed announcement line."""
|
||||
_stub_dogfeed(monkeypatch, private=False, numbers=(12, 50))
|
||||
provider = _RecordingProvider()
|
||||
sent = announce.send_announcement_to_instance(
|
||||
provider, "IFACE", "https://mesh.example", "meshcore"
|
||||
)
|
||||
assert sent is True
|
||||
assert provider.sent == [
|
||||
(
|
||||
"IFACE",
|
||||
"MeshCore activity in the last 24h: 12 active nodes, "
|
||||
"50 packets/hour. https://mesh.example",
|
||||
)
|
||||
]
|
||||
|
||||
def test_private_instance_is_skipped_fail_closed(self, monkeypatch):
|
||||
"""A private instance is skipped (nothing transmitted)."""
|
||||
_stub_dogfeed(monkeypatch, private=True)
|
||||
provider = _RecordingProvider()
|
||||
assert (
|
||||
announce.send_announcement_to_instance(
|
||||
provider, "I", "https://m", "meshcore"
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert provider.sent == []
|
||||
|
||||
def test_private_fetch_error_fails_closed(self, monkeypatch):
|
||||
"""An unknown privacy state (None) is treated as private (fail-closed)."""
|
||||
_stub_dogfeed(monkeypatch, private=None)
|
||||
provider = _RecordingProvider()
|
||||
assert (
|
||||
announce.send_announcement_to_instance(
|
||||
provider, "I", "https://m", "meshcore"
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert provider.sent == []
|
||||
|
||||
def test_skips_when_activity_numbers_unavailable(self, monkeypatch):
|
||||
"""No announcement when the stats dogfeed yields nothing."""
|
||||
monkeypatch.setattr(announce, "fetch_private_mode", lambda url, **_kw: False)
|
||||
monkeypatch.setattr(announce, "fetch_activity", lambda url, proto, **_kw: None)
|
||||
provider = _RecordingProvider()
|
||||
assert (
|
||||
announce.send_announcement_to_instance(
|
||||
provider, "I", "https://m", "meshcore"
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert provider.sent == []
|
||||
|
||||
def test_skips_when_provider_cannot_send(self, monkeypatch):
|
||||
"""A provider without send_channel_announcement is a no-op."""
|
||||
_stub_dogfeed(monkeypatch, private=False, numbers=(1, 2))
|
||||
assert (
|
||||
announce.send_announcement_to_instance(
|
||||
object(), "I", "https://m", "meshcore"
|
||||
)
|
||||
is False
|
||||
)
|
||||
|
||||
def test_logs_transmitting_with_full_context(self, monkeypatch):
|
||||
"""Every outbound announcement emits an ``announce.tx`` debug line so the
|
||||
TX is traceable end-to-end (target, protocol, numbers, text)."""
|
||||
_stub_dogfeed(monkeypatch, private=False, numbers=(12, 50))
|
||||
logs = []
|
||||
monkeypatch.setattr(
|
||||
announce.config,
|
||||
"_debug_log",
|
||||
lambda message, **meta: logs.append((message, meta)),
|
||||
)
|
||||
announce.send_announcement_to_instance(
|
||||
_RecordingProvider(), "IFACE", "https://mesh.example", "meshcore"
|
||||
)
|
||||
transmit = [
|
||||
meta for msg, meta in logs if msg == "Activity announcement transmitting"
|
||||
]
|
||||
assert len(transmit) == 1
|
||||
meta = transmit[0]
|
||||
assert meta["context"] == "announce.tx"
|
||||
assert meta["url"] == "https://mesh.example"
|
||||
assert meta["protocol"] == "meshcore"
|
||||
assert meta["active_nodes"] == 12
|
||||
assert meta["packets_per_hour"] == 50
|
||||
assert "packets/hour" in meta["text"]
|
||||
|
||||
def test_logs_skip_reason_when_private(self, monkeypatch):
|
||||
"""A skipped (private/unreachable) instance emits a skip debug line."""
|
||||
_stub_dogfeed(monkeypatch, private=True)
|
||||
logs = []
|
||||
monkeypatch.setattr(
|
||||
announce.config,
|
||||
"_debug_log",
|
||||
lambda message, **meta: logs.append((message, meta)),
|
||||
)
|
||||
announce.send_announcement_to_instance(
|
||||
_RecordingProvider(), "I", "https://m", "meshcore"
|
||||
)
|
||||
assert any("skipped: instance private" in msg for msg, _meta in logs)
|
||||
|
||||
|
||||
class TestRunAnnouncementCycle:
|
||||
"""Tests for :func:`announce.run_announcement_cycle` (SPEC MA8 per-domain)."""
|
||||
|
||||
def test_domains_announces_each_configured_instance(self, monkeypatch):
|
||||
"""Every configured instance is announced to, with its own link."""
|
||||
monkeypatch.setattr(
|
||||
announce.config, "INSTANCES", (("https://a", ""), ("https://b", ""))
|
||||
)
|
||||
monkeypatch.setattr(announce.config, "PROTOCOL", "meshtastic")
|
||||
_stub_dogfeed(monkeypatch, private=False, numbers=(3, 7))
|
||||
provider = _RecordingProvider()
|
||||
assert announce.run_announcement_cycle(provider, "IFACE") is True
|
||||
urls = [text.rsplit(" ", 1)[-1] for _iface, text in provider.sent]
|
||||
assert urls == ["https://a", "https://b"]
|
||||
|
||||
def test_cycle_survives_one_instance_error(self, monkeypatch):
|
||||
"""A failure against one instance never aborts the others."""
|
||||
monkeypatch.setattr(
|
||||
announce.config, "INSTANCES", (("https://bad", ""), ("https://good", ""))
|
||||
)
|
||||
monkeypatch.setattr(announce.config, "PROTOCOL", "meshcore")
|
||||
|
||||
def _priv(url, **_kw):
|
||||
if url == "https://bad":
|
||||
raise RuntimeError("boom")
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(announce, "fetch_private_mode", _priv)
|
||||
monkeypatch.setattr(
|
||||
announce, "fetch_activity", lambda url, proto, **_kw: (1, 1)
|
||||
)
|
||||
provider = _RecordingProvider()
|
||||
assert announce.run_announcement_cycle(provider, "I") is True
|
||||
assert [t.rsplit(" ", 1)[-1] for _i, t in provider.sent] == ["https://good"]
|
||||
|
||||
|
||||
class TestMaybeRunAnnouncements:
|
||||
"""Tests for the daemon entry point :func:`announce.maybe_run_announcements`."""
|
||||
|
||||
def test_gate_noop_when_not_enabled(self, monkeypatch):
|
||||
"""No cycle runs and last_announce is unchanged when RX_ONLY forbids TX."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", True)
|
||||
ran = []
|
||||
monkeypatch.setattr(
|
||||
announce, "run_announcement_cycle", lambda *a, **k: ran.append(True)
|
||||
)
|
||||
result = announce.maybe_run_announcements(
|
||||
"P", "I", start_time=0, last_announce=42.0, now=1_000_000
|
||||
)
|
||||
assert result == 42.0
|
||||
assert ran == []
|
||||
|
||||
def test_cadence_noop_when_not_due(self, monkeypatch):
|
||||
"""No cycle runs before the 24 h delay; last_announce is unchanged."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", False)
|
||||
ran = []
|
||||
monkeypatch.setattr(
|
||||
announce, "run_announcement_cycle", lambda *a, **k: ran.append(True)
|
||||
)
|
||||
now = 1_000_000
|
||||
result = announce.maybe_run_announcements(
|
||||
"P", "I", start_time=now - 1000, last_announce=None, now=now
|
||||
)
|
||||
assert result is None
|
||||
assert ran == []
|
||||
|
||||
def test_cadence_runs_cycle_and_advances_timestamp(self, monkeypatch):
|
||||
"""When due, the cycle runs and last_announce advances to now."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", False)
|
||||
ran = []
|
||||
monkeypatch.setattr(
|
||||
announce,
|
||||
"run_announcement_cycle",
|
||||
lambda provider, iface, **k: ran.append((provider, iface)) or True,
|
||||
)
|
||||
now = 1_000_000
|
||||
result = announce.maybe_run_announcements(
|
||||
"P", "IFACE", start_time=now - 90_000, last_announce=None, now=now
|
||||
)
|
||||
assert result == now
|
||||
assert ran == [("P", "IFACE")]
|
||||
|
||||
def test_cadence_defaults_now_to_wall_clock(self, monkeypatch):
|
||||
"""With ``now`` omitted the current wall clock is used (gate-off path)."""
|
||||
monkeypatch.setattr(announce.config, "RX_ONLY", True)
|
||||
result = announce.maybe_run_announcements(
|
||||
"P", "I", start_time=0, last_announce=7.0
|
||||
)
|
||||
assert result == 7.0
|
||||
@@ -829,6 +829,33 @@ def test_loop_iteration_full_pass_returns_false(monkeypatch):
|
||||
assert daemon._loop_iteration(state) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _process_announcements
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_process_announcements_noop_without_iface():
|
||||
"""Returns the unchanged last_announce when no interface is connected."""
|
||||
state = _make_state(iface=None, last_announce=123.0)
|
||||
assert daemon._process_announcements(state) == 123.0
|
||||
|
||||
|
||||
def test_process_announcements_delegates_when_connected(monkeypatch):
|
||||
"""Delegates to announce.maybe_run_announcements with the daemon's provider,
|
||||
interface, ingestor start time, and last_announce, and returns its result."""
|
||||
calls = []
|
||||
|
||||
def _fake(provider, iface, *, start_time, last_announce):
|
||||
calls.append((provider, iface, start_time, last_announce))
|
||||
return 999.0
|
||||
|
||||
monkeypatch.setattr(daemon.announce, "maybe_run_announcements", _fake)
|
||||
monkeypatch.setattr(daemon.ingestors, "ingestor_start_time", lambda: 555)
|
||||
state = _make_state(iface="IFACE", provider="PROV", last_announce=None)
|
||||
assert daemon._process_announcements(state) == 999.0
|
||||
assert calls == [("PROV", "IFACE", 555, None)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PROTOCOL env-var selection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1957,6 +1957,118 @@ def test_poll_contact_telemetry_paths(monkeypatch):
|
||||
assert captured == []
|
||||
|
||||
|
||||
def test_poll_contact_telemetry_counts_tx(monkeypatch):
|
||||
"""Each on-air poll request counts as a transmission (SPEC MA1).
|
||||
|
||||
The stubbed handlers' ``_mark_packet_seen`` is a no-op, so the merged
|
||||
activity counter moves only via the ``activity.record_tx()`` calls inside
|
||||
``_poll_contact_telemetry`` — one per on-air request. This pins the exact
|
||||
TX count for the telemetry pull, the status fallback, an error path, and
|
||||
the no-contact short-circuit.
|
||||
"""
|
||||
import types
|
||||
|
||||
import data.mesh_ingestor.activity as activity
|
||||
|
||||
mc_tel, iface, stub, _captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
|
||||
def _mc(
|
||||
telemetry_result=None,
|
||||
telemetry_error=None,
|
||||
status_result=None,
|
||||
status_error=None,
|
||||
):
|
||||
class _Commands:
|
||||
async def req_telemetry_sync(self, contact):
|
||||
if telemetry_error:
|
||||
raise telemetry_error
|
||||
return telemetry_result
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
if status_error:
|
||||
raise status_error
|
||||
return status_result
|
||||
|
||||
return types.SimpleNamespace(commands=_Commands())
|
||||
|
||||
# LPP success → exactly one on-air request (telemetry pull), no fallback.
|
||||
activity.take_packet_count() # drain any activity from earlier tests
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_result=[{"type": "temperature", "value": 21.5}]),
|
||||
iface,
|
||||
stub,
|
||||
{},
|
||||
)
|
||||
)
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
# Empty telemetry → status fallback: two on-air requests counted.
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_result=None, status_result={"bat": 4056}), iface, stub, {}
|
||||
)
|
||||
)
|
||||
assert activity.take_packet_count() == 2
|
||||
|
||||
# Telemetry request raises → the attempt is still counted (one TX), no fallback.
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
_mc(telemetry_error=RuntimeError("timeout")), iface, stub, {}
|
||||
)
|
||||
)
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
# No resolvable contact → returns before any request → nothing counted.
|
||||
empty = _MeshcoreInterface(target=None)
|
||||
asyncio.run(mc_tel._poll_contact_telemetry(_mc(), empty, stub, {}))
|
||||
assert activity.take_packet_count() == 0
|
||||
|
||||
|
||||
def test_poll_contact_telemetry_logs_initiation_and_empty_result(monkeypatch):
|
||||
"""Each poll logs initiation; an all-empty poll also logs 'returned no data'.
|
||||
|
||||
``req_telemetry_sync``/``req_status_sync`` return ``None`` on timeout without
|
||||
raising, so these two ``meshcore.telemetry.poll`` debug lines are the only
|
||||
signal separating an unanswered on-air poll from a disabled poll loop.
|
||||
"""
|
||||
import types
|
||||
|
||||
mc_tel, iface, stub, captured = _telemetry_env(
|
||||
monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}]
|
||||
)
|
||||
logs: list = []
|
||||
monkeypatch.setattr(
|
||||
mc_tel.config,
|
||||
"_debug_log",
|
||||
lambda message, **meta: logs.append((message, meta)),
|
||||
)
|
||||
|
||||
class _Silent:
|
||||
async def req_telemetry_sync(self, contact):
|
||||
return None
|
||||
|
||||
async def req_status_sync(self, contact):
|
||||
return None
|
||||
|
||||
asyncio.run(
|
||||
mc_tel._poll_contact_telemetry(
|
||||
types.SimpleNamespace(commands=_Silent()), iface, stub, {}
|
||||
)
|
||||
)
|
||||
|
||||
assert captured == []
|
||||
node_id = iface.lookup_node_id(_TEST_CONTACT_KEY[:12])
|
||||
assert [message for message, _meta in logs] == [
|
||||
"MeshCore contact telemetry poll initiated",
|
||||
"MeshCore contact telemetry poll returned no data",
|
||||
]
|
||||
assert all(meta["context"] == "meshcore.telemetry.poll" for _msg, meta in logs)
|
||||
assert all(meta["node_id"] == node_id for _msg, meta in logs)
|
||||
|
||||
|
||||
def test_telemetry_poll_loop_disabled_and_ticking(monkeypatch):
|
||||
"""The poll loop exits when disabled and fires both poll kinds when enabled."""
|
||||
import types
|
||||
@@ -4884,3 +4996,119 @@ def test_run_meshcore_autoadd_set_rejected_logs_warning(monkeypatch):
|
||||
and kw.get("autoadd_config") == 0x01
|
||||
for _msg, kw in logs
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# send_channel_announcement (SPEC MA6/MA9): optional duck-typed provider TX
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_send_channel_announcement_meshtastic_sends_and_counts():
|
||||
"""Meshtastic sends on CHANNEL_INDEX and counts the transmission (MA1)."""
|
||||
import data.mesh_ingestor.activity as activity
|
||||
import data.mesh_ingestor.config as config
|
||||
|
||||
calls = []
|
||||
|
||||
class _Iface:
|
||||
def sendText(self, text, channelIndex=0):
|
||||
calls.append((text, channelIndex))
|
||||
|
||||
activity.take_packet_count() # drain
|
||||
MeshtasticProvider().send_channel_announcement(_Iface(), "hello mesh")
|
||||
assert calls == [("hello mesh", config.CHANNEL_INDEX)]
|
||||
assert activity.take_packet_count() == 1
|
||||
|
||||
|
||||
def test_send_channel_announcement_meshtastic_noop_without_sendtext():
|
||||
"""An interface lacking sendText is a no-op that counts no TX."""
|
||||
import data.mesh_ingestor.activity as activity
|
||||
|
||||
activity.take_packet_count()
|
||||
MeshtasticProvider().send_channel_announcement(object(), "hi")
|
||||
assert activity.take_packet_count() == 0
|
||||
|
||||
|
||||
def test_send_channel_announcement_meshcore_sends_and_counts():
|
||||
"""MeshCore schedules send_chan_msg on its loop and counts the TX (MA1)."""
|
||||
import asyncio as _asyncio
|
||||
import threading as _threading
|
||||
import types as _types
|
||||
import data.mesh_ingestor.activity as activity
|
||||
import data.mesh_ingestor.config as config
|
||||
|
||||
iface = _MeshcoreInterface(target=None)
|
||||
sent = []
|
||||
|
||||
class _Commands:
|
||||
async def send_chan_msg(self, chan, msg, timestamp=None):
|
||||
sent.append((chan, msg))
|
||||
return _types.SimpleNamespace()
|
||||
|
||||
iface._mc = _types.SimpleNamespace(commands=_Commands())
|
||||
|
||||
loop = _asyncio.new_event_loop()
|
||||
thread = _threading.Thread(target=loop.run_forever, daemon=True)
|
||||
thread.start()
|
||||
iface._loop = loop
|
||||
try:
|
||||
activity.take_packet_count() # drain
|
||||
MeshcoreProvider().send_channel_announcement(iface, "hello mesh")
|
||||
assert sent == [(config.CHANNEL_INDEX, "hello mesh")]
|
||||
assert activity.take_packet_count() == 1
|
||||
finally:
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
thread.join(timeout=5)
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_send_channel_announcement_meshcore_noop_guards():
|
||||
"""No-op (no TX) for a wrong iface type or a missing/closed loop or handle."""
|
||||
import asyncio as _asyncio
|
||||
import types as _types
|
||||
import data.mesh_ingestor.activity as activity
|
||||
|
||||
provider = MeshcoreProvider()
|
||||
activity.take_packet_count() # drain
|
||||
|
||||
# Wrong interface type.
|
||||
provider.send_channel_announcement(object(), "x")
|
||||
# Missing mc / loop (fresh interface).
|
||||
provider.send_channel_announcement(_MeshcoreInterface(target=None), "x")
|
||||
# Closed loop.
|
||||
closed = _MeshcoreInterface(target=None)
|
||||
loop = _asyncio.new_event_loop()
|
||||
loop.close()
|
||||
closed._mc = _types.SimpleNamespace(commands=_types.SimpleNamespace())
|
||||
closed._loop = loop
|
||||
provider.send_channel_announcement(closed, "x")
|
||||
|
||||
assert activity.take_packet_count() == 0
|
||||
|
||||
|
||||
def test_send_channel_announcement_is_optional_MeshProtocol_member():
|
||||
"""The send method is an optional duck-typed extension: both providers expose
|
||||
it and still satisfy MeshProtocol, but it is NOT a required member — a minimal
|
||||
conforming provider without it still passes isinstance (MA9 / A4b)."""
|
||||
assert hasattr(MeshtasticProvider(), "send_channel_announcement")
|
||||
assert hasattr(MeshcoreProvider(), "send_channel_announcement")
|
||||
assert isinstance(MeshtasticProvider(), MeshProtocol)
|
||||
assert isinstance(MeshcoreProvider(), MeshProtocol)
|
||||
|
||||
class _Minimal:
|
||||
name = "minimal"
|
||||
|
||||
def subscribe(self):
|
||||
return []
|
||||
|
||||
def connect(self, *, active_candidate):
|
||||
return (None, None, None)
|
||||
|
||||
def extract_host_node_id(self, iface):
|
||||
return None
|
||||
|
||||
def node_snapshot_items(self, iface):
|
||||
return []
|
||||
|
||||
assert isinstance(_Minimal(), MeshProtocol)
|
||||
assert not hasattr(_Minimal(), "send_channel_announcement")
|
||||
|
||||
@@ -67,6 +67,8 @@ module PotatoMesh
|
||||
SQL
|
||||
end
|
||||
|
||||
record_ingestor_activity(db, node_id, last_seen_time, payload["packets"], protocol)
|
||||
|
||||
true
|
||||
rescue SQLite3::SQLException => e
|
||||
warn_log(
|
||||
@@ -78,6 +80,41 @@ module PotatoMesh
|
||||
)
|
||||
false
|
||||
end
|
||||
|
||||
# Append a per-heartbeat activity row for the moving-average time-series
|
||||
# (SPEC MA2/MA3), called after the ingestor liveness upsert.
|
||||
#
|
||||
# A missing, non-numeric, or negative +packets+ value records **no** row:
|
||||
# the field is additive, so older ingestors that never send it simply
|
||||
# contribute nothing to the time-series. A failure here is logged and
|
||||
# swallowed rather than propagated — the supplementary activity write must
|
||||
# never sink a heartbeat whose liveness upsert has already committed.
|
||||
#
|
||||
# @param db [SQLite3::Database] open database handle.
|
||||
# @param ingestor_id [String] canonical ingestor node id.
|
||||
# @param at [Integer] heartbeat timestamp used as the activity bucket time.
|
||||
# @param raw_packets [Object] the payload's +packets+ value (may be nil).
|
||||
# @param protocol [String] the ingestor's declared protocol.
|
||||
# @return [void]
|
||||
def record_ingestor_activity(db, ingestor_id, at, raw_packets, protocol)
|
||||
packets = coerce_integer(raw_packets)
|
||||
return if packets.nil? || packets.negative?
|
||||
|
||||
with_busy_retry do
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES(?,?,?,?)",
|
||||
[ingestor_id, at, packets, protocol],
|
||||
)
|
||||
end
|
||||
rescue SQLite3::SQLException => e
|
||||
warn_log(
|
||||
"Failed to record ingestor activity",
|
||||
context: "data_processing.ingestors",
|
||||
node_id: ingestor_id,
|
||||
error_class: e.class.name,
|
||||
error_message: e.message,
|
||||
)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@@ -110,7 +110,7 @@ module PotatoMesh
|
||||
def init_db
|
||||
FileUtils.mkdir_p(File.dirname(PotatoMesh::Config.db_path))
|
||||
db = open_database
|
||||
%w[nodes messages positions telemetry neighbors instances traces ingestors].each do |schema|
|
||||
%w[nodes messages positions telemetry neighbors instances traces ingestors ingestor_activity].each do |schema|
|
||||
sql_file = File.expand_path("../../../../data/#{schema}.sql", __dir__)
|
||||
db.execute_batch(File.read(sql_file))
|
||||
end
|
||||
@@ -537,6 +537,17 @@ module PotatoMesh
|
||||
db.execute("UPDATE ingestors SET protocol = 'meshtastic' WHERE protocol IS NULL OR TRIM(protocol) = ''")
|
||||
end
|
||||
end
|
||||
|
||||
# The per-heartbeat activity time-series (SPEC MA3) is a standalone,
|
||||
# append-only table; older installations gain it here without any data
|
||||
# backfill (the moving average simply starts populating from the next
|
||||
# heartbeat that carries a `packets` count).
|
||||
activity_tables =
|
||||
db.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ingestor_activity'").flatten
|
||||
if activity_tables.empty?
|
||||
activity_schema = File.expand_path("../../../../data/ingestor_activity.sql", __dir__)
|
||||
db.execute_batch(File.read(activity_schema))
|
||||
end
|
||||
rescue SQLite3::SQLException, Errno::ENOENT => e
|
||||
warn_log(
|
||||
"Failed to apply schema upgrade",
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
require_relative "queries/common"
|
||||
require_relative "queries/node_queries"
|
||||
require_relative "queries/ingestor_queries"
|
||||
require_relative "queries/chat_queries"
|
||||
require_relative "queries/telemetry_queries"
|
||||
require_relative "queries/federation_queries"
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# frozen_string_literal: true
|
||||
|
||||
module PotatoMesh
|
||||
module App
|
||||
module Queries
|
||||
# Rolling window used for the packets/hour moving average (SPEC MA4): each
|
||||
# ingestor's packet total over the last 24 hours.
|
||||
PACKETS_PER_HOUR_WINDOW_SECONDS = 86_400
|
||||
|
||||
# Fixed denominator that turns a window total into an hourly rate. Derived
|
||||
# from the window (24 h) rather than the *elapsed* span so the rate stays
|
||||
# stable and never spikes for a freshly-started ingestor with only a few
|
||||
# hours of data (SPEC MA4). The announcement only fires ≥ 24 h after start
|
||||
# (MA7), so by then a live ingestor has a full window anyway.
|
||||
PACKETS_PER_HOUR_DIVISOR = PACKETS_PER_HOUR_WINDOW_SECONDS / 3600.0
|
||||
|
||||
# Compute the mesh-wide packets/hour moving average per protocol scope
|
||||
# (SPEC MA4/MA5), aggregated **MAX-per-protocol** across ingestors.
|
||||
#
|
||||
# A single radio can only hear ≤ what is actually transmitted, so the
|
||||
# busiest single vantage is the best dedup-free estimate of unique air
|
||||
# traffic and can never double-count a frame heard by two radios (true
|
||||
# per-frame dedup is impossible — ignored/errored frames carry no id).
|
||||
# For each protocol the rate is
|
||||
# +MAX(ingestor's 24 h packet total) ÷ 24+, rounded; +total+ is the same
|
||||
# MAX taken over **every** ingestor regardless of protocol (so an ingestor
|
||||
# reporting several protocols contributes its combined total). +reticulum+
|
||||
# is a forward-looking always-zero stub (SPEC S6/MA5).
|
||||
#
|
||||
# @param now [Integer] reference unix timestamp in seconds.
|
||||
# @param db [SQLite3::Database, nil] optional open database handle to reuse.
|
||||
# @return [Hash{String => Integer}] +{ "total", "meshcore", "meshtastic",
|
||||
# "reticulum" }+ => rounded packets/hour.
|
||||
def query_packets_per_hour(now: Time.now.to_i, db: nil)
|
||||
handle = db || open_database(readonly: true)
|
||||
handle.results_as_hash = true
|
||||
reference_now = coerce_integer(now) || Time.now.to_i
|
||||
cutoff = reference_now - PACKETS_PER_HOUR_WINDOW_SECONDS
|
||||
|
||||
rows = with_busy_retry do
|
||||
handle.execute(
|
||||
"SELECT ingestor_id, protocol AS p, SUM(packets) AS total " \
|
||||
"FROM ingestor_activity WHERE at >= ? GROUP BY ingestor_id, protocol",
|
||||
[cutoff],
|
||||
)
|
||||
end
|
||||
|
||||
per_protocol_max = Hash.new(0)
|
||||
per_ingestor_total = Hash.new(0)
|
||||
rows.each do |row|
|
||||
protocol = row["p"]
|
||||
total = row["total"].to_i
|
||||
per_protocol_max[protocol] = [per_protocol_max[protocol], total].max if protocol
|
||||
per_ingestor_total[row["ingestor_id"]] += total
|
||||
end
|
||||
|
||||
{
|
||||
"total" => packets_per_hour_rate(per_ingestor_total.values.max || 0),
|
||||
"meshcore" => packets_per_hour_rate(per_protocol_max["meshcore"]),
|
||||
"meshtastic" => packets_per_hour_rate(per_protocol_max["meshtastic"]),
|
||||
"reticulum" => 0,
|
||||
}
|
||||
ensure
|
||||
handle&.close unless db
|
||||
end
|
||||
|
||||
# Convert a 24-hour packet total into a rounded packets/hour rate (MA4).
|
||||
#
|
||||
# @param total_packets [Integer] packets observed over the 24 h window.
|
||||
# @return [Integer] rounded hourly rate.
|
||||
def packets_per_hour_rate(total_packets)
|
||||
(total_packets / PACKETS_PER_HOUR_DIVISOR).round
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -51,6 +51,7 @@ module PotatoMesh
|
||||
["telemetry", "rx_time"],
|
||||
["traces", "rx_time"],
|
||||
["ingestors", "last_seen_time"],
|
||||
["ingestor_activity", "at"],
|
||||
["nodes", "last_heard"],
|
||||
].freeze
|
||||
|
||||
|
||||
@@ -115,9 +115,16 @@ module PotatoMesh
|
||||
content_type :json
|
||||
priv = private_mode? ? 1 : 0
|
||||
cached = PotatoMesh::App::ApiCache.fetch("api:stats:#{priv}", ttl_seconds: 15) do
|
||||
# Scope → metric → window tree (SPEC S1). +sampled+ stays last and
|
||||
# Scope → metric → window tree (SPEC S1) plus the additive
|
||||
# +packets_per_hour+ moving-average map (SPEC MA5); both are drawn
|
||||
# from independent read-side queries. +sampled+ stays last and
|
||||
# +false+ for backward continuity with the prior payload.
|
||||
query_active_node_stats.merge("sampled" => false).to_json
|
||||
query_active_node_stats
|
||||
.merge(
|
||||
"packets_per_hour" => query_packets_per_hour,
|
||||
"sampled" => false,
|
||||
)
|
||||
.to_json
|
||||
end
|
||||
|
||||
etag cached[:etag], kind: :weak
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
# Copyright © 2025-26 l5yth & contributors
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# frozen_string_literal: true
|
||||
|
||||
require "spec_helper"
|
||||
require "json"
|
||||
require "time"
|
||||
|
||||
# Web-side coverage for the per-heartbeat activity time-series (SPEC MA3) and
|
||||
# the additive +packets+ delta on the ingestor heartbeat (MA2). A heartbeat
|
||||
# carrying a non-negative integer +packets+ appends exactly one append-only
|
||||
# +ingestor_activity+ row; anything else records nothing.
|
||||
RSpec.describe "Ingestor activity time-series (MA3)" do
|
||||
let(:app) { Sinatra::Application }
|
||||
let(:api_token) { "secret-token" }
|
||||
let(:auth_headers) do
|
||||
{
|
||||
"CONTENT_TYPE" => "application/json",
|
||||
"HTTP_AUTHORIZATION" => "Bearer #{api_token}",
|
||||
}
|
||||
end
|
||||
|
||||
before do
|
||||
@original_token = ENV["API_TOKEN"]
|
||||
ENV["API_TOKEN"] = api_token
|
||||
clear_tables
|
||||
end
|
||||
|
||||
after do
|
||||
ENV["API_TOKEN"] = @original_token
|
||||
clear_tables
|
||||
end
|
||||
|
||||
def clear_tables
|
||||
with_db do |db|
|
||||
db.execute("DELETE FROM ingestor_activity")
|
||||
db.execute("DELETE FROM ingestors")
|
||||
end
|
||||
end
|
||||
|
||||
def with_db(readonly: false)
|
||||
db = PotatoMesh::Application.open_database(readonly: readonly)
|
||||
db.busy_timeout = PotatoMesh::Config.db_busy_timeout_ms
|
||||
db.execute("PRAGMA foreign_keys = ON")
|
||||
yield db
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
|
||||
def post_heartbeat(overrides = {})
|
||||
now = Time.now.to_i
|
||||
payload = {
|
||||
node_id: "!abc12345",
|
||||
start_time: now - 3600,
|
||||
last_seen_time: now - 60,
|
||||
version: "0.6.0",
|
||||
protocol: "meshtastic",
|
||||
}.merge(overrides)
|
||||
post "/api/ingestors", payload.to_json, auth_headers
|
||||
end
|
||||
|
||||
def activity_rows
|
||||
with_db(readonly: true) do |db|
|
||||
db.execute(
|
||||
"SELECT ingestor_id, at, packets, protocol FROM ingestor_activity ORDER BY id",
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
describe "POST /api/ingestors" do
|
||||
it "records ingestor activity for a heartbeat carrying packets" do
|
||||
now = Time.now.to_i
|
||||
post_heartbeat(last_seen_time: now - 60, packets: 42, protocol: "meshcore")
|
||||
expect(last_response.status).to eq(201)
|
||||
|
||||
rows = activity_rows
|
||||
expect(rows.length).to eq(1)
|
||||
ingestor_id, at, packets, protocol = rows.first
|
||||
expect(ingestor_id).to eq("!abc12345")
|
||||
expect(at).to eq(now - 60)
|
||||
expect(packets).to eq(42)
|
||||
expect(protocol).to eq("meshcore")
|
||||
end
|
||||
|
||||
it "records ingestor activity of zero for an idle heartbeat" do
|
||||
post_heartbeat(packets: 0)
|
||||
expect(last_response.status).to eq(201)
|
||||
rows = activity_rows
|
||||
expect(rows.length).to eq(1)
|
||||
expect(rows.first[2]).to eq(0)
|
||||
end
|
||||
|
||||
it "records ingestor activity independently per ingestor" do
|
||||
post_heartbeat(node_id: "!abc12345", packets: 10)
|
||||
post_heartbeat(node_id: "!def67890", packets: 25)
|
||||
expect(activity_rows.map { |r| [r[0], r[2]] }).to contain_exactly(
|
||||
["!abc12345", 10],
|
||||
["!def67890", 25],
|
||||
)
|
||||
end
|
||||
|
||||
it "appends a fresh activity row on every heartbeat (append-only)" do
|
||||
post_heartbeat(node_id: "!abc12345", packets: 3)
|
||||
post_heartbeat(node_id: "!abc12345", packets: 4)
|
||||
expect(activity_rows.map { |r| r[2] }).to eq([3, 4])
|
||||
end
|
||||
|
||||
it "records no activity when packets is absent (older ingestor)" do
|
||||
post_heartbeat
|
||||
expect(last_response.status).to eq(201)
|
||||
expect(activity_rows).to be_empty
|
||||
end
|
||||
|
||||
it "records no activity when packets is negative or non-numeric" do
|
||||
post_heartbeat(node_id: "!abc12345", packets: -5)
|
||||
post_heartbeat(node_id: "!def67890", packets: "abc")
|
||||
expect(activity_rows).to be_empty
|
||||
end
|
||||
|
||||
it "still records the heartbeat when the activity insert fails" do
|
||||
# Force the supplementary activity INSERT to raise by removing its table,
|
||||
# then prove the liveness heartbeat still returns 201 (graceful
|
||||
# degradation). The table is restored so the surrounding suite is
|
||||
# unaffected.
|
||||
with_db { |db| db.execute("ALTER TABLE ingestor_activity RENAME TO ingestor_activity_bak") }
|
||||
begin
|
||||
post_heartbeat(packets: 99)
|
||||
expect(last_response.status).to eq(201)
|
||||
ensure
|
||||
with_db do |db|
|
||||
db.execute("ALTER TABLE ingestor_activity_bak RENAME TO ingestor_activity")
|
||||
end
|
||||
end
|
||||
expect(activity_rows).to be_empty
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -108,6 +108,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
db.execute("DELETE FROM positions")
|
||||
db.execute("DELETE FROM telemetry")
|
||||
db.execute("DELETE FROM ingestors")
|
||||
db.execute("DELETE FROM ingestor_activity")
|
||||
end
|
||||
ensure_self_instance_record!
|
||||
end
|
||||
@@ -6709,6 +6710,45 @@ RSpec.describe "Potato Mesh Sinatra app" do
|
||||
end
|
||||
|
||||
describe "GET /api/stats" do
|
||||
it "exposes the additive packets_per_hour MAX-per-protocol map" do
|
||||
clear_database
|
||||
now = reference_time.to_i
|
||||
allow(Time).to receive(:now).and_return(reference_time)
|
||||
|
||||
with_db do |db|
|
||||
# meshcore: busiest ingestor 1200 pkts/24h ⇒ 50/h; a quieter second
|
||||
# ingestor of the same protocol must not inflate the MAX.
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
["!core0001", now - 100, 1200, "meshcore"],
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
["!core0002", now - 100, 720, "meshcore"],
|
||||
)
|
||||
# meshtastic: 720 pkts/24h ⇒ 30/h.
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
["!tast0001", now - 100, 720, "meshtastic"],
|
||||
)
|
||||
end
|
||||
|
||||
get "/api/stats"
|
||||
|
||||
expect(last_response).to be_ok
|
||||
payload = JSON.parse(last_response.body)
|
||||
pph = payload["packets_per_hour"]
|
||||
expect(pph.keys).to contain_exactly("total", "meshcore", "meshtastic", "reticulum")
|
||||
expect(pph["meshcore"]).to eq(50)
|
||||
expect(pph["meshtastic"]).to eq(30)
|
||||
expect(pph["total"]).to eq(50) # MAX over every ingestor = 1200 / 24
|
||||
expect(pph["reticulum"]).to eq(0)
|
||||
# The additive field leaves the S1 scope × metric × window tree intact.
|
||||
expect(payload["sampled"]).to eq(false)
|
||||
expect(payload["total"]).to have_key("nodes")
|
||||
expect(payload).not_to have_key("active_nodes")
|
||||
end
|
||||
|
||||
it "returns exact SQL-backed activity counts with per-protocol breakdowns" do
|
||||
clear_database
|
||||
now = reference_time.to_i
|
||||
|
||||
@@ -1263,6 +1263,86 @@ RSpec.describe PotatoMesh::App::Queries do
|
||||
end
|
||||
end
|
||||
|
||||
describe "#query_packets_per_hour" do
|
||||
before do
|
||||
with_db { |db| db.execute("DELETE FROM ingestor_activity") }
|
||||
end
|
||||
|
||||
after do
|
||||
with_db { |db| db.execute("DELETE FROM ingestor_activity") }
|
||||
end
|
||||
|
||||
def seed_activity(rows)
|
||||
with_db do |db|
|
||||
rows.each do |ingestor_id, at, packets, protocol|
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
[ingestor_id, at, packets, protocol],
|
||||
)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
it "returns all zeros when there is no activity" do
|
||||
expect(queries.query_packets_per_hour(now: now)).to eq(
|
||||
"total" => 0, "meshcore" => 0, "meshtastic" => 0, "reticulum" => 0,
|
||||
)
|
||||
end
|
||||
|
||||
it "takes the MAX per protocol of each ingestor's 24h total over 24" do
|
||||
seed_activity(
|
||||
[
|
||||
# meshcore ingestor A: 1200 over the window (two heartbeats).
|
||||
["!coreaaaa", now - 100, 700, "meshcore"],
|
||||
["!coreaaaa", now - 3700, 500, "meshcore"],
|
||||
# meshcore ingestor B: 900 — the quieter vantage must not inflate MAX.
|
||||
["!corebbbb", now - 200, 900, "meshcore"],
|
||||
# meshtastic ingestor C: 720.
|
||||
["!tastcccc", now - 300, 720, "meshtastic"],
|
||||
],
|
||||
)
|
||||
result = queries.query_packets_per_hour(now: now)
|
||||
expect(result["meshcore"]).to eq(50) # MAX(1200, 900) / 24
|
||||
expect(result["meshtastic"]).to eq(30) # 720 / 24
|
||||
expect(result["total"]).to eq(50) # MAX over every ingestor = 1200 / 24
|
||||
expect(result["reticulum"]).to eq(0)
|
||||
end
|
||||
|
||||
it "reports zero for a protocol with no active ingestor" do
|
||||
seed_activity([["!coreaaaa", now - 100, 240, "meshcore"]])
|
||||
result = queries.query_packets_per_hour(now: now)
|
||||
expect(result["meshcore"]).to eq(10) # 240 / 24
|
||||
expect(result["meshtastic"]).to eq(0)
|
||||
expect(result["reticulum"]).to eq(0)
|
||||
end
|
||||
|
||||
it "excludes activity older than the 24h window" do
|
||||
seed_activity(
|
||||
[
|
||||
["!coreaaaa", now - 100, 480, "meshcore"], # inside → counts
|
||||
["!coreaaaa", now - 90_000, 100_000, "meshcore"], # >24h → excluded
|
||||
],
|
||||
)
|
||||
expect(queries.query_packets_per_hour(now: now)["meshcore"]).to eq(20) # 480 / 24
|
||||
end
|
||||
|
||||
it "keeps reticulum a zero stub even when reticulum activity exists" do
|
||||
seed_activity([["!reti0001", now - 100, 720, "reticulum"]])
|
||||
result = queries.query_packets_per_hour(now: now)
|
||||
# The reticulum scope is always emitted as zero (forward-looking stub)…
|
||||
expect(result["reticulum"]).to eq(0)
|
||||
# …but a reticulum ingestor still contributes to the protocol-agnostic
|
||||
# total (MAX across every ingestor).
|
||||
expect(result["total"]).to eq(30) # 720 / 24
|
||||
end
|
||||
|
||||
it "rounds the hourly rate to the nearest integer" do
|
||||
seed_activity([["!coreaaaa", now - 100, 100, "meshcore"]])
|
||||
# 100 / 24 = 4.166… → 4
|
||||
expect(queries.query_packets_per_hour(now: now)["meshcore"]).to eq(4)
|
||||
end
|
||||
end
|
||||
|
||||
describe "#query_telemetry_buckets" do
|
||||
it "clamps oversized window_seconds to the 28-day visibility cap" do
|
||||
huge_window = PotatoMesh::Config.four_weeks_seconds * 50
|
||||
|
||||
@@ -208,6 +208,37 @@ RSpec.describe PotatoMesh::App::Retention do
|
||||
expect(removed["ingestors"]).to eq(1)
|
||||
end
|
||||
|
||||
it "prunes ingestor_activity rows past the retention window" do
|
||||
fresh = now - 100
|
||||
stale = now - PotatoMesh::Config.year_seconds - 86_400
|
||||
|
||||
db = SQLite3::Database.new(PotatoMesh::Config.db_path)
|
||||
begin
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
["!fffff001", fresh, 42, "meshtastic"],
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO ingestor_activity(ingestor_id, at, packets, protocol) VALUES (?,?,?,?)",
|
||||
["!aaaaa001", stale, 7, "meshcore"],
|
||||
)
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
|
||||
removed = harness_class.purge_old_data!(now: now)
|
||||
expect(removed["ingestor_activity"]).to eq(1)
|
||||
|
||||
db = SQLite3::Database.new(PotatoMesh::Config.db_path, readonly: true)
|
||||
begin
|
||||
remaining =
|
||||
db.execute("SELECT ingestor_id FROM ingestor_activity ORDER BY ingestor_id").flatten
|
||||
expect(remaining).to eq(["!fffff001"])
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
end
|
||||
|
||||
it "keeps every row when the database is empty" do
|
||||
removed = harness_class.purge_old_data!(now: now)
|
||||
expect(removed.values.uniq).to eq([0])
|
||||
|
||||
Reference in New Issue
Block a user