mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-09-25 12:53:40 +00:00
## Summary Adds `GET /api/analytics/retransmissions` and a "Retransmission Pressure (proxy)" chart on the Analytics Topology tab, implementing the metric agreed in #1699: for each flood, the number of distinct repeaters in the union of the paths of all its observations (`[A]`, `[A,B,C]`, `[A,D]` gives 4), averaged per time bucket. Topology is the tab that already shows hop counts and repeaters in paths, so the chart sits there instead of in a new tab. ## Definition - Flood routes only (`route_type` 0/1); TRACE excluded. Direct routes carry the route still to travel (firmware `src/Mesh.cpp:78-106,334-342`), zero-hop sends are direct (`src/Mesh.cpp:717-737`), TRACE path bytes are SNR values (`src/Mesh.cpp:59-61`, refused by `sendFlood` at `src/Mesh.cpp:637-641`). Firmware commit 0679dbef. - **Flood events, not hashes.** `transmissions.hash` is UNIQUE and the packet hash excludes the path (`src/Packet.cpp:41-50`), so when the same bytes flood again the observations land on the same transmission. Observations are sorted by time and split into events wherever two consecutive observations are more than 5 minutes apart. Each event is counted on its own and bucketed by its first observation. - Why 5 minutes: a node holds a flood for at most 32 s (`src/Dispatcher.cpp:11,243-251`) plus a random retransmit delay. On live over 7 days, 72,806 of 74,347 flood transmissions span 60 s or less, and of 1,372,283 consecutive observation gaps, 52 fall between 60 s and 300 s against 1,823 above 300 s. - Events that start before the store retention floor (now minus `retentionHours`) are left out for every request shape. The store keeps older observations only for hashes heard again recently, so they do not represent that period. Eviction of those transmissions is tracked in #2024. - A flood event heard only with an empty path counts as 0 repeaters. - **Prefixes are not resolved to nodes, and a prefix counts once per event**, whether it repeats across observations or inside one path. On live (7 days), a repeated 2-byte prefix inside one path occurs in 1.08% of flood transmissions and 6,178 of 6,596 such repeats match exactly one known node; for 3-byte it is 0.69% and 104 of 104. That is one node forwarding again after its 160-slot cyclic duplicate filter dropped the hash (`src/helpers/SimpleMeshTables.h:9,52-57`). A repeated 1-byte prefix (44.9% of 1-byte transmissions) is mostly two nodes; counting it once keeps the value a lower bound. `summary.one_byte_packets` reports how many events that affects. - Observations are stored once per observer and path per hash, so a later event of the same hash only holds pairs not stored before; its count is a lower bound too. On live these are 1,466 of 75,356 events (1.9%), and they are kept in the average. - Resolution was not used: on live, 1-byte observations nearly all have `resolved_path` NULL, and cold load refuses context-based resolution of history (`cmd/server/neighbor_persist.go:155-168`). - Buckets `5m|15m|1h|6h|1d`. `region` filters on observers like `/api/analytics/rf`, after the event split; a region with no known observers is not filtered, the same as the other analytics endpoints. `area` is not supported. ## Implementation - `cmd/server/retransmission_pressure.go:255` `addPath`: scans path JSON directly into a generation-stamped hash set, no allocation per observation. - `cmd/server/retransmission_pressure.go:367` `computeRetransmissionPressure`: one pass under `s.mu.RLock`. Per flood transmission it sorts the observations by cached parsed time into a reused scratch slice, splits events and counts each in `addEvent` (`:319`). O(T + O log k + H). - `cmd/server/retransmission_pressure.go:470` `GetRetransmissionPressure`: default shape from the recomputer (#1659 warm-up gate). Other shapes come from a typed TTL cache (max 64 entries) cleared on new paths and eviction (`cmd/server/store.go:2289,2336`); concurrent misses on one key share one compute through singleflight (`store.go:204`). - `cmd/server/retransmission_pressure.go:515` handler, `cmd/server/routes.go:331`, `cmd/server/openapi.go:108`, `docs/api-spec.md:1283`. - `public/analytics.js:761` card, `:855` `renderRetransmissionChart` (CSS variables only, lines break at missing buckets, caption states it is a proxy, names the observer coverage bias, the once-per-flood prefix rule and the 5 minute event split), `:829` loader with stale-response guard. ## Performance - `BenchmarkComputeRetransmissionPressure`, 50k transmissions x 20 observations, `-cpu 1`, i5-1335U: median 161 ms/op (132 ms/op before the event split); first pass after startup with timestamps not yet parsed 192 ms/op. About 22 KB and 281 allocations per op. - On staging the default shape is served from the recomputer in 0.3 s; the post-load recompute of this recomputer took 994 ms on a 121k-transmission store (log line quoted in #2025). A 336h store would be about twice that, every recompute interval, under the store read lock. ## Tests - `cmd/server/retransmission_pressure_test.go`: union counting (reporter example, overlaps, once per event for 1/2/3-byte, width/case, growth); route/TRACE/zero-hop filter, bucketing, window by event start; event split and the 5 minute settle gap (boundary, chained steps, unsorted input), retention floor; region filter, region applied after the split, unknown region, 1-byte share; recomputer read, TTL cache invalidation on new paths and on eviction, cache expiry, singleflight, recomputer gate wiring, handler, warm-up gate. - `test-issue-1699-retransmission-chart.js` (33 tests, registered in `test-all.sh` and `deploy.yml`). - Mutation-checked: 18 mutations of the event split, floor, prefix rule, bucketing, region order, cache clears, expiry, gate wiring and singleflight, all killed. - `go test ./...` in cmd/server passes; `scripts/check-css-vars.js` OK. ## Staging validation Build `c646310f` (this rework plus #2025 and the other review follow-ups), after a container restart and full load: default shape 74,974 flood events, average 27.08 repeaters, 169 hourly buckets from 2026-09-06 16:00 (the 168h floor) to the current hour, highest hourly average 53.3. Before the rework the same instance showed buckets back to 2026-07-18, averages up to 148, and for the first minutes after a restart only 5,911 packets. ## Merge order with #2025 #2025 fixes the recomputer startup for all analytics endpoints (the stale first snapshot seen here). Whichever of the two merges second has to add `recompRetransmissions` to `analyticsRecomputersLocked`, wire it to that PR's `loadedGate` instead of `LoadComplete`, bump the recomputer count in `TestAnalyticsRecomputers_PostLoadOrder` from 9 to 10, and make `TestStartAnalyticsRecomputers_RetransmissionsGatedOnLoadComplete` call `signalStartupLoadDone()`. That resolution is what ran on staging. ## Not verified - Recompute timing on a production-size (336h) store; only extrapolated. - Phone-width layout and dark theme of the reworked chart. - E2E Playwright suite. Fixes #1699 --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
efiten
@@ -221,8 +221,8 @@ func TestAnalyticsRecomputers_PostLoadOrder(t *testing.T) {
|
||||
for i, rc := range list {
|
||||
pos[rc.name] = i
|
||||
}
|
||||
if len(list) != 9 || len(pos) != 9 {
|
||||
t.Fatalf("want 9 distinct recomputers, got %d (%d distinct)", len(list), len(pos))
|
||||
if len(list) != 10 || len(pos) != 10 {
|
||||
t.Fatalf("want 10 distinct recomputers, got %d (%d distinct)", len(list), len(pos))
|
||||
}
|
||||
for _, name := range []string{"rf", "topology", "channels"} {
|
||||
if pos[name] > 2 {
|
||||
|
||||
@@ -258,6 +258,7 @@ func (s *PacketStore) analyticsRecomputersLocked() []*analyticsRecomputer {
|
||||
s.recompDistance, s.recompHashCollisions, s.recompHashSizes,
|
||||
s.recompObserversClockSkew, s.recompNodesClockSkew,
|
||||
s.recompRoles,
|
||||
s.recompRetransmissions,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -331,6 +332,12 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o
|
||||
"nodes-clock-skew", pickInterval(ov.NodesClockSkew, defaultInterval),
|
||||
func() interface{} { return s.computeFleetClockSkew() },
|
||||
)
|
||||
s.recompRetransmissions = newAnalyticsRecomputer(
|
||||
"retransmissions", defaultInterval,
|
||||
func() interface{} {
|
||||
return s.computeRetransmissionPressure("", TimeWindow{}, retransmissionDefaultBucket)
|
||||
},
|
||||
)
|
||||
all := s.analyticsRecomputersLocked()
|
||||
s.analyticsRecomputerMu.Unlock()
|
||||
|
||||
@@ -352,6 +359,7 @@ func (s *PacketStore) StartAnalyticsRecomputers(defaultInterval time.Duration, o
|
||||
s.recompRF.setWarmupReadyGate_1659(loadedGate)
|
||||
s.recompTopology.setWarmupReadyGate_1659(loadedGate)
|
||||
s.recompChannels.setWarmupReadyGate_1659(loadedGate)
|
||||
s.recompRetransmissions.setWarmupReadyGate_1659(loadedGate)
|
||||
|
||||
for _, rc := range all {
|
||||
rc.Start()
|
||||
|
||||
@@ -109,6 +109,14 @@ func routeDescriptions() map[string]routeMeta {
|
||||
"GET /api/analytics/subpaths-bulk": {Summary: "Bulk subpath analysis", Tag: "analytics"},
|
||||
"GET /api/analytics/subpath-detail": {Summary: "Subpath detail", Tag: "analytics"},
|
||||
"GET /api/analytics/neighbor-graph": {Summary: "Neighbor graph", Description: "Full neighbor affinity graph for visualization.", Tag: "analytics"},
|
||||
"GET /api/analytics/retransmissions": {Summary: "Retransmission pressure over time", Description: "Collision-pressure proxy (#1699): per time bucket, the average number of distinct repeaters in the union of all observed paths of each flood event (route types 0/1, TRACE excluded). A transmission's observations are split into flood events at gaps of more than 5 minutes; each event is bucketed by its first observation, and events before the store retention floor are left out. Hop prefixes are not resolved: a prefix counts once per event, so colliding 1-byte prefixes make this a lower bound. Only repeaters some observer heard are counted, so the value also follows observer coverage; each bucket carries its observer count.", Tag: "analytics",
|
||||
QueryParams: []paramMeta{
|
||||
{Name: "region", Description: "Comma-separated IATA codes; only observations from observers in the region are counted. A region with no known observers is not filtered", Type: "string"},
|
||||
{Name: "window", Description: "Relative window: 1h, 24h, 3d, 7d or 30d", Type: "string"},
|
||||
{Name: "from", Description: "Absolute window start (RFC3339)", Type: "string"},
|
||||
{Name: "to", Description: "Absolute window end (RFC3339)", Type: "string"},
|
||||
{Name: "bucket", Description: "Bucket size: 5m, 15m, 1h, 6h or 1d (default 1h)", Type: "string"},
|
||||
}},
|
||||
|
||||
// Channels
|
||||
"GET /api/channels": {Summary: "List channels", Description: "Returns known mesh channels with message counts.", Tag: "channels"},
|
||||
|
||||
@@ -0,0 +1,534 @@
|
||||
package main
|
||||
|
||||
// retransmission_pressure.go: issue #1699.
|
||||
//
|
||||
// Network-wide time series of how many distinct repeaters took part in
|
||||
// relaying each flood, as a proxy for collision pressure. Definition agreed
|
||||
// in the #1699 thread: for one flood, take the union of the paths of ALL its
|
||||
// observations and count the distinct repeaters in it (paths [A], [A,B,C],
|
||||
// [A,D] -> 4). Per time bucket we report the average of that count over the
|
||||
// floods that started in the bucket.
|
||||
//
|
||||
// It is a proxy, not a measured collision rate: a repeater that forwarded a
|
||||
// packet no observer heard is invisible, so the number moves with observer
|
||||
// coverage too. The response carries the per-bucket observer count so the UI
|
||||
// can show that.
|
||||
//
|
||||
// Protocol facts (MeshCore firmware, commit 0679dbef):
|
||||
// - Only flood routes build a path of forwarders: routeRecvPacket appends
|
||||
// the forwarder's own hash to the end of the path (src/Mesh.cpp:344-356).
|
||||
// ROUTE_TYPE_TRANSPORT_FLOOD (0) and ROUTE_TYPE_FLOOD (1) are both flood
|
||||
// (src/Packet.h:14-15,64).
|
||||
// - Direct routes carry the route still to travel; each hop removes itself
|
||||
// (removeSelfFromPath, src/Mesh.cpp:78-106,334-342), so the observed path
|
||||
// is not the forwarders. Zero-hop sends are ROUTE_TYPE_DIRECT /
|
||||
// TRANSPORT_DIRECT with path_len 0 (src/Mesh.cpp:717-737). Both are
|
||||
// excluded by the route filter.
|
||||
// - TRACE never floods (sendFlood refuses it, src/Mesh.cpp:637-641) and its
|
||||
// path bytes are SNR values, not hashes (src/Mesh.cpp:59-61). Excluded
|
||||
// explicitly as well.
|
||||
// - A path entry is the first 1-3 bytes of the forwarder's public key; the
|
||||
// width is chosen by the originator and is the same for every hop of one
|
||||
// packet (src/Mesh.cpp:649, src/Packet.h:79-83, src/Identity.h:23-25).
|
||||
// - The duplicate filter (wasSeen/markSeen, e.g. src/Mesh.cpp:121-126) is a
|
||||
// cyclic buffer of 160 packet hashes (src/helpers/SimpleMeshTables.h:9,
|
||||
// 52-57). Once a hash is overwritten, the node forwards that packet again
|
||||
// if it comes back, so the same node can appear twice in one path.
|
||||
// - A node holds a received flood for at most 32 s before handling it
|
||||
// (MAX_RX_DELAY_MILLIS, src/Dispatcher.cpp:11,243-251), plus a random
|
||||
// retransmit delay of a few airtimes (examples/simple_repeater/
|
||||
// MyMesh.cpp:547-550).
|
||||
//
|
||||
// Flood events. transmissions.hash is UNIQUE and the packet hash excludes the
|
||||
// path (src/Packet.cpp:41-50), so when the same bytes are flooded again later
|
||||
// the new observations are appended to the existing transmission. The union
|
||||
// over all of them would merge separate floods. We sort a transmission's
|
||||
// observations by time and start a new event when the gap to the previous
|
||||
// observation exceeds retransmissionEventGap; each event is counted on its
|
||||
// own, in the bucket of its first observation.
|
||||
//
|
||||
// Retention floor. Events that start before now - retentionHours are dropped
|
||||
// for every request shape: the store keeps observations from before the
|
||||
// floor only for hashes that were heard again recently, so they are not the
|
||||
// traffic of that period.
|
||||
//
|
||||
// Prefixes are not resolved. A 1-byte prefix is shared by many repeaters, and
|
||||
// we do NOT resolve hops to public keys here: the resolved-pubkey index is
|
||||
// empty for observations whose resolved_path is NULL (on live nearly every
|
||||
// 1-byte observation), and context-based resolution of history is refused on
|
||||
// purpose elsewhere (resolvePathForObsColdLoad, PR #1643). Counting rule: a
|
||||
// prefix counts once per event, whether it repeats inside one path or across
|
||||
// paths. A repeated 2- or 3-byte prefix is one node forwarding twice (see the
|
||||
// duplicate filter above). A repeated 1-byte prefix can also be two nodes;
|
||||
// counting it once keeps the value a lower bound, as does merging repeaters
|
||||
// that share a prefix across paths. The share of 1-byte packets is reported
|
||||
// so the undercount can be judged.
|
||||
//
|
||||
// Region. region filters on the observers of that region, like
|
||||
// /api/analytics/rf. Events are split on all observations, so the filter
|
||||
// does not change where an event starts. A region with no known observers is
|
||||
// not filtered, the same as the other analytics endpoints.
|
||||
//
|
||||
// Complexity: one pass over s.packets under s.mu.RLock, O(T + O log k + H)
|
||||
// for T transmissions, O observations of flood packets (k per transmission,
|
||||
// sorted by time) and H hop entries. Timestamps are parsed once per
|
||||
// observation and cached (StoreObs.ParsedTime). Memory: one entry per
|
||||
// non-empty bucket, a per-pass observer index and scratch reused across
|
||||
// transmissions. Served from the analytics recomputer for the default shape
|
||||
// and from a TTL cache otherwise, never computed per request on a warm
|
||||
// cache; concurrent misses on one key share a single compute.
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"math/bits"
|
||||
"net/http"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RetransmissionBucket is one time bucket of the series.
|
||||
type RetransmissionBucket struct {
|
||||
Start string `json:"start"` // bucket start, RFC3339 UTC
|
||||
Packets int `json:"packets"` // flood events that started in the bucket
|
||||
RepeaterSum int `json:"repeater_sum"` // sum of distinct repeaters over those events
|
||||
AvgRepeaters float64 `json:"avg_repeaters"` // repeater_sum / packets
|
||||
Observers int `json:"observers"` // distinct observers that heard those events
|
||||
}
|
||||
|
||||
// RetransmissionSummary aggregates the whole response window.
|
||||
type RetransmissionSummary struct {
|
||||
Packets int `json:"packets"`
|
||||
AvgRepeaters float64 `json:"avg_repeaters"`
|
||||
Observers int `json:"observers"`
|
||||
OneBytePackets int `json:"one_byte_packets"` // events whose hops are 1-byte hashes (most ambiguous)
|
||||
NoRepeaterPackets int `json:"no_repeater_packets"` // events heard with an empty path only
|
||||
}
|
||||
|
||||
// RetransmissionResponse is the /api/analytics/retransmissions body.
|
||||
type RetransmissionResponse struct {
|
||||
BucketSeconds int `json:"bucket_seconds"`
|
||||
Window string `json:"window"`
|
||||
Region string `json:"region"`
|
||||
Summary RetransmissionSummary `json:"summary"`
|
||||
Buckets []RetransmissionBucket `json:"buckets"`
|
||||
}
|
||||
|
||||
type retransmissionCacheEntry struct {
|
||||
data RetransmissionResponse
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// retransmissionCacheMax bounds the TTL cache: ?from=&to= makes the key space
|
||||
// open-ended, and invalidation only runs when new paths arrive.
|
||||
const retransmissionCacheMax = 64
|
||||
|
||||
const retransmissionDefaultBucket = time.Hour
|
||||
|
||||
// retransmissionEventGap is the settle time that separates two flood events
|
||||
// of one transmission. It is well above the longest per-hop hold in firmware
|
||||
// (32 s plus a retransmit delay, see the file header), so a flood still
|
||||
// spreading is not cut. Gaps between re-floods of the same bytes range from
|
||||
// minutes to weeks; a re-flood within 5 minutes merges into one event.
|
||||
const retransmissionEventGap = 5 * time.Minute
|
||||
|
||||
// parseRetransmissionBucket maps the ?bucket= value to a duration. Unknown
|
||||
// values fall back to the default, matching how ParseTimeWindow ignores
|
||||
// invalid input.
|
||||
func parseRetransmissionBucket(v string) time.Duration {
|
||||
switch v {
|
||||
case "5m":
|
||||
return 5 * time.Minute
|
||||
case "15m":
|
||||
return 15 * time.Minute
|
||||
case "6h":
|
||||
return 6 * time.Hour
|
||||
case "1d":
|
||||
return 24 * time.Hour
|
||||
}
|
||||
return retransmissionDefaultBucket
|
||||
}
|
||||
|
||||
// repeaterUnion counts the distinct hop prefixes across the observed paths of
|
||||
// one flood event (see the counting rule in the file header). Hops are keyed
|
||||
// as (byte width, value) so case does not matter and a 1-byte "AB" differs
|
||||
// from a 2-byte "AB00". Reuse one value across events via reset().
|
||||
//
|
||||
// The set is an open-addressing hash table whose slots are stamped with a
|
||||
// generation, so reset() is O(1) instead of clearing the table.
|
||||
type repeaterUnion struct {
|
||||
slotKey []uint64
|
||||
slotGen []uint32 // slot is live when slotGen[i] == gen
|
||||
gen uint32
|
||||
used int
|
||||
width int // byte width of the first hop seen, 0 if none
|
||||
}
|
||||
|
||||
const repeaterUnionMinSlots = 256
|
||||
|
||||
func (u *repeaterUnion) reset() {
|
||||
if u.slotKey == nil {
|
||||
u.allocSlots(repeaterUnionMinSlots)
|
||||
}
|
||||
u.gen++
|
||||
if u.gen == 0 {
|
||||
clear(u.slotGen)
|
||||
u.gen = 1
|
||||
}
|
||||
u.used = 0
|
||||
u.width = 0
|
||||
}
|
||||
|
||||
func (u *repeaterUnion) allocSlots(n int) {
|
||||
u.slotKey = make([]uint64, n)
|
||||
u.slotGen = make([]uint32, n)
|
||||
}
|
||||
|
||||
// slot returns the index holding k, or the free index where k belongs.
|
||||
func (u *repeaterUnion) slot(k uint64) int {
|
||||
mask := len(u.slotKey) - 1
|
||||
i := int((k * 0x9E3779B97F4A7C15) >> 40 & uint64(mask))
|
||||
for u.slotGen[i] == u.gen && u.slotKey[i] != k {
|
||||
i = (i + 1) & mask
|
||||
}
|
||||
return i
|
||||
}
|
||||
|
||||
// grow doubles the table, keeping the live entries. Load stays below 1/2,
|
||||
// so slot() always finds a free index.
|
||||
func (u *repeaterUnion) grow() {
|
||||
oldKey, oldGen, gen := u.slotKey, u.slotGen, u.gen
|
||||
u.allocSlots(2 * len(oldKey))
|
||||
u.gen = 1
|
||||
for i := range oldKey {
|
||||
if oldGen[i] == gen {
|
||||
j := u.slot(oldKey[i])
|
||||
u.slotKey[j], u.slotGen[j] = oldKey[i], u.gen
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (u *repeaterUnion) add(k uint64) {
|
||||
if u.width == 0 {
|
||||
u.width = int(k >> 32)
|
||||
}
|
||||
i := u.slot(k)
|
||||
if u.slotGen[i] == u.gen {
|
||||
return
|
||||
}
|
||||
if 2*(u.used+1) > len(u.slotKey) {
|
||||
u.grow()
|
||||
i = u.slot(k)
|
||||
}
|
||||
u.slotKey[i], u.slotGen[i] = k, u.gen
|
||||
u.used++
|
||||
}
|
||||
|
||||
// hopKey parses a hex hop into (width<<32 | value). ok is false for anything
|
||||
// that is not 1-4 bytes of hex.
|
||||
func hopKey(hop string) (uint64, bool) {
|
||||
if len(hop) == 0 || len(hop) > 8 || len(hop)%2 != 0 {
|
||||
return 0, false
|
||||
}
|
||||
var v uint64
|
||||
for i := 0; i < len(hop); i++ {
|
||||
c := hop[i]
|
||||
switch {
|
||||
case c >= '0' && c <= '9':
|
||||
c -= '0'
|
||||
case c >= 'a' && c <= 'f':
|
||||
c = c - 'a' + 10
|
||||
case c >= 'A' && c <= 'F':
|
||||
c = c - 'A' + 10
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
v = v<<4 | uint64(c)
|
||||
}
|
||||
return uint64(len(hop)/2)<<32 | v, true
|
||||
}
|
||||
|
||||
// addPath folds one observation's path_json (a JSON array of hex strings)
|
||||
// into the union. Scans the string directly: hop tokens are plain hex, so no
|
||||
// JSON decoder and no allocation are needed.
|
||||
func (u *repeaterUnion) addPath(pathJSON string) {
|
||||
for i := 0; i < len(pathJSON); i++ {
|
||||
if pathJSON[i] != '"' {
|
||||
continue
|
||||
}
|
||||
end := strings.IndexByte(pathJSON[i+1:], '"')
|
||||
if end < 0 {
|
||||
break
|
||||
}
|
||||
if k, ok := hopKey(pathJSON[i+1 : i+1+end]); ok {
|
||||
u.add(k)
|
||||
}
|
||||
i += end + 1
|
||||
}
|
||||
}
|
||||
|
||||
func (u *repeaterUnion) count() int {
|
||||
return u.used
|
||||
}
|
||||
|
||||
type retransmissionBucketAgg struct {
|
||||
packets int
|
||||
repeaterSum int
|
||||
observers []uint64 // bitset over the pass-local observer index
|
||||
}
|
||||
|
||||
func setBit(set []uint64, i int) []uint64 {
|
||||
for len(set) <= i/64 {
|
||||
set = append(set, 0)
|
||||
}
|
||||
set[i/64] |= 1 << uint(i%64)
|
||||
return set
|
||||
}
|
||||
|
||||
func popCount(set []uint64) int {
|
||||
n := 0
|
||||
for _, w := range set {
|
||||
n += bits.OnesCount64(w)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// timedObs is an observation with its parsed time, for sorting one
|
||||
// transmission's observations into flood events.
|
||||
type timedObs struct {
|
||||
at int64 // unix nanoseconds
|
||||
obs *StoreObs
|
||||
}
|
||||
|
||||
// retransmissionPass accumulates one computeRetransmissionPressure pass.
|
||||
type retransmissionPass struct {
|
||||
regionObs map[string]bool
|
||||
floor time.Time // events starting before it are dropped; zero = none
|
||||
since, until time.Time
|
||||
bucketSec int64
|
||||
aggs map[int64]*retransmissionBucketAgg
|
||||
obsIndex map[string]int
|
||||
allObservers []uint64
|
||||
union repeaterUnion
|
||||
summary RetransmissionSummary
|
||||
totalRepeaters int
|
||||
}
|
||||
|
||||
// addEvent counts one flood event: ev holds its observations in time order.
|
||||
func (p *retransmissionPass) addEvent(ev []timedObs) {
|
||||
t := time.Unix(0, ev[0].at)
|
||||
if (!p.floor.IsZero() && t.Before(p.floor)) ||
|
||||
(!p.since.IsZero() && t.Before(p.since)) || (!p.until.IsZero() && t.After(p.until)) {
|
||||
return
|
||||
}
|
||||
start := t.Unix() - t.Unix()%p.bucketSec
|
||||
agg := p.aggs[start]
|
||||
u := &p.union
|
||||
u.reset()
|
||||
heard := false
|
||||
for _, e := range ev {
|
||||
obs := e.obs
|
||||
if p.regionObs != nil && !p.regionObs[obs.ObserverID] {
|
||||
continue
|
||||
}
|
||||
if agg == nil {
|
||||
agg = &retransmissionBucketAgg{}
|
||||
p.aggs[start] = agg
|
||||
}
|
||||
heard = true
|
||||
u.addPath(obs.PathJSON)
|
||||
idx, ok := p.obsIndex[obs.ObserverID]
|
||||
if !ok {
|
||||
idx = len(p.obsIndex)
|
||||
p.obsIndex[obs.ObserverID] = idx
|
||||
}
|
||||
agg.observers = setBit(agg.observers, idx)
|
||||
p.allObservers = setBit(p.allObservers, idx)
|
||||
}
|
||||
if !heard {
|
||||
return
|
||||
}
|
||||
n := u.count()
|
||||
agg.packets++
|
||||
agg.repeaterSum += n
|
||||
p.summary.Packets++
|
||||
p.totalRepeaters += n
|
||||
if n == 0 {
|
||||
p.summary.NoRepeaterPackets++
|
||||
}
|
||||
if u.width == 1 {
|
||||
p.summary.OneBytePackets++
|
||||
}
|
||||
}
|
||||
|
||||
// computeRetransmissionPressure builds the series (see the file header for
|
||||
// flood events, the retention floor, the counting rule and region).
|
||||
func (s *PacketStore) computeRetransmissionPressure(region string, window TimeWindow, bucket time.Duration) RetransmissionResponse {
|
||||
if bucket <= 0 {
|
||||
bucket = retransmissionDefaultBucket
|
||||
}
|
||||
p := retransmissionPass{
|
||||
bucketSec: int64(bucket / time.Second),
|
||||
aggs: make(map[int64]*retransmissionBucketAgg),
|
||||
obsIndex: make(map[string]int),
|
||||
}
|
||||
if region != "" {
|
||||
p.regionObs = s.resolveRegionObservers(region)
|
||||
}
|
||||
if window.Since != "" {
|
||||
p.since, _ = parseAnyRFC3339(window.Since)
|
||||
}
|
||||
if window.Until != "" {
|
||||
p.until, _ = parseAnyRFC3339(window.Until)
|
||||
}
|
||||
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
if s.retentionHours > 0 {
|
||||
p.floor = time.Now().Add(-time.Duration(s.retentionHours * float64(time.Hour)))
|
||||
}
|
||||
gap := int64(retransmissionEventGap)
|
||||
var timed []timedObs
|
||||
|
||||
for _, tx := range s.packets {
|
||||
if tx.RouteType == nil || (*tx.RouteType != RouteFlood && *tx.RouteType != RouteTransportFlood) {
|
||||
continue
|
||||
}
|
||||
if tx.PayloadType != nil && *tx.PayloadType == PayloadTRACE {
|
||||
continue
|
||||
}
|
||||
timed = timed[:0]
|
||||
for _, obs := range tx.Observations {
|
||||
if at, ok := obs.ParsedTime(); ok {
|
||||
timed = append(timed, timedObs{at: at.UnixNano(), obs: obs})
|
||||
}
|
||||
}
|
||||
slices.SortFunc(timed, func(a, b timedObs) int { return cmp.Compare(a.at, b.at) })
|
||||
for start := 0; start < len(timed); {
|
||||
end := start + 1
|
||||
for end < len(timed) && timed[end].at-timed[end-1].at <= gap {
|
||||
end++
|
||||
}
|
||||
p.addEvent(timed[start:end])
|
||||
start = end
|
||||
}
|
||||
}
|
||||
|
||||
starts := make([]int64, 0, len(p.aggs))
|
||||
for st := range p.aggs {
|
||||
starts = append(starts, st)
|
||||
}
|
||||
sort.Slice(starts, func(i, j int) bool { return starts[i] < starts[j] })
|
||||
buckets := make([]RetransmissionBucket, 0, len(starts))
|
||||
for _, st := range starts {
|
||||
a := p.aggs[st]
|
||||
buckets = append(buckets, RetransmissionBucket{
|
||||
Start: time.Unix(st, 0).UTC().Format(time.RFC3339),
|
||||
Packets: a.packets,
|
||||
RepeaterSum: a.repeaterSum,
|
||||
AvgRepeaters: float64(a.repeaterSum) / float64(a.packets),
|
||||
Observers: popCount(a.observers),
|
||||
})
|
||||
}
|
||||
summary := p.summary
|
||||
if summary.Packets > 0 {
|
||||
summary.AvgRepeaters = float64(p.totalRepeaters) / float64(summary.Packets)
|
||||
}
|
||||
summary.Observers = popCount(p.allObservers)
|
||||
|
||||
label := window.Label
|
||||
if label == "" && !window.IsZero() {
|
||||
label = window.Since + "/" + window.Until
|
||||
}
|
||||
return RetransmissionResponse{
|
||||
BucketSeconds: int(p.bucketSec),
|
||||
Window: label,
|
||||
Region: region,
|
||||
Summary: summary,
|
||||
Buckets: buckets,
|
||||
}
|
||||
}
|
||||
|
||||
func isDefaultRetransmissionShape(region string, window TimeWindow, bucket time.Duration) bool {
|
||||
return region == "" && window.IsZero() && bucket == retransmissionDefaultBucket
|
||||
}
|
||||
|
||||
// retransCacheGet returns a fresh cached result for key. Caller must hold
|
||||
// s.cacheMu.
|
||||
func (s *PacketStore) retransCacheGet(key string) (RetransmissionResponse, bool) {
|
||||
if e, ok := s.retransCache[key]; ok && time.Now().Before(e.expiresAt) {
|
||||
return e.data, true
|
||||
}
|
||||
return RetransmissionResponse{}, false
|
||||
}
|
||||
|
||||
// GetRetransmissionPressure serves the default shape from the recomputer
|
||||
// snapshot and every other shape from the TTL cache (compute on miss,
|
||||
// concurrent misses on one key share the compute).
|
||||
func (s *PacketStore) GetRetransmissionPressure(region string, window TimeWindow, bucket time.Duration) RetransmissionResponse {
|
||||
if isDefaultRetransmissionShape(region, window, bucket) {
|
||||
s.analyticsRecomputerMu.RLock()
|
||||
rc := s.recompRetransmissions
|
||||
s.analyticsRecomputerMu.RUnlock()
|
||||
if rc != nil {
|
||||
if r, ok := rc.Load().(RetransmissionResponse); ok {
|
||||
s.cacheMu.Lock()
|
||||
s.cacheHits++
|
||||
s.cacheMu.Unlock()
|
||||
return r
|
||||
}
|
||||
}
|
||||
}
|
||||
key := region + "|" + window.CacheKey() + "|" + bucket.String()
|
||||
s.cacheMu.Lock()
|
||||
if r, ok := s.retransCacheGet(key); ok {
|
||||
s.cacheHits++
|
||||
s.cacheMu.Unlock()
|
||||
return r
|
||||
}
|
||||
s.cacheMisses++
|
||||
s.cacheMu.Unlock()
|
||||
|
||||
v, _, _ := s.retransSF.Do(key, func() (interface{}, error) {
|
||||
// A caller that joins right after a winner stored its result must
|
||||
// not start a second pass.
|
||||
s.cacheMu.Lock()
|
||||
r, ok := s.retransCacheGet(key)
|
||||
s.cacheMu.Unlock()
|
||||
if ok {
|
||||
return r, nil
|
||||
}
|
||||
result := s.computeRetransmissionPressure(region, window, bucket)
|
||||
s.cacheMu.Lock()
|
||||
if s.retransCache == nil || len(s.retransCache) >= retransmissionCacheMax {
|
||||
s.retransCache = make(map[string]*retransmissionCacheEntry)
|
||||
}
|
||||
s.retransCache[key] = &retransmissionCacheEntry{data: result, expiresAt: time.Now().Add(s.rfCacheTTL)}
|
||||
s.cacheMu.Unlock()
|
||||
return result, nil
|
||||
})
|
||||
return v.(RetransmissionResponse)
|
||||
}
|
||||
|
||||
func (s *Server) handleAnalyticsRetransmissions(w http.ResponseWriter, r *http.Request) {
|
||||
region := r.URL.Query().Get("region")
|
||||
window := ParseTimeWindow(r)
|
||||
bucket := parseRetransmissionBucket(r.URL.Query().Get("bucket"))
|
||||
if s.store == nil {
|
||||
writeJSON(w, RetransmissionResponse{BucketSeconds: int(bucket / time.Second), Buckets: []RetransmissionBucket{}})
|
||||
return
|
||||
}
|
||||
// #1659 warmup gate (see handleAnalyticsRF for rationale).
|
||||
if isDefaultRetransmissionShape(region, window, bucket) {
|
||||
s.store.analyticsRecomputerMu.RLock()
|
||||
rc := s.store.recompRetransmissions
|
||||
s.store.analyticsRecomputerMu.RUnlock()
|
||||
if rc != nil && rc.IsWarmingUp_1659() {
|
||||
writeAnalyticsWarmup503(w)
|
||||
return
|
||||
}
|
||||
}
|
||||
writeJSON(w, s.store.GetRetransmissionPressure(region, window, bucket))
|
||||
}
|
||||
@@ -0,0 +1,685 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// --- repeaterUnion: the per-transmission count (issue #1699) ---
|
||||
|
||||
func unionOf(paths ...string) int {
|
||||
var u repeaterUnion
|
||||
u.reset()
|
||||
for _, p := range paths {
|
||||
u.addPath(p)
|
||||
}
|
||||
return u.count()
|
||||
}
|
||||
|
||||
// The reporter's worked example from the #1699 thread: observer A saw
|
||||
// [A] and [A,B,C], observer B saw [A,D]. Four repeaters took part.
|
||||
func TestRepeaterUnion_ReporterExample(t *testing.T) {
|
||||
got := unionOf(`["A1"]`, `["A1","B2","C3"]`, `["A1","D4"]`)
|
||||
if got != 4 {
|
||||
t.Fatalf("union = %d, want 4", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepeaterUnion_OverlappingPathsCountOnce(t *testing.T) {
|
||||
got := unionOf(`["FD35","95F8","3363"]`, `["FD35","95F8","3363"]`, `["FD35","95F8"]`, `["FD35","95F8","4F47"]`)
|
||||
if got != 4 {
|
||||
t.Fatalf("union = %d, want 4 (FD35, 95F8, 3363, 4F47)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A hop prefix counts once per flood, wherever and however often it appears.
|
||||
// Inside one path a repeated 2- or 3-byte prefix is on live data almost
|
||||
// always one known node forwarding twice (the firmware dedup table is a
|
||||
// cyclic 160-slot buffer, SimpleMeshTables.h:9,52-57), so it is one
|
||||
// repeater. A repeated 1-byte prefix may be two nodes, but counting it once
|
||||
// keeps the value a lower bound, like the cross-observation merge.
|
||||
func TestRepeaterUnion_PrefixCountsOncePerFlood(t *testing.T) {
|
||||
if got := unionOf(`["12","34","12"]`); got != 2 {
|
||||
t.Errorf("1-byte prefix twice in one path = %d, want 2", got)
|
||||
}
|
||||
if got := unionOf(`["AB01","CD02","AB01"]`); got != 2 {
|
||||
t.Errorf("2-byte prefix twice in one path = %d, want 2", got)
|
||||
}
|
||||
if got := unionOf(`["AB01C3","CD02D4","AB01C3"]`); got != 2 {
|
||||
t.Errorf("3-byte prefix twice in one path = %d, want 2", got)
|
||||
}
|
||||
if got := unionOf(`["12","34"]`, `["56","12"]`); got != 3 {
|
||||
t.Errorf("same prefix in two paths = %d, want 3 (merged, lower bound)", got)
|
||||
}
|
||||
if got := unionOf(`["12","34","12"]`, `["12"]`, `["12","12","12"]`); got != 2 {
|
||||
t.Errorf("repeats within and across paths = %d, want 2 (12, 34)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepeaterUnion_HashSizeAndCaseAreDistinctKeys(t *testing.T) {
|
||||
if got := unionOf(`["ab"]`, `["AB"]`); got != 1 {
|
||||
t.Errorf("case variants = %d, want 1", got)
|
||||
}
|
||||
// A 1-byte "AB" and a 2-byte "AB00" are different hash widths.
|
||||
if got := unionOf(`["AB"]`, `["AB00"]`); got != 2 {
|
||||
t.Errorf("1-byte vs 2-byte = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepeaterUnion_EmptyAndMalformed(t *testing.T) {
|
||||
if got := unionOf("", `[]`); got != 0 {
|
||||
t.Errorf("empty paths = %d, want 0", got)
|
||||
}
|
||||
if got := unionOf(`["ZZ","A1"]`); got != 1 {
|
||||
t.Errorf("non-hex hop must be ignored, got %d want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepeaterUnion_ResetClearsState(t *testing.T) {
|
||||
var u repeaterUnion
|
||||
u.reset()
|
||||
u.addPath(`["A1","B2"]`)
|
||||
u.reset()
|
||||
u.addPath(`["C3"]`)
|
||||
if got := u.count(); got != 1 {
|
||||
t.Fatalf("after reset count = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRepeaterUnion_GrowsPastInitialTable(t *testing.T) {
|
||||
var u repeaterUnion
|
||||
u.reset()
|
||||
const distinct = 1000 // well past repeaterUnionMinSlots/2
|
||||
for i := 0; i < distinct; i += 4 {
|
||||
u.addPath(fmt.Sprintf(`["%04X","%04X","%04X","%04X"]`, i, i+1, i+2, i+3))
|
||||
}
|
||||
u.addPath(`["0000","0001"]`) // already present, must not add
|
||||
if got := u.count(); got != distinct {
|
||||
t.Fatalf("count after growth = %d, want %d", got, distinct)
|
||||
}
|
||||
u.reset()
|
||||
u.addPath(`["0000","ABCD"]`)
|
||||
if got := u.count(); got != 2 {
|
||||
t.Fatalf("count after reset on grown table = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
// --- computeRetransmissionPressure ---
|
||||
|
||||
func rtxInt(v int) *int { return &v }
|
||||
|
||||
func rtxTx(id int, route, payload int, firstSeen string, obs ...*StoreObs) *StoreTx {
|
||||
tx := &StoreTx{
|
||||
ID: id,
|
||||
Hash: fmt.Sprintf("h%06d", id),
|
||||
FirstSeen: firstSeen,
|
||||
RouteType: rtxInt(route),
|
||||
PayloadType: rtxInt(payload),
|
||||
}
|
||||
for _, o := range obs {
|
||||
o.TransmissionID = id
|
||||
if o.Timestamp == "" {
|
||||
o.Timestamp = firstSeen
|
||||
}
|
||||
tx.Observations = append(tx.Observations, o)
|
||||
}
|
||||
return tx
|
||||
}
|
||||
|
||||
// rtxObs is an observation heard at the transmission's first_seen.
|
||||
func rtxObs(observer, path string) *StoreObs {
|
||||
return &StoreObs{ObserverID: observer, PathJSON: path}
|
||||
}
|
||||
|
||||
func rtxObsAt(observer, path, ts string) *StoreObs {
|
||||
return &StoreObs{ObserverID: observer, PathJSON: path, Timestamp: ts}
|
||||
}
|
||||
|
||||
func rtxStore(packets ...*StoreTx) *PacketStore {
|
||||
return &PacketStore{packets: packets}
|
||||
}
|
||||
|
||||
func bucketByStart(t *testing.T, r RetransmissionResponse, start string) RetransmissionBucket {
|
||||
t.Helper()
|
||||
for _, b := range r.Buckets {
|
||||
if b.Start == start {
|
||||
return b
|
||||
}
|
||||
}
|
||||
t.Fatalf("no bucket %s in %+v", start, r.Buckets)
|
||||
return RetransmissionBucket{}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_RouteAndPayloadFilter(t *testing.T) {
|
||||
ts := "2026-09-13T10:10:00Z"
|
||||
s := rtxStore(
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, ts, rtxObs("o1", `["A1","B2"]`), rtxObs("o2", `["A1","C3"]`)),
|
||||
rtxTx(2, RouteTransportFlood, PayloadGRP_TXT, ts, rtxObs("o1", `["D4"]`)),
|
||||
// Flood heard only straight from the originator: a packet with 0
|
||||
// observed repeaters, kept in the denominator.
|
||||
rtxTx(3, RouteFlood, PayloadGRP_TXT, ts, rtxObs("o1", `[]`)),
|
||||
// Direct routes carry the remaining route, not the forwarders.
|
||||
rtxTx(4, RouteDirect, PayloadTXT_MSG, ts, rtxObs("o1", `["E5","F6"]`)),
|
||||
rtxTx(5, RouteTransportDirect, PayloadTXT_MSG, ts, rtxObs("o1", `["E5"]`)),
|
||||
// TRACE path bytes are SNR values, never forwarder hashes.
|
||||
rtxTx(6, RouteFlood, PayloadTRACE, ts, rtxObs("o1", `["E5","F6"]`)),
|
||||
// Missing route type cannot be classified.
|
||||
&StoreTx{ID: 7, FirstSeen: ts, Observations: []*StoreObs{rtxObs("o1", `["E5"]`)}},
|
||||
)
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
if r.BucketSeconds != 3600 {
|
||||
t.Errorf("bucket_seconds = %d, want 3600", r.BucketSeconds)
|
||||
}
|
||||
if r.Summary.Packets != 3 {
|
||||
t.Fatalf("summary packets = %d, want 3 (tx 1,2,3)", r.Summary.Packets)
|
||||
}
|
||||
b := bucketByStart(t, r, "2026-09-13T10:00:00Z")
|
||||
if b.Packets != 3 || b.RepeaterSum != 4 {
|
||||
t.Fatalf("bucket = %+v, want packets 3 repeater_sum 4 (3+1+0)", b)
|
||||
}
|
||||
if want := 4.0 / 3.0; b.AvgRepeaters < want-1e-9 || b.AvgRepeaters > want+1e-9 {
|
||||
t.Errorf("avg_repeaters = %v, want %v", b.AvgRepeaters, want)
|
||||
}
|
||||
if b.Observers != 2 {
|
||||
t.Errorf("observers = %d, want 2", b.Observers)
|
||||
}
|
||||
if r.Summary.NoRepeaterPackets != 1 {
|
||||
t.Errorf("no_repeater_packets = %d, want 1", r.Summary.NoRepeaterPackets)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_Bucketing(t *testing.T) {
|
||||
s := rtxStore(
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:05:00Z", rtxObs("o1", `["A1","B2"]`)),
|
||||
rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:55:59Z", rtxObs("o2", `["A1"]`)),
|
||||
rtxTx(3, RouteFlood, PayloadADVERT, "2026-09-13T11:00:00Z", rtxObs("o1", `["A1","B2","C3","D4"]`)),
|
||||
rtxTx(4, RouteFlood, PayloadADVERT, "not-a-time", rtxObs("o1", `["A1"]`)),
|
||||
)
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
if len(r.Buckets) != 2 {
|
||||
t.Fatalf("buckets = %+v, want 2", r.Buckets)
|
||||
}
|
||||
if r.Buckets[0].Start != "2026-09-13T10:00:00Z" || r.Buckets[1].Start != "2026-09-13T11:00:00Z" {
|
||||
t.Fatalf("bucket order/starts wrong: %+v", r.Buckets)
|
||||
}
|
||||
b10 := r.Buckets[0]
|
||||
if b10.Packets != 2 || b10.RepeaterSum != 3 || b10.AvgRepeaters != 1.5 || b10.Observers != 2 {
|
||||
t.Errorf("10:00 bucket = %+v, want packets 2 sum 3 avg 1.5 observers 2", b10)
|
||||
}
|
||||
b11 := r.Buckets[1]
|
||||
if b11.Packets != 1 || b11.RepeaterSum != 4 || b11.Observers != 1 {
|
||||
t.Errorf("11:00 bucket = %+v", b11)
|
||||
}
|
||||
if r.Summary.Packets != 3 || r.Summary.Observers != 2 {
|
||||
t.Errorf("summary = %+v, want packets 3 observers 2", r.Summary)
|
||||
}
|
||||
if want := 7.0 / 3.0; r.Summary.AvgRepeaters < want-1e-9 || r.Summary.AvgRepeaters > want+1e-9 {
|
||||
t.Errorf("summary avg = %v, want %v", r.Summary.AvgRepeaters, want)
|
||||
}
|
||||
|
||||
r15 := s.computeRetransmissionPressure("", TimeWindow{}, 15*time.Minute)
|
||||
if len(r15.Buckets) != 3 {
|
||||
t.Fatalf("15m buckets = %+v, want 3", r15.Buckets)
|
||||
}
|
||||
bucketByStart(t, r15, "2026-09-13T10:00:00Z")
|
||||
bucketByStart(t, r15, "2026-09-13T10:45:00Z")
|
||||
bucketByStart(t, r15, "2026-09-13T11:00:00Z")
|
||||
if r15.BucketSeconds != 900 {
|
||||
t.Errorf("bucket_seconds = %d, want 900", r15.BucketSeconds)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_WindowFilter(t *testing.T) {
|
||||
s := rtxStore(
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-12T10:00:00Z", rtxObs("o1", `["A1"]`)),
|
||||
rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1","B2"]`)),
|
||||
rtxTx(3, RouteFlood, PayloadADVERT, "2026-09-14T10:00:00Z", rtxObs("o1", `["A1","B2","C3"]`)),
|
||||
)
|
||||
w := TimeWindow{Since: "2026-09-13T00:00:00Z", Until: "2026-09-13T23:59:59Z"}
|
||||
r := s.computeRetransmissionPressure("", w, time.Hour)
|
||||
if r.Summary.Packets != 1 || len(r.Buckets) != 1 || r.Buckets[0].RepeaterSum != 2 {
|
||||
t.Fatalf("window result = %+v, want only tx 2", r)
|
||||
}
|
||||
}
|
||||
|
||||
// transmissions.hash is UNIQUE, so a re-hearing of the same content hours
|
||||
// later is appended to the existing transmission. Each flood event is
|
||||
// counted on its own, in the bucket of its first observation. Observations
|
||||
// are listed newest first, the order the cold load appends them in.
|
||||
func TestComputeRetransmissionPressure_SplitsFloodEvents(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadGRP_TXT, "2026-09-13T10:00:00Z",
|
||||
rtxObsAt("o3", `["D4"]`, "2026-09-13T12:00:00.000Z"),
|
||||
rtxObsAt("o2", `["A1","C3"]`, "2026-09-13T10:00:20.000Z"),
|
||||
rtxObsAt("o1", `["A1","B2"]`, "2026-09-13T10:00:00.000Z"),
|
||||
))
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
if r.Summary.Packets != 2 || len(r.Buckets) != 2 {
|
||||
t.Fatalf("result = %+v, want 2 flood events in 2 buckets", r)
|
||||
}
|
||||
if b := bucketByStart(t, r, "2026-09-13T10:00:00Z"); b.Packets != 1 || b.RepeaterSum != 3 || b.Observers != 2 {
|
||||
t.Errorf("10:00 bucket = %+v, want packets 1 sum 3 (A1,B2,C3) observers 2", b)
|
||||
}
|
||||
if b := bucketByStart(t, r, "2026-09-13T12:00:00Z"); b.Packets != 1 || b.RepeaterSum != 1 || b.Observers != 1 {
|
||||
t.Errorf("12:00 bucket = %+v, want packets 1 sum 1 (D4) observers 1", b)
|
||||
}
|
||||
if r.Summary.AvgRepeaters != 2 {
|
||||
t.Errorf("summary avg = %v, want 2", r.Summary.AvgRepeaters)
|
||||
}
|
||||
}
|
||||
|
||||
// An event ends when the gap to the previous observation exceeds the settle
|
||||
// time (5 minutes). The gap is measured observation to observation, so a
|
||||
// slow flood whose steps each stay under it remains one event.
|
||||
func TestComputeRetransmissionPressure_EventSettleGap(t *testing.T) {
|
||||
events := func(ts ...string) int {
|
||||
var obs []*StoreObs
|
||||
for i, x := range ts {
|
||||
obs = append(obs, rtxObsAt(fmt.Sprintf("o%d", i), `["A1"]`, x))
|
||||
}
|
||||
return rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, ts[0], obs...)).
|
||||
computeRetransmissionPressure("", TimeWindow{}, time.Hour).Summary.Packets
|
||||
}
|
||||
if got := events("2026-09-13T10:00:00Z", "2026-09-13T10:05:00Z"); got != 1 {
|
||||
t.Errorf("gap of exactly 5m = %d events, want 1", got)
|
||||
}
|
||||
if got := events("2026-09-13T10:00:00Z", "2026-09-13T10:05:01Z"); got != 2 {
|
||||
t.Errorf("gap of 5m01s = %d events, want 2", got)
|
||||
}
|
||||
if got := events("2026-09-13T10:00:00Z", "2026-09-13T10:04:00Z", "2026-09-13T10:08:00Z", "2026-09-13T10:12:00Z"); got != 1 {
|
||||
t.Errorf("4m steps over 12m = %d events, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
// The window selects flood events by their first observation, not by the
|
||||
// transmission's first_seen: a hash first heard weeks ago and flooded again
|
||||
// today counts today.
|
||||
func TestComputeRetransmissionPressure_WindowSelectsEvents(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-08-20T10:00:00Z",
|
||||
rtxObsAt("o1", `["A1","B2"]`, "2026-08-20T10:00:00Z"),
|
||||
rtxObsAt("o1", `["C3"]`, "2026-09-13T10:00:00Z"),
|
||||
))
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{Since: "2026-09-13T00:00:00Z"}, time.Hour)
|
||||
if r.Summary.Packets != 1 || len(r.Buckets) != 1 || r.Buckets[0].RepeaterSum != 1 {
|
||||
t.Fatalf("result = %+v, want only the 2026-09-13 event (C3)", r)
|
||||
}
|
||||
}
|
||||
|
||||
// Flood events that start before the store's retention floor are dropped,
|
||||
// for every request shape: the store keeps old observations only for hashes
|
||||
// that were heard again recently, so what it holds before the floor is not
|
||||
// the traffic of that period.
|
||||
func TestComputeRetransmissionPressure_RetentionFloor(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
ago := func(d time.Duration) string { return now.Add(-d).Format("2006-01-02T15:04:05.000Z") }
|
||||
s := rtxStore(
|
||||
// Weeks-old first event, recent second event: only the second counts.
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, ago(30*24*time.Hour),
|
||||
rtxObsAt("o1", `["A1","B2"]`, ago(30*24*time.Hour)),
|
||||
rtxObsAt("o1", `["C3"]`, ago(time.Hour))),
|
||||
// An event that starts before the floor is dropped as a whole, not
|
||||
// counted from its first observation after the floor.
|
||||
rtxTx(2, RouteFlood, PayloadADVERT, ago(24*time.Hour+time.Minute),
|
||||
rtxObsAt("o1", `["D4"]`, ago(24*time.Hour+time.Minute)),
|
||||
rtxObsAt("o2", `["E5"]`, ago(24*time.Hour-time.Minute))),
|
||||
rtxTx(3, RouteFlood, PayloadADVERT, ago(2*time.Hour), rtxObsAt("o1", `["F6","A7"]`, ago(2*time.Hour))),
|
||||
)
|
||||
s.retentionHours = 24
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
if r.Summary.Packets != 2 || r.Summary.AvgRepeaters != 1.5 {
|
||||
t.Fatalf("summary = %+v, want 2 events (C3; F6,A7) avg 1.5", r.Summary)
|
||||
}
|
||||
if got := s.computeRetransmissionPressure("", TimeWindow{Since: ago(40 * 24 * time.Hour)}, time.Hour).Summary.Packets; got != 2 {
|
||||
t.Errorf("window reaching past the floor = %d events, want 2", got)
|
||||
}
|
||||
s.retentionHours = 0
|
||||
if got := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour).Summary.Packets; got != 4 {
|
||||
t.Errorf("unlimited retention = %d events, want 4", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_RegionFilter(t *testing.T) {
|
||||
s := rtxStore(
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z",
|
||||
rtxObs("brussels", `["A1","B2"]`), rtxObs("amsterdam", `["A1","C3","D4"]`)),
|
||||
rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:10:00Z",
|
||||
rtxObs("amsterdam", `["E5"]`)),
|
||||
)
|
||||
s.regionObsCache = map[string]map[string]bool{"BRU": {"brussels": true}}
|
||||
s.regionObsCacheTime = time.Now()
|
||||
|
||||
r := s.computeRetransmissionPressure("BRU", TimeWindow{}, time.Hour)
|
||||
if r.Region != "BRU" {
|
||||
t.Errorf("region = %q, want BRU", r.Region)
|
||||
}
|
||||
if r.Summary.Packets != 1 {
|
||||
t.Fatalf("packets = %d, want 1 (tx 2 has no BRU observation)", r.Summary.Packets)
|
||||
}
|
||||
if got := r.Buckets[0].RepeaterSum; got != 2 {
|
||||
t.Errorf("repeater_sum = %d, want 2 (only the brussels path counts)", got)
|
||||
}
|
||||
if got := r.Buckets[0].Observers; got != 1 {
|
||||
t.Errorf("observers = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Events are split on all observations before the region filter, so a flood
|
||||
// the region heard at its start and end stays one event even when the
|
||||
// region's own observations are further apart than the settle time.
|
||||
func TestComputeRetransmissionPressure_RegionDoesNotSplitEvents(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z",
|
||||
rtxObsAt("brussels", `["A1"]`, "2026-09-13T10:00:00Z"),
|
||||
rtxObsAt("amsterdam", `["B2"]`, "2026-09-13T10:04:00Z"),
|
||||
rtxObsAt("brussels", `["A1","C3"]`, "2026-09-13T10:08:00Z"),
|
||||
))
|
||||
s.regionObsCache = map[string]map[string]bool{"BRU": {"brussels": true}}
|
||||
s.regionObsCacheTime = time.Now()
|
||||
|
||||
r := s.computeRetransmissionPressure("BRU", TimeWindow{}, time.Hour)
|
||||
if r.Summary.Packets != 1 || r.Buckets[0].RepeaterSum != 2 {
|
||||
t.Fatalf("region result = %+v, want 1 event with A1, C3", r)
|
||||
}
|
||||
}
|
||||
|
||||
// A region with no known observers is not filtered, the same as
|
||||
// /api/analytics/rf and the other analytics endpoints (resolveRegionObservers
|
||||
// returns nil). Pinned so a change is a deliberate, documented one.
|
||||
func TestComputeRetransmissionPressure_UnknownRegionIsNotFiltered(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z",
|
||||
rtxObs("brussels", `["A1","B2"]`), rtxObs("amsterdam", `["C3"]`)))
|
||||
s.regionObsCache = map[string]map[string]bool{"XXX": nil}
|
||||
s.regionObsCacheTime = time.Now()
|
||||
|
||||
r := s.computeRetransmissionPressure("XXX", TimeWindow{}, time.Hour)
|
||||
if r.Summary.Packets != 1 || r.Summary.Observers != 2 || r.Buckets[0].RepeaterSum != 3 {
|
||||
t.Fatalf("unknown region = %+v, want the network-wide result", r)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_OneByteShare(t *testing.T) {
|
||||
ts := "2026-09-13T10:00:00Z"
|
||||
s := rtxStore(
|
||||
rtxTx(1, RouteFlood, PayloadADVERT, ts, rtxObs("o1", `["A1","B2"]`)),
|
||||
rtxTx(2, RouteFlood, PayloadADVERT, ts, rtxObs("o1", `["A1B2","C3D4"]`)),
|
||||
rtxTx(3, RouteFlood, PayloadADVERT, ts, rtxObs("o1", `[]`), rtxObs("o2", `["A1B2C3"]`)),
|
||||
rtxTx(4, RouteFlood, PayloadADVERT, ts, rtxObs("o1", `[]`)),
|
||||
)
|
||||
r := s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
if r.Summary.OneBytePackets != 1 {
|
||||
t.Errorf("one_byte_packets = %d, want 1", r.Summary.OneBytePackets)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComputeRetransmissionPressure_EmptyStoreHasNonNilBuckets(t *testing.T) {
|
||||
r := rtxStore().computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
b, _ := json.Marshal(r)
|
||||
if !strings.Contains(string(b), `"buckets":[]`) {
|
||||
t.Fatalf("empty result must encode buckets as [], got %s", b)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseRetransmissionBucket(t *testing.T) {
|
||||
cases := map[string]time.Duration{
|
||||
"": time.Hour,
|
||||
"5m": 5 * time.Minute,
|
||||
"15m": 15 * time.Minute,
|
||||
"1h": time.Hour,
|
||||
"6h": 6 * time.Hour,
|
||||
"1d": 24 * time.Hour,
|
||||
"7m": time.Hour,
|
||||
"abc": time.Hour,
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := parseRetransmissionBucket(in); got != want {
|
||||
t.Errorf("parseRetransmissionBucket(%q) = %v, want %v", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --- caching ---
|
||||
|
||||
func TestGetRetransmissionPressure_DefaultShapeServedFromRecomputer(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1"]`)))
|
||||
sentinel := RetransmissionResponse{BucketSeconds: 3600, Buckets: []RetransmissionBucket{}, Window: "sentinel"}
|
||||
rc := newAnalyticsRecomputer("retransmissions", time.Hour, func() interface{} { return sentinel })
|
||||
rc.runOnce()
|
||||
s.recompRetransmissions = rc
|
||||
|
||||
if got := s.GetRetransmissionPressure("", TimeWindow{}, time.Hour); got.Window != "sentinel" {
|
||||
t.Fatalf("default shape must come from the recomputer snapshot, got %+v", got)
|
||||
}
|
||||
if got := s.GetRetransmissionPressure("", TimeWindow{}, 15*time.Minute); got.Window == "sentinel" {
|
||||
t.Fatalf("non-default bucket must not be served from the default snapshot")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetRetransmissionPressure_TTLCacheAndInvalidation(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1"]`)))
|
||||
s.rfCacheTTL = time.Hour
|
||||
w := TimeWindow{Since: "2026-09-13T00:00:00Z", Label: "fixture"}
|
||||
|
||||
first := s.GetRetransmissionPressure("", w, time.Hour)
|
||||
if first.Summary.Packets != 1 {
|
||||
t.Fatalf("first = %+v", first)
|
||||
}
|
||||
// Mutate the store behind the cache: a cached read must not see it.
|
||||
s.packets = append(s.packets, rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:30:00Z", rtxObs("o1", `["B2"]`)))
|
||||
if got := s.GetRetransmissionPressure("", w, time.Hour); got.Summary.Packets != 1 {
|
||||
t.Fatalf("expected cache hit with 1 packet, got %d", got.Summary.Packets)
|
||||
}
|
||||
s.applyCacheInvalidation(cacheInvalidation{hasNewPaths: true})
|
||||
if got := s.GetRetransmissionPressure("", w, time.Hour); got.Summary.Packets != 2 {
|
||||
t.Fatalf("after hasNewPaths invalidation expected 2 packets, got %d", got.Summary.Packets)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetRetransmissionPressure_EvictionClearsCache(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1"]`)))
|
||||
s.rfCacheTTL = time.Hour
|
||||
w := TimeWindow{Since: "2026-09-13T00:00:00Z", Label: "fixture"}
|
||||
|
||||
s.GetRetransmissionPressure("", w, time.Hour)
|
||||
s.packets = append(s.packets, rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:30:00Z", rtxObs("o1", `["B2"]`)))
|
||||
s.invalidateCachesFor(cacheInvalidation{eviction: true})
|
||||
if got := s.GetRetransmissionPressure("", w, time.Hour); got.Summary.Packets != 2 {
|
||||
t.Fatalf("after eviction invalidation expected 2 packets, got %d", got.Summary.Packets)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetRetransmissionPressure_CacheEntryExpires(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1"]`)))
|
||||
s.rfCacheTTL = time.Hour
|
||||
w := TimeWindow{Since: "2026-09-13T00:00:00Z", Label: "fixture"}
|
||||
|
||||
s.GetRetransmissionPressure("", w, time.Hour)
|
||||
s.packets = append(s.packets, rtxTx(2, RouteFlood, PayloadADVERT, "2026-09-13T10:30:00Z", rtxObs("o1", `["B2"]`)))
|
||||
if got := s.GetRetransmissionPressure("", w, time.Hour); got.Summary.Packets != 1 {
|
||||
t.Fatalf("fresh entry must be served from cache, got %d packets", got.Summary.Packets)
|
||||
}
|
||||
if len(s.retransCache) != 1 {
|
||||
t.Fatalf("cache entries = %d, want 1", len(s.retransCache))
|
||||
}
|
||||
for _, e := range s.retransCache {
|
||||
e.expiresAt = time.Now().Add(-time.Second)
|
||||
}
|
||||
if got := s.GetRetransmissionPressure("", w, time.Hour); got.Summary.Packets != 2 {
|
||||
t.Fatalf("expired entry must be recomputed, got %d packets", got.Summary.Packets)
|
||||
}
|
||||
}
|
||||
|
||||
// Concurrent requests for the same uncached shape share one compute: every
|
||||
// caller gets the result of that single pass (same bucket backing array).
|
||||
func TestGetRetransmissionPressure_CollapsesConcurrentMisses(t *testing.T) {
|
||||
s := rtxStore(rtxTx(1, RouteFlood, PayloadADVERT, "2026-09-13T10:00:00Z", rtxObs("o1", `["A1"]`)))
|
||||
s.rfCacheTTL = time.Hour
|
||||
w := TimeWindow{Since: "2026-09-13T00:00:00Z", Label: "fixture"}
|
||||
|
||||
const n = 16
|
||||
results := make([]RetransmissionResponse, n)
|
||||
var wg sync.WaitGroup
|
||||
s.mu.Lock() // park every compute on the store lock
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
results[i] = s.GetRetransmissionPressure("", w, time.Hour)
|
||||
}(i)
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
s.mu.Unlock()
|
||||
wg.Wait()
|
||||
for i, r := range results {
|
||||
if len(r.Buckets) != 1 {
|
||||
t.Fatalf("result %d = %+v", i, r)
|
||||
}
|
||||
if &r.Buckets[0] != &results[0].Buckets[0] {
|
||||
t.Fatalf("result %d came from a separate compute; concurrent misses must share one", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// StartAnalyticsRecomputers must wire the #1659 readiness gate on the
|
||||
// retransmissions recomputer: a pass before the cold load completes keeps
|
||||
// the default shape at 503.
|
||||
func TestStartAnalyticsRecomputers_RetransmissionsGatedOnLoadComplete(t *testing.T) {
|
||||
db := setupTestDB(t)
|
||||
defer db.Close()
|
||||
store := NewPacketStore(db, nil)
|
||||
stop := store.StartAnalyticsRecomputers(time.Hour)
|
||||
defer stop()
|
||||
|
||||
if !store.recompRetransmissions.IsWarmingUp_1659() {
|
||||
t.Fatal("a pass before LoadComplete must not open the retransmissions gate")
|
||||
}
|
||||
store.signalStartupLoadDone()
|
||||
store.recompRetransmissions.runOnce()
|
||||
if store.recompRetransmissions.IsWarmingUp_1659() {
|
||||
t.Fatal("a pass after LoadComplete must open the retransmissions gate")
|
||||
}
|
||||
}
|
||||
|
||||
// --- handler ---
|
||||
|
||||
func TestHandleAnalyticsRetransmissions(t *testing.T) {
|
||||
_, router := setupTestServer(t)
|
||||
req := httptest.NewRequest("GET", "/api/analytics/retransmissions?bucket=1d", nil)
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, body %s", w.Code, w.Body.String())
|
||||
}
|
||||
var body RetransmissionResponse
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("decode: %v (%s)", err, w.Body.String())
|
||||
}
|
||||
if body.BucketSeconds != 86400 {
|
||||
t.Errorf("bucket_seconds = %d, want 86400", body.BucketSeconds)
|
||||
}
|
||||
// Seed: tx1 flood ADVERT paths ["aa","bb"] + ["aa"] = 2 repeaters,
|
||||
// tx2 flood GRP_TXT path [] = 0, tx3 flood ADVERT ["cc"] = 1.
|
||||
if body.Summary.Packets != 3 {
|
||||
t.Fatalf("summary.packets = %d, want 3 (%s)", body.Summary.Packets, w.Body.String())
|
||||
}
|
||||
sum := 0
|
||||
for _, b := range body.Buckets {
|
||||
sum += b.RepeaterSum
|
||||
}
|
||||
if sum != 3 {
|
||||
t.Errorf("total repeater_sum = %d, want 3", sum)
|
||||
}
|
||||
for _, key := range []string{`"bucket_seconds"`, `"summary"`, `"avg_repeaters"`, `"observers"`, `"one_byte_packets"`, `"no_repeater_packets"`, `"buckets"`} {
|
||||
if !strings.Contains(w.Body.String(), key) {
|
||||
t.Errorf("response missing %s: %s", key, w.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAnalyticsRetransmissions_WarmupGate(t *testing.T) {
|
||||
srv, router := setupTestServer(t)
|
||||
rc := newAnalyticsRecomputer("retransmissions", time.Hour, func() interface{} { return nil })
|
||||
rc.noteWarmupStart_1659()
|
||||
rc.setWarmupReadyGate_1659(func() bool { return false })
|
||||
srv.store.recompRetransmissions = rc
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, httptest.NewRequest("GET", "/api/analytics/retransmissions", nil))
|
||||
if w.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("default shape during warmup: status = %d, want 503", w.Code)
|
||||
}
|
||||
w = httptest.NewRecorder()
|
||||
router.ServeHTTP(w, httptest.NewRequest("GET", "/api/analytics/retransmissions?window=24h", nil))
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("windowed shape bypasses the gate: status = %d, want 200", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// --- benchmark (perf proof for AGENTS.md rule 0) ---
|
||||
|
||||
// BenchmarkComputeRetransmissionPressure sizes the fixture on live
|
||||
// magnitudes (2026-09-13): ~15k flood transmissions/day, ~20 observations
|
||||
// each, ~5.3 hops per path. 50k tx x 20 obs = 1M observations, roughly
|
||||
// 3.3 days of flood traffic; a 14-day store scales linearly (x4.3).
|
||||
//
|
||||
// Observations carry timestamps in the store's format, newest first per
|
||||
// transmission, which is the order the cold load appends them in
|
||||
// (ORDER BY o.timestamp DESC).
|
||||
func retransmissionBenchStore() *PacketStore {
|
||||
const nTx, obsPerTx = 50000, 20
|
||||
base := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
|
||||
packets := make([]*StoreTx, 0, nTx)
|
||||
for i := 0; i < nTx; i++ {
|
||||
first := base.Add(time.Duration(i) * 6 * time.Second)
|
||||
obs := make([]*StoreObs, 0, obsPerTx)
|
||||
for j := 0; j < obsPerTx; j++ {
|
||||
hops := make([]string, 0, 6)
|
||||
for h := 0; h < 3+(i+j)%4; h++ {
|
||||
hops = append(hops, fmt.Sprintf("%04X", (i*7+h*131+j*(h+1))%4096))
|
||||
}
|
||||
pj, _ := json.Marshal(hops)
|
||||
ts := first.Add(time.Duration(obsPerTx-1-j) * time.Second).Format("2006-01-02T15:04:05.000Z")
|
||||
obs = append(obs, &StoreObs{ObserverID: fmt.Sprintf("obs%02d", j*3%60), PathJSON: string(pj), Timestamp: ts})
|
||||
}
|
||||
packets = append(packets, rtxTx(i+1, RouteFlood, PayloadADVERT, first.Format(time.RFC3339), obs...))
|
||||
}
|
||||
return rtxStore(packets...)
|
||||
}
|
||||
|
||||
// BenchmarkComputeRetransmissionPressure measures a recompute pass on a
|
||||
// store whose observation timestamps were already parsed (the steady state:
|
||||
// StoreObs.ParsedTime caches per observation).
|
||||
func BenchmarkComputeRetransmissionPressure(b *testing.B) {
|
||||
s := retransmissionBenchStore()
|
||||
s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
runtime.GC() // fixture garbage must not be collected inside the timed loop
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
}
|
||||
}
|
||||
|
||||
// BenchmarkComputeRetransmissionPressureColdTimestamps measures the first
|
||||
// pass after startup, when no observation timestamp has been parsed yet.
|
||||
func BenchmarkComputeRetransmissionPressureColdTimestamps(b *testing.B) {
|
||||
s := retransmissionBenchStore()
|
||||
runtime.GC()
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
b.StopTimer()
|
||||
for _, tx := range s.packets {
|
||||
for _, o := range tx.Observations {
|
||||
o.tsParseOnce = sync.Once{}
|
||||
o.tsParsed, o.tsParsedOK = time.Time{}, false
|
||||
}
|
||||
}
|
||||
b.StartTimer()
|
||||
s.computeRetransmissionPressure("", TimeWindow{}, time.Hour)
|
||||
}
|
||||
}
|
||||
@@ -329,6 +329,7 @@ func (s *Server) RegisterRoutes(r *mux.Router) {
|
||||
r.HandleFunc("/api/analytics/subpath-detail", s.handleAnalyticsSubpathDetail).Methods("GET")
|
||||
r.HandleFunc("/api/analytics/neighbor-graph", s.handleNeighborGraph).Methods("GET")
|
||||
r.HandleFunc("/api/analytics/relay-airtime-share", s.handleAnalyticsRelayAirtimeShare).Methods("GET")
|
||||
r.HandleFunc("/api/analytics/retransmissions", s.handleAnalyticsRetransmissions).Methods("GET")
|
||||
|
||||
// Other endpoints
|
||||
r.HandleFunc("/api/resolve-hops", s.handleResolveHops).Methods("GET")
|
||||
|
||||
@@ -197,6 +197,12 @@ type PacketStore struct {
|
||||
subpathCache map[string]*cachedResult // params → cached subpaths result
|
||||
rfCacheTTL time.Duration
|
||||
collisionCacheTTL time.Duration
|
||||
|
||||
// region|window|bucket → retransmission pressure (#1699). Typed, so it
|
||||
// cannot share the map[string]interface{} caches above; nil until first use.
|
||||
retransCache map[string]*retransmissionCacheEntry
|
||||
retransSF singleflight.Group // collapses concurrent misses on one retransCache key
|
||||
|
||||
// Steady-state analytics recomputers (issue #1240). Each holds the
|
||||
// latest snapshot for the default region="" / zero-window query of
|
||||
// an analytics endpoint in an atomic.Value, refreshed by a
|
||||
@@ -214,6 +220,7 @@ type PacketStore struct {
|
||||
recompRoles *analyticsRecomputer
|
||||
recompObserversClockSkew *analyticsRecomputer
|
||||
recompNodesClockSkew *analyticsRecomputer
|
||||
recompRetransmissions *analyticsRecomputer
|
||||
cacheHits int64
|
||||
cacheMisses int64
|
||||
// Rate-limited invalidation (fixes #533: caches cleared faster than hit)
|
||||
@@ -2284,6 +2291,7 @@ func (s *PacketStore) invalidateCachesFor(inv cacheInvalidation) {
|
||||
s.chanCache = make(map[string]*cachedResult)
|
||||
s.distCache = make(map[string]*cachedResult)
|
||||
s.subpathCache = make(map[string]*cachedResult)
|
||||
s.retransCache = nil
|
||||
s.channelsCacheMu.Lock()
|
||||
s.channelsCacheRes = nil
|
||||
s.channelsCacheMu.Unlock()
|
||||
@@ -2330,6 +2338,7 @@ func (s *PacketStore) applyCacheInvalidation(inv cacheInvalidation) {
|
||||
s.topoCache = make(map[string]*cachedResult)
|
||||
s.distCache = make(map[string]*cachedResult)
|
||||
s.subpathCache = make(map[string]*cachedResult)
|
||||
s.retransCache = nil
|
||||
}
|
||||
if inv.hasNewTransmissions {
|
||||
s.hashCache = make(map[string]*cachedResult)
|
||||
|
||||
@@ -36,6 +36,7 @@
|
||||
- [GET /api/channels/:hash/messages](#get-apichannelshashmessages)
|
||||
- [GET /api/analytics/rf](#get-apianalyticsrf)
|
||||
- [GET /api/analytics/topology](#get-apianalyticstopology)
|
||||
- [GET /api/analytics/retransmissions](#get-apianalyticsretransmissions)
|
||||
- [GET /api/analytics/channels](#get-apianalyticschannels)
|
||||
- [GET /api/analytics/distance](#get-apianalyticsdistance)
|
||||
- [GET /api/analytics/hash-sizes](#get-apianalyticshash-sizes)
|
||||
@@ -1356,6 +1357,93 @@ Network topology analytics.
|
||||
|
||||
---
|
||||
|
||||
## GET /api/analytics/retransmissions
|
||||
|
||||
Retransmission pressure over time (#1699): a collision-pressure **proxy**, not a
|
||||
measured collision rate.
|
||||
|
||||
For each flood event of a flood-routed packet (`route_type` 0 or 1, TRACE
|
||||
excluded) the server takes the union of the paths of all its observations and
|
||||
counts the distinct repeaters in it: paths `[A]`, `[A,B,C]` and `[A,D]` give 4.
|
||||
Direct routes are excluded because their path is the route still to travel, not
|
||||
the forwarders; zero-hop sends are direct routes. A flood event heard only with
|
||||
an empty path counts as 0 repeaters.
|
||||
|
||||
A transmission is one packet hash, and the same bytes can flood again later:
|
||||
those observations are stored on the same transmission. Its observations are
|
||||
therefore sorted by time and split into flood events wherever two consecutive
|
||||
observations are more than 5 minutes apart. Each event is counted on its own and
|
||||
bucketed by its first observation. A firmware node holds a flood for at most
|
||||
32 s before forwarding it, plus a random retransmit delay, so a flood that is
|
||||
still spreading is not split. Every count in the response (`packets`,
|
||||
`one_byte_packets`, `no_repeater_packets`) counts flood events. Observations are
|
||||
stored once per observer and path per transmission, so a later event holds only
|
||||
the observer and path pairs not already stored for that hash, and its count is a
|
||||
lower bound.
|
||||
|
||||
Events that start before the store's retention floor (now minus
|
||||
`retentionHours`) are left out for every request shape, including explicit
|
||||
`window`, `from` and `to`: the store keeps observations older than that only for
|
||||
hashes heard again recently, so they do not represent that period.
|
||||
|
||||
Hop prefixes are not resolved to nodes. A prefix counts once per flood event,
|
||||
whether it repeats across observations or inside one path. A 2- or 3-byte prefix
|
||||
that repeats inside one path is one node forwarding the flood again after its
|
||||
duplicate filter (a cyclic buffer of 160 hashes) dropped the hash. A repeated
|
||||
1-byte prefix can also be two nodes; counting it once keeps the value a lower
|
||||
bound, as does merging repeaters that share a prefix across observations. Only
|
||||
repeaters that some observer heard are counted, so the value also follows
|
||||
observer coverage.
|
||||
|
||||
The default shape (no `region`, no window, `bucket=1h`) is served from the
|
||||
analytics recomputer; other shapes use the TTL cache, and concurrent requests
|
||||
for the same uncached shape share one computation. During startup the default
|
||||
shape returns `503` with `Retry-After` until the recomputer completes a pass
|
||||
after the hot startup window has loaded, or for at most 60 s after the
|
||||
recomputer started, whichever comes first. The history beyond the hot window
|
||||
keeps loading in the background after that, so until the first recompute pass
|
||||
after that load finishes the default shape can cover less than the retention
|
||||
window. `?area=` is not
|
||||
supported: the area filter works on resolved node public keys and this metric
|
||||
does not resolve prefixes.
|
||||
|
||||
### Query Parameters
|
||||
|
||||
| Param | Type | Default | Description |
|
||||
|----------|--------|---------|-------------------------------------|
|
||||
| `region` | string | none | Comma-separated IATA codes; only observations from the region's observers feed the union, events none of them heard are skipped. Events are split before this filter. A region with no known observers is not filtered and returns network-wide data, as `/api/analytics/rf` does |
|
||||
| `window` | string | none | `1h`, `24h`, `3d`, `7d` or `30d` (relative to now) |
|
||||
| `from`, `to` | string (ISO) | none | Absolute window bounds, take precedence over `window` |
|
||||
| `bucket` | string | `1h` | `5m`, `15m`, `1h`, `6h` or `1d`; other values fall back to `1h` |
|
||||
|
||||
### Response `200`
|
||||
|
||||
```jsonc
|
||||
{
|
||||
"bucket_seconds": number,
|
||||
"window": string, // window label, "" for all data
|
||||
"region": string,
|
||||
"summary": {
|
||||
"packets": number, // flood events that started in the window
|
||||
"avg_repeaters": number, // mean distinct repeaters per event
|
||||
"observers": number, // distinct observers that heard them
|
||||
"one_byte_packets": number, // events on 1-byte hop hashes (most ambiguous)
|
||||
"no_repeater_packets": number // events heard with an empty path only
|
||||
},
|
||||
"buckets": [ // ascending, empty buckets omitted
|
||||
{
|
||||
"start": string (ISO), // bucket start, UTC
|
||||
"packets": number, // flood events that started in the bucket
|
||||
"repeater_sum": number,
|
||||
"avg_repeaters": number,
|
||||
"observers": number
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## GET /api/analytics/channels
|
||||
|
||||
Channel analytics.
|
||||
|
||||
@@ -758,6 +758,12 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="analytics-card" id="retransmission-pressure">
|
||||
<h3><svg class="ph-icon" aria-hidden="true"><use href="/icons/phosphor-sprite.svg#ph-chart-line"/></svg> Retransmission Pressure (proxy)</h3>
|
||||
<p class="text-muted">Average number of distinct repeaters seen in the observed paths of each flood packet, over time.</p>
|
||||
<div id="rtxPressureChart"><div class="text-muted" style="padding:20px">Loading…</div></div>
|
||||
</div>
|
||||
|
||||
<div class="analytics-row">
|
||||
<div class="analytics-card flex-1">
|
||||
<h3><svg class="ph-icon" aria-hidden="true"><use href="/icons/phosphor-sprite.svg#ph-handshake"/></svg> Repeater Pair Heatmap</h3>
|
||||
@@ -808,6 +814,141 @@
|
||||
obsId === '__all' ? renderAllObserversReach(topo.perObserverReach) : renderPerObserverReach(topo.perObserverReach, obsId);
|
||||
});
|
||||
}
|
||||
loadRetransmissionPressure();
|
||||
}
|
||||
|
||||
// ===================== RETRANSMISSION PRESSURE (#1699) =====================
|
||||
// Bucket per time-window choice. "All data" stays on 1h so it matches the
|
||||
// server's recomputed default shape instead of a TTL-cache compute.
|
||||
var RTX_BUCKET_BY_WINDOW = { '1h': '5m', '24h': '1h', '7d': '1h', '30d': '6h' };
|
||||
function retransmissionBucketFor(win) {
|
||||
return Object.prototype.hasOwnProperty.call(RTX_BUCKET_BY_WINDOW, win) ? RTX_BUCKET_BY_WINDOW[win] : '1h';
|
||||
}
|
||||
|
||||
var _rtxRequestSeq = 0;
|
||||
async function loadRetransmissionPressure() {
|
||||
if (!document.getElementById('rtxPressureChart')) return;
|
||||
var seq = ++_rtxRequestSeq;
|
||||
var twEl = document.getElementById('analyticsTimeWindow');
|
||||
var tw = twEl ? twEl.value : '';
|
||||
var qs = RegionFilter.regionQueryString() +
|
||||
(tw ? '&window=' + encodeURIComponent(tw) : '') +
|
||||
'&bucket=' + retransmissionBucketFor(tw);
|
||||
var html;
|
||||
try {
|
||||
var data = await api('/analytics/retransmissions?' + qs.slice(1), { ttl: CLIENT_TTL.analyticsRF });
|
||||
html = renderRetransmissionChart(data);
|
||||
} catch (e) {
|
||||
html = '<div class="text-muted" role="alert" style="padding:20px">Failed to load retransmission pressure: ' + esc(e.message) + '</div>';
|
||||
}
|
||||
// A newer request (window/region change) or a tab switch wins.
|
||||
var host = document.getElementById('rtxPressureChart');
|
||||
if (seq !== _rtxRequestSeq || !host) return;
|
||||
host.innerHTML = html;
|
||||
}
|
||||
|
||||
// renderRetransmissionChart draws avg distinct repeaters per flood packet
|
||||
// (the primary line, with a y axis) plus two context series scaled to their
|
||||
// own maximum: flood packets per bucket (bars) and observers per bucket
|
||||
// (dashed). Buckets further apart than bucket_seconds break the lines, so a
|
||||
// gap in data is never drawn as a trend.
|
||||
function renderRetransmissionChart(data) {
|
||||
var noData = '<div class="text-muted" style="padding:20px">No flood packets in this window.</div>';
|
||||
var raw = data && Array.isArray(data.buckets) ? data.buckets : [];
|
||||
var pts = [];
|
||||
for (var i = 0; i < raw.length; i++) {
|
||||
var t = Date.parse(raw[i].start);
|
||||
if (!isFinite(t)) continue;
|
||||
pts.push({ t: t, start: String(raw[i].start), avg: Number(raw[i].avg_repeaters) || 0,
|
||||
packets: Number(raw[i].packets) || 0, observers: Number(raw[i].observers) || 0 });
|
||||
}
|
||||
if (!pts.length) return noData;
|
||||
var stepMs = (Number(data.bucket_seconds) || 3600) * 1000;
|
||||
var w = 800, h = 220, padL = 40, padR = 16, padT = 12, padB = 28;
|
||||
var plotW = w - padL - padR, plotH = h - padT - padB;
|
||||
var tMin = pts[0].t, tMax = pts[pts.length - 1].t;
|
||||
var maxAvg = 1, maxPkts = 1, maxObs = 1;
|
||||
pts.forEach(function (p) {
|
||||
if (p.avg > maxAvg) maxAvg = p.avg;
|
||||
if (p.packets > maxPkts) maxPkts = p.packets;
|
||||
if (p.observers > maxObs) maxObs = p.observers;
|
||||
});
|
||||
var yMax = Math.ceil(maxAvg);
|
||||
// The x domain spans whole buckets, so bars and points (at bucket
|
||||
// centres) stay inside the plot, also for a single bucket.
|
||||
var span = tMax + stepMs - tMin;
|
||||
function x(tt) { return padL + (tt + stepMs / 2 - tMin) / span * plotW; }
|
||||
function yAvg(v) { return padT + plotH - (v / yMax) * plotH; }
|
||||
function yObs(v) { return padT + plotH - (v / maxObs) * plotH * 0.9; }
|
||||
function f(n) { return n.toFixed(1); }
|
||||
|
||||
var svg = '<svg viewBox="0 0 ' + w + ' ' + h + '" style="width:100%;max-height:' + h + 'px" role="img" aria-label="Average distinct repeaters per flood packet over time"><title>Average distinct repeaters per flood packet over time</title>';
|
||||
for (var g = 0; g <= 4; g++) {
|
||||
var gy = padT + plotH * g / 4;
|
||||
svg += '<line x1="' + padL + '" y1="' + f(gy) + '" x2="' + (w - padR) + '" y2="' + f(gy) + '" stroke="var(--border)" stroke-dasharray="2"/>';
|
||||
svg += '<text x="' + (padL - 4) + '" y="' + f(gy + 4) + '" text-anchor="end" font-size="10" fill="var(--text-muted)">' + esc(String(Math.round(yMax * (4 - g) / 4 * 10) / 10)) + '</text>';
|
||||
}
|
||||
// Context: flood packets per bucket, scaled to 30% of the plot height.
|
||||
var barW = Math.max(1, stepMs / span * plotW * 0.8);
|
||||
pts.forEach(function (p) {
|
||||
var bh = (p.packets / maxPkts) * plotH * 0.3;
|
||||
svg += '<rect class="rtx-packets-bar" x="' + f(x(p.t) - barW / 2) + '" y="' + f(padT + plotH - bh) + '" width="' + f(barW) + '" height="' + f(bh) + '" fill="var(--accent)" opacity="0.15"/>';
|
||||
});
|
||||
// Split into runs of consecutive buckets.
|
||||
var runs = [], run = [];
|
||||
pts.forEach(function (p, idx) {
|
||||
if (idx > 0 && p.t - pts[idx - 1].t > stepMs) { runs.push(run); run = []; }
|
||||
run.push(p);
|
||||
});
|
||||
runs.push(run);
|
||||
runs.forEach(function (r) {
|
||||
if (r.length === 1) {
|
||||
svg += '<circle class="rtx-avg-dot" cx="' + f(x(r[0].t)) + '" cy="' + f(yAvg(r[0].avg)) + '" r="3" fill="var(--accent)"/>';
|
||||
return;
|
||||
}
|
||||
svg += '<polyline class="rtx-obs-line" points="' + r.map(function (p) { return f(x(p.t)) + ',' + f(yObs(p.observers)); }).join(' ') + '" fill="none" stroke="var(--text-muted)" stroke-width="1" stroke-dasharray="4 3"/>';
|
||||
svg += '<polyline class="rtx-avg-line" points="' + r.map(function (p) { return f(x(p.t)) + ',' + f(yAvg(p.avg)); }).join(' ') + '" fill="none" stroke="var(--accent)" stroke-width="2"/>';
|
||||
});
|
||||
// Per-bucket hover targets.
|
||||
pts.forEach(function (p) {
|
||||
var tip = p.start + '\nAvg distinct repeaters: ' + p.avg.toFixed(1) + '\nFlood packets: ' + p.packets + '\nObservers: ' + p.observers;
|
||||
svg += '<circle class="rtx-hit" cx="' + f(x(p.t)) + '" cy="' + f(yAvg(p.avg)) + '" r="5" fill="transparent"><title>' + esc(tip) + '</title></circle>';
|
||||
});
|
||||
var multiDay = tMax - tMin > 2 * 86400000;
|
||||
var labelEvery = Math.max(1, Math.ceil(pts.length / 6));
|
||||
for (var li = 0; li < pts.length; li += labelEvery) {
|
||||
var lbl = multiDay ? pts[li].start.slice(5, 10) : pts[li].start.slice(11, 16);
|
||||
svg += '<text x="' + f(x(pts[li].t)) + '" y="' + (h - 8) + '" text-anchor="middle" font-size="9" fill="var(--text-muted)">' + esc(lbl) + '</text>';
|
||||
}
|
||||
svg += '</svg>';
|
||||
|
||||
var s = (data && data.summary) || {};
|
||||
var sPackets = Number(s.packets) || 0;
|
||||
var oneBytePct = sPackets ? Math.round((Number(s.one_byte_packets) || 0) / sPackets * 100) : 0;
|
||||
var html = svg;
|
||||
html += '<div class="timeline-legend">' +
|
||||
'<span><span class="legend-dot" style="background:var(--accent)"></span>Avg distinct repeaters per flood packet</span>' +
|
||||
'<span><span class="legend-dot" style="background:var(--text-muted)"></span>Observers (dashed, scaled)</span>' +
|
||||
'<span><span class="legend-dot" style="background:var(--accent);opacity:0.3"></span>Flood packets (bars, scaled)</span>' +
|
||||
'</div>';
|
||||
html += '<div class="rf-stats">' +
|
||||
'<span>Avg: <strong>' + esc((Number(s.avg_repeaters) || 0).toFixed(1)) + ' repeaters/packet</strong></span>' +
|
||||
'<span>Flood packets: <strong>' + esc(sPackets.toLocaleString()) + '</strong></span>' +
|
||||
'<span>Observers: <strong>' + esc(String(Number(s.observers) || 0)) + '</strong></span>' +
|
||||
'<span>1-byte hashes: <strong>' + oneBytePct + '%</strong></span>' +
|
||||
'<span>Heard without repeaters: <strong>' + esc(String(Number(s.no_repeater_packets) || 0)) + '</strong></span>' +
|
||||
'</div>';
|
||||
html += '<p class="text-muted" style="font-size:12px;margin-top:8px">' +
|
||||
'A proxy for collision pressure, not a measured collision rate. ' +
|
||||
'It counts only repeaters that at least one observer heard, so observer coverage moves the line too: ' +
|
||||
'adding or losing observers (dashed line) changes it without any change on air. ' +
|
||||
'Hop prefixes are not resolved to nodes; a prefix counts once per flood, also when two repeaters share it, so the value is a lower bound, ' +
|
||||
'most of all for packets on 1-byte hashes. Flood routes only (TRACE and direct routes excluded). ' +
|
||||
'A packet heard again more than 5 minutes after its previous observation counts as a new flood; ' +
|
||||
'each flood sits in the bucket where it started, so the newest bucket may still be filling. ' +
|
||||
'The area filter does not apply to this chart.' +
|
||||
'</p>';
|
||||
return html;
|
||||
}
|
||||
|
||||
function renderRepeaterTable(repeaters) {
|
||||
@@ -2986,6 +3127,8 @@ function destroy() { _stopRolesRefresh(); _stopScopesRefresh(); _analyticsData =
|
||||
window._analyticsRenderMultiByteAdopters = renderMultiByteAdopters;
|
||||
window._analyticsHashStatCardsHtml = hashStatCardsHtml;
|
||||
window._analyticsRenderCollisionsFromServer = renderCollisionsFromServer;
|
||||
window._analyticsRetransmissionBucketFor = retransmissionBucketFor;
|
||||
window._analyticsRenderRetransmissionChart = renderRetransmissionChart;
|
||||
window._analyticsScopeAdvertsByRoleHtml = scopeAdvertsByRoleHtml;
|
||||
}
|
||||
|
||||
|
||||
@@ -115,6 +115,7 @@ node test-issue-1668-m2-contrast.js
|
||||
node test-issue-1668-m3-typography.js
|
||||
node test-issue-1668-m4-per-route.js
|
||||
node test-issue-1697-mqtt-mobile-e2e.js
|
||||
node test-issue-1699-retransmission-chart.js
|
||||
node test-issue-1705-subpath-contrast.js
|
||||
node test-issue-1753-copy-url-slash.js
|
||||
node test-issue-1770-mobile-row-clamp.js
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* #1699: Topology tab, retransmission pressure chart.
|
||||
*
|
||||
* Loads public/analytics.js into a stub browser context and exercises the
|
||||
* pure render helpers it exposes for testing:
|
||||
* - _analyticsRetransmissionBucketFor(window) picks a bucket per window
|
||||
* - _analyticsRenderRetransmissionChart(data) renders the SVG + captions
|
||||
*
|
||||
* Usage: node test-issue-1699-retransmission-chart.js
|
||||
*/
|
||||
'use strict';
|
||||
|
||||
const fs = require('fs');
|
||||
const path = require('path');
|
||||
const vm = require('vm');
|
||||
|
||||
let passed = 0, failed = 0;
|
||||
function assert(cond, msg) {
|
||||
if (cond) { passed++; console.log(' ✓ ' + msg); }
|
||||
else { failed++; console.error(' ✗ ' + msg); }
|
||||
}
|
||||
|
||||
const ctx = {
|
||||
console, Math, JSON, Date, Number, String, Array, Object, Set, Map, RegExp, URLSearchParams,
|
||||
setTimeout, clearTimeout, requestAnimationFrame: () => 0,
|
||||
getComputedStyle: () => ({ getPropertyValue: () => '' }),
|
||||
localStorage: { getItem: () => null, setItem() {}, removeItem() {} },
|
||||
registerPage: () => {},
|
||||
api: async () => ({}),
|
||||
CLIENT_TTL: {},
|
||||
RegionFilter: { regionQueryString: () => '' },
|
||||
AreaFilter: { areaQueryString: () => '' },
|
||||
};
|
||||
ctx.window = ctx;
|
||||
ctx.document = {
|
||||
documentElement: {},
|
||||
createElement: () => ({ style: {}, addEventListener() {} }),
|
||||
addEventListener() {}, removeEventListener() {},
|
||||
querySelector: () => null, querySelectorAll: () => [],
|
||||
getElementById: () => null,
|
||||
};
|
||||
vm.createContext(ctx);
|
||||
vm.runInContext(fs.readFileSync(path.join(__dirname, 'public/analytics.js'), 'utf8'), ctx);
|
||||
|
||||
const bucketFor = ctx._analyticsRetransmissionBucketFor;
|
||||
const render = ctx._analyticsRenderRetransmissionChart;
|
||||
|
||||
console.log('\n=== #1699 retransmission chart: exports ===');
|
||||
assert(typeof bucketFor === 'function', '_analyticsRetransmissionBucketFor exposed');
|
||||
assert(typeof render === 'function', '_analyticsRenderRetransmissionChart exposed');
|
||||
if (typeof bucketFor !== 'function' || typeof render !== 'function') {
|
||||
console.log(`\n${passed} passed, ${failed} failed`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
console.log('\n=== bucket per window ===');
|
||||
assert(bucketFor('') === '1h', 'all data -> 1h (matches the server default shape)');
|
||||
assert(bucketFor('1h') === '5m', '1h window -> 5m');
|
||||
assert(bucketFor('24h') === '1h', '24h window -> 1h');
|
||||
assert(bucketFor('7d') === '1h', '7d window -> 1h');
|
||||
assert(bucketFor('30d') === '6h', '30d window -> 6h');
|
||||
assert(bucketFor('bogus') === '1h', 'unknown window -> 1h');
|
||||
|
||||
console.log('\n=== empty data ===');
|
||||
const empty = render({ bucket_seconds: 3600, summary: { packets: 0 }, buckets: [] });
|
||||
assert(/No flood packets/.test(empty), 'empty buckets render a no-data message');
|
||||
assert(!/<svg/.test(empty), 'empty buckets render no SVG');
|
||||
assert(/No flood packets/.test(render(null)), 'null data renders the no-data message');
|
||||
|
||||
console.log('\n=== chart ===');
|
||||
const H = 3600;
|
||||
const data = {
|
||||
bucket_seconds: H,
|
||||
summary: { packets: 400, avg_repeaters: 7.5, observers: 20, one_byte_packets: 100, no_repeater_packets: 3 },
|
||||
buckets: [
|
||||
{ start: '2026-09-13T10:00:00Z', packets: 100, repeater_sum: 700, avg_repeaters: 7, observers: 18 },
|
||||
{ start: '2026-09-13T11:00:00Z', packets: 100, repeater_sum: 800, avg_repeaters: 8, observers: 20 },
|
||||
// 3h gap: the line must break here, not interpolate across it.
|
||||
{ start: '2026-09-13T14:00:00Z', packets: 100, repeater_sum: 600, avg_repeaters: 6, observers: 12 },
|
||||
{ start: '2026-09-13T15:00:00Z', packets: 100, repeater_sum: 900, avg_repeaters: 9, observers: 19 },
|
||||
],
|
||||
};
|
||||
const html = render(data);
|
||||
assert(/<svg/.test(html), 'renders an SVG');
|
||||
const avgLines = html.match(/class="rtx-avg-line"/g) || [];
|
||||
assert(avgLines.length === 2, 'avg line split into 2 segments at the gap (got ' + avgLines.length + ')');
|
||||
const obsLines = html.match(/class="rtx-obs-line"/g) || [];
|
||||
assert(obsLines.length === 2, 'observer line split into 2 segments at the gap (got ' + obsLines.length + ')');
|
||||
assert(/proxy/i.test(html), 'labelled as a proxy');
|
||||
assert(/not a measured collision rate/i.test(html), 'says it is not a measured collision rate');
|
||||
assert(/observer/i.test(html) && /coverage/i.test(html), 'caption names observer coverage bias');
|
||||
assert(/lower bound/i.test(html), 'caption names the 1-byte prefix lower bound');
|
||||
assert(/counts once per flood/i.test(html), 'caption says a prefix counts once per flood');
|
||||
assert(/more than 5 minutes/i.test(html) && /new flood/i.test(html), 'caption explains the 5 minute flood event split');
|
||||
assert(!/where they were first seen/i.test(html), 'caption no longer buckets by first_seen');
|
||||
assert(/25%/.test(html), '1-byte share shown as 25% (100 of 400)');
|
||||
assert(/7\.5/.test(html), 'summary average shown');
|
||||
assert(/area filter/i.test(html), 'caption says the area filter does not apply');
|
||||
assert(!/#[0-9a-fA-F]{3,8}\b/.test(html), 'no hardcoded hex colours in the chart markup');
|
||||
assert(/var\(--accent\)/.test(html), 'uses CSS variables for colours');
|
||||
|
||||
console.log('\n=== y axis and points ===');
|
||||
const avgPts = (html.match(/class="rtx-avg-line" points="([^"]*)"/g) || [])
|
||||
.map(s => s.replace(/.*points="([^"]*)"/, '$1').trim().split(/\s+/).length);
|
||||
assert(avgPts.join(',') === '2,2', 'each segment carries its 2 points (got ' + avgPts.join(',') + ')');
|
||||
assert(/>9</.test(html) || /9\.0/.test(html) || />10</.test(html), 'y axis reaches the max average');
|
||||
|
||||
function barsInPlot(markup) {
|
||||
const bars = [...markup.matchAll(/class="rtx-packets-bar" x="([\d.-]+)" y="[\d.-]+" width="([\d.-]+)"/g)];
|
||||
// viewBox 0..800 with 40px left and 16px right padding.
|
||||
return bars.length > 0 && bars.every(m => Number(m[1]) >= 40 && Number(m[1]) + Number(m[2]) <= 784);
|
||||
}
|
||||
assert(barsInPlot(html), 'packet bars stay inside the plot area');
|
||||
|
||||
console.log('\n=== single bucket ===');
|
||||
const one = render({ bucket_seconds: H, summary: { packets: 5, avg_repeaters: 3, observers: 2, one_byte_packets: 0 },
|
||||
buckets: [{ start: '2026-09-13T10:00:00Z', packets: 5, repeater_sum: 15, avg_repeaters: 3, observers: 2 }] });
|
||||
assert(/<circle[^>]*class="rtx-avg-dot"/.test(one), 'a lone bucket is drawn as a dot');
|
||||
assert(barsInPlot(one), 'a lone bucket bar stays inside the plot area');
|
||||
|
||||
console.log('\n=== untrusted server strings ===');
|
||||
const evil = render({ bucket_seconds: H, summary: { packets: 1, avg_repeaters: 1, observers: 1, one_byte_packets: 0 },
|
||||
buckets: [{ start: '<img src=x onerror=alert(1)>', packets: 1, repeater_sum: 1, avg_repeaters: 1, observers: 1 }] });
|
||||
assert(!/<img/.test(evil), 'an unparseable bucket start is dropped, never echoed');
|
||||
const tipHtml = render(data);
|
||||
assert(/<title>2026-09-13T10:00:00Z\nAvg distinct repeaters: 7\.0\nFlood packets: 100\nObservers: 18<\/title>/.test(tipHtml),
|
||||
'per-bucket tooltip carries start, average, packets and observers');
|
||||
|
||||
console.log(`\n${passed} passed, ${failed} failed`);
|
||||
process.exit(failed ? 1 : 0);
|
||||
Reference in New Issue
Block a user