feat(#1290): exclude listener-only observers from path-hop disambiguator

Firmware 1.16 publishes a `repeat: on|off` flag in the MQTT /status
JSON (confirmed by cwichura on the issue; see
MQTTMessageBuilder.cpp:58 in agessaman/MeshCore
mqtt-bridge-implementation-flex). Listener-only observers
(repeat:off) by firmware contract never relay packets, so they
cannot legitimately be a hop in someone else's resolved path.

Pipeline:

* internal/dbschema: idempotent `can_relay BOOLEAN DEFAULT 1`
  migration on observers, plus AssertReady probe (server fatal-logs
  if absent). Mirrored in cmd/ingestor/db.go CREATE TABLE for fresh
  DBs.

* cmd/ingestor: `extractObserverMeta` accepts `repeat` as bool,
  case-insensitive string (`on|off|true|false|yes|no`), or numeric
  0/1; missing field → nil → COALESCE preserves prior column value
  (back-compat with legacy observers). Plumbed through
  UpsertObserverAt and the prepared upsert statement.

* cmd/server: GetNonRelayObserverPubkeys + new prefixMap.markNonRelay
  drop matching candidates inside resolveWithContext. Filter is
  applied at the top of the resolver so all 4 tiers see the
  pruned candidate set. ObserverResp.CanRelay is surfaced to
  /api/observers and /api/observers/{id}; GetNodeHealth enriches
  the per-observer rows with can_relay so the node-detail badge
  works. Probe-and-fall-back when can_relay column is absent
  (legacy test fixtures).

* public/: listener vs repeater pill on observers list,
  observer detail Relay stat card, and node-detail Heard By table.
  CSS pill styles inherit existing theme vars.

Green for the test added in the prior red commit:
TestResolveWithContext_ExcludesNonRelayObservers_Issue1290.
Back-compat preserved: legacy observers (no `repeat` field) keep
the default can_relay=1 and remain candidates.

cross-stack: justified — backend persists firmware-side `repeat` hint to a new observers column, frontend surfaces the listener/repeater status as a badge on the observers list and node-detail Heard By table per the issue's UI acceptance criterion.
This commit is contained in:
Kpa-clawbot
2026-06-08 06:51:04 +00:00
parent 5f7fdb9620
commit f12911dcc9
11 changed files with 222 additions and 13 deletions
+27 -7
View File
@@ -197,7 +197,8 @@ func applySchema(db *sql.DB) error {
last_packet_at TEXT DEFAULT NULL,
clock_skew_seconds INTEGER DEFAULT NULL,
clock_skew_count_24h INTEGER DEFAULT 0,
clock_last_naive_at TEXT DEFAULT NULL
clock_last_naive_at TEXT DEFAULT NULL,
can_relay INTEGER DEFAULT 1
);
CREATE INDEX IF NOT EXISTS idx_nodes_last_seen ON nodes(last_seen);
@@ -727,8 +728,8 @@ func (s *Store) prepareStatements() error {
}
s.stmtUpsertObserver, err = s.db.Prepare(`
INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count, model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor)
VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?)
INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count, model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor, can_relay)
VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, 1))
ON CONFLICT(id) DO UPDATE SET
name = COALESCE(?, name),
iata = COALESCE(?, iata),
@@ -740,7 +741,8 @@ func (s *Store) prepareStatements() error {
radio = COALESCE(?, radio),
battery_mv = COALESCE(?, battery_mv),
uptime_secs = COALESCE(?, uptime_secs),
noise_floor = COALESCE(?, noise_floor)
noise_floor = COALESCE(?, noise_floor),
can_relay = COALESCE(?, can_relay)
`)
if err != nil {
return err
@@ -973,6 +975,13 @@ type ObserverMeta struct {
RecvErrors *int // cumulative CRC/decode failures since boot
PacketsSent *int // cumulative packets sent since boot
PacketsRecv *int // cumulative packets received since boot
// CanRelay reflects the firmware 1.16 /status `repeat` flag (#1290).
// nil means the firmware did not send the field — caller must
// preserve the existing observers.can_relay value (default 1).
// true → relay-capable (`repeat:on`); false → listener-only
// (`repeat:off`), which causes the server-side disambiguator to
// exclude this observer's pubkey from path-hop candidate sets.
CanRelay *bool
}
// UpsertObserver inserts or updates an observer using the current wall-clock
@@ -995,7 +1004,7 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last
normalizedIATA := strings.TrimSpace(strings.ToUpper(iata))
var model, firmware, clientVersion, radio interface{}
var batteryMv, uptimeSecs, noiseFloor interface{}
var batteryMv, uptimeSecs, noiseFloor, canRelay interface{}
if meta != nil {
if meta.Model != nil {
model = *meta.Model
@@ -1018,11 +1027,22 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last
if meta.NoiseFloor != nil {
noiseFloor = *meta.NoiseFloor
}
// Issue #1290: nil → leave DB column unchanged (COALESCE in
// the prepared stmt); 0/1 written when firmware provided
// the `repeat` field. INSERT branch defaults to 1 via the
// COALESCE in the VALUES clause.
if meta.CanRelay != nil {
if *meta.CanRelay {
canRelay = 1
} else {
canRelay = 0
}
}
}
_, err := s.stmtUpsertObserver.Exec(
id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor,
name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor,
id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay,
name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay,
)
if err != nil {
s.Stats.WriteErrors.Add(1)
+31
View File
@@ -1124,6 +1124,37 @@ func extractObserverMeta(msg map[string]interface{}) *ObserverMeta {
}
}
// Issue #1290: firmware 1.16 publishes a `repeat` flag at the top
// level of the /status JSON (MQTTMessageBuilder.cpp:58 — see
// agessaman/MeshCore mqtt-bridge-implementation-flex). Accept
// either a boolean or a case-insensitive `on|off|true|false|1|0`
// string. Missing field → leave CanRelay nil; the writer preserves
// the prior column value (default 1, back-compat).
if v, ok := msg["repeat"]; ok && v != nil {
switch t := v.(type) {
case bool:
b := t
meta.CanRelay = &b
hasData = true
case string:
s := strings.ToLower(strings.TrimSpace(t))
switch s {
case "on", "true", "1", "yes":
b := true
meta.CanRelay = &b
hasData = true
case "off", "false", "0", "no":
b := false
meta.CanRelay = &b
hasData = true
}
case float64:
b := t != 0
meta.CanRelay = &b
hasData = true
}
}
if !hasData {
return nil
}
+56 -4
View File
@@ -12,6 +12,7 @@ import (
"sync"
"time"
"github.com/meshcore-analyzer/dbschema"
"github.com/meshcore-analyzer/geofilter"
_ "modernc.org/sqlite"
)
@@ -251,6 +252,11 @@ type Observer struct {
ClockSkewSeconds *int64 `json:"clock_skew_seconds"`
ClockSkewCount24h int `json:"clock_skew_count_24h"`
ClockLastNaiveAt *string `json:"clock_last_naive_at"`
// Issue #1290: firmware 1.16 `repeat: on|off` flag persisted by the
// ingestor. true = relay-capable (default for legacy observers).
// false = listener-only; cmd/server/store.go pm.resolveWithContext
// excludes these from path-hop candidate sets.
CanRelay bool `json:"can_relay"`
}
// Transmission represents a row from the transmissions table.
@@ -1148,9 +1154,17 @@ func (db *DB) getObservationsForTransmissions(txIDs []int) map[int][]map[string]
// GetObservers returns active observers (not soft-deleted) sorted by last_seen DESC.
func (db *DB) GetObservers() ([]Observer, error) {
// Issue #1290: can_relay is read via COALESCE(can_relay, 1). The
// column is added by internal/dbschema; older test fixtures and
// pre-migration DBs may lack it, so we probe and fall back.
canRelayClause := "COALESCE(can_relay, 1)"
if hasCol, _ := dbschema.TableHasColumn(db.conn, "observers", "can_relay"); !hasCol {
canRelayClause = "1"
}
rows, err := db.conn.Query(`SELECT id, name, iata, last_seen, first_seen, packet_count,
model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor, last_packet_at,
clock_skew_seconds, clock_skew_count_24h, clock_last_naive_at
clock_skew_seconds, clock_skew_count_24h, clock_last_naive_at,
` + canRelayClause + `
FROM observers WHERE inactive IS NULL OR inactive = 0 ORDER BY last_seen DESC`)
if err != nil {
return nil, err
@@ -1163,11 +1177,13 @@ func (db *DB) GetObservers() ([]Observer, error) {
var batteryMv, uptimeSecs, clockSkewSec sql.NullInt64
var clockSkewCount sql.NullInt64
var noiseFloor sql.NullFloat64
var canRelay int
if err := rows.Scan(&o.ID, &o.Name, &o.IATA, &o.LastSeen, &o.FirstSeen, &o.PacketCount,
&o.Model, &o.Firmware, &o.ClientVersion, &o.Radio, &batteryMv, &uptimeSecs, &noiseFloor, &o.LastPacketAt,
&clockSkewSec, &clockSkewCount, &o.ClockLastNaiveAt); err != nil {
&clockSkewSec, &clockSkewCount, &o.ClockLastNaiveAt, &canRelay); err != nil {
continue
}
o.CanRelay = canRelay != 0
if batteryMv.Valid {
v := int(batteryMv.Int64)
o.BatteryMv = &v
@@ -1190,22 +1206,58 @@ func (db *DB) GetObservers() ([]Observer, error) {
return observers, nil
}
// GetNonRelayObserverPubkeys returns the lowercase observer.id pubkeys
// for observers that have advertised `repeat:off` (#1290). The server's
// path-hop disambiguator consumes this to exclude listener-only nodes
// from the candidate set. Inactive observers are excluded for
// consistency with GetObservers; reactivation flips can_relay only on
// the next status message.
func (db *DB) GetNonRelayObserverPubkeys() ([]string, error) {
// Graceful no-op when can_relay column is absent (legacy DB / older
// test fixture). Avoids noisy schema-degradation log spam.
if hasCol, _ := dbschema.TableHasColumn(db.conn, "observers", "can_relay"); !hasCol {
return nil, nil
}
rows, err := db.conn.Query(`SELECT LOWER(id) FROM observers
WHERE COALESCE(can_relay, 1) = 0
AND (inactive IS NULL OR inactive = 0)`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []string
for rows.Next() {
var pk string
if err := rows.Scan(&pk); err == nil && pk != "" {
out = append(out, pk)
}
}
return out, rows.Err()
}
// GetObserverByID returns a single observer.
func (db *DB) GetObserverByID(id string) (*Observer, error) {
var o Observer
var batteryMv, uptimeSecs, clockSkewSec sql.NullInt64
var clockSkewCount sql.NullInt64
var noiseFloor sql.NullFloat64
var canRelay int
canRelayClause := "COALESCE(can_relay, 1)"
if hasCol, _ := dbschema.TableHasColumn(db.conn, "observers", "can_relay"); !hasCol {
canRelayClause = "1"
}
err := db.conn.QueryRow(`SELECT id, name, iata, last_seen, first_seen, packet_count,
model, firmware, client_version, radio, battery_mv, uptime_secs, noise_floor, last_packet_at,
clock_skew_seconds, clock_skew_count_24h, clock_last_naive_at
clock_skew_seconds, clock_skew_count_24h, clock_last_naive_at,
`+canRelayClause+`
FROM observers WHERE id = ?`, id).
Scan(&o.ID, &o.Name, &o.IATA, &o.LastSeen, &o.FirstSeen, &o.PacketCount,
&o.Model, &o.Firmware, &o.ClientVersion, &o.Radio, &batteryMv, &uptimeSecs, &noiseFloor, &o.LastPacketAt,
&clockSkewSec, &clockSkewCount, &o.ClockLastNaiveAt)
&clockSkewSec, &clockSkewCount, &o.ClockLastNaiveAt, &canRelay)
if err != nil {
return nil, err
}
o.CanRelay = canRelay != 0
if batteryMv.Valid {
v := int(batteryMv.Int64)
o.BatteryMv = &v
+2
View File
@@ -2534,6 +2534,7 @@ func (s *Server) buildObserversDefaultResponse() (ObserverListResponse, error) {
LastPacketAt: o.LastPacketAt,
PacketsLastHour: plh,
Lat: lat, Lon: lon, NodeRole: nodeRole,
CanRelay: o.CanRelay,
}
applyObserverNaiveClock(&resp, o, nowTime)
result = append(result, resp)
@@ -2578,6 +2579,7 @@ func (s *Server) handleObserverDetail(w http.ResponseWriter, r *http.Request) {
NoiseFloor: obs.NoiseFloor,
LastPacketAt: obs.LastPacketAt,
PacketsLastHour: plh,
CanRelay: obs.CanRelay,
}
applyObserverNaiveClock(&resp, obs, time.Now().UTC())
return resp
+47
View File
@@ -6194,6 +6194,17 @@ func (s *PacketStore) getCachedNodesAndPM() ([]nodeInfo, *prefixMap) {
nodes := s.getAllNodes()
pm := buildPrefixMap(nodes)
// Issue #1290: exclude observers that advertised `repeat:off` from
// the path-hop candidate set. Failure is non-fatal — we log via the
// schema-degradation channel and proceed with an empty filter (i.e.
// pre-#1290 behavior).
if s.db != nil && s.db.conn != nil {
if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil {
pm.markNonRelay(pks)
} else {
s.logSchemaDegradationOnce("observers.can_relay read failed; path-hop disambiguator will not filter listener-only observers: " + err.Error())
}
}
s.cacheMu.Lock()
s.nodeCache = nodes
@@ -6257,6 +6268,25 @@ func (pm *prefixMap) resolve(hop string) *nodeInfo {
func (pm *prefixMap) resolveWithContext(hop string, contextPubkeys []string, graph *NeighborGraph) (*nodeInfo, string, float64) {
h := strings.ToLower(hop)
candidates := pm.m[h]
// Issue #1290: drop observer-known listener-only nodes from the
// candidate set. By firmware contract a node that advertises
// `repeat:off` in its MQTT /status will never relay a packet, so it
// cannot legitimately be a hop in someone else's path. Filtering
// here shrinks ambiguous candidate sets without affecting any
// upstream caller (the returned shape and confidence labels are
// preserved; only no_match becomes more likely when the only
// matching prefix belonged to a listener). Empty pm.nonRelay
// preserves the pre-#1290 behavior exactly (back-compat).
if len(pm.nonRelay) > 0 && len(candidates) > 0 {
filtered := candidates[:0:0]
for i := range candidates {
if _, isListener := pm.nonRelay[strings.ToLower(candidates[i].PublicKey)]; isListener {
continue
}
filtered = append(filtered, candidates[i])
}
candidates = filtered
}
if len(candidates) == 0 {
return nil, "no_match", 0
}
@@ -8870,6 +8900,18 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro
}
observerRows := make([]map[string]interface{}, 0)
// Issue #1290: surface listener/repeater hint on node detail by
// looking up can_relay for each observer that heard this node.
// One-shot fetch of the non-relay set keeps this O(observers) on
// rare events; nil on error degrades to "neither badge" client-side.
nonRelaySet := map[string]struct{}{}
if s.db != nil && s.db.conn != nil {
if pks, err := s.db.GetNonRelayObserverPubkeys(); err == nil {
for _, pk := range pks {
nonRelaySet[strings.ToUpper(pk)] = struct{}{}
}
}
}
for id, o := range observerStats {
var avgSnr, avgRssi interface{}
if o.snrCount > 0 {
@@ -8878,9 +8920,14 @@ func (s *PacketStore) GetNodeHealth(pubkey string) (map[string]interface{}, erro
if o.rssiCount > 0 {
avgRssi = o.rssiSum / float64(o.rssiCount)
}
canRelay := true
if _, isListener := nonRelaySet[strings.ToUpper(id)]; isListener {
canRelay = false
}
observerRows = append(observerRows, map[string]interface{}{
"observer_id": id, "observer_name": o.name,
"avgSnr": avgSnr, "avgRssi": avgRssi, "packetCount": o.count,
"can_relay": canRelay,
})
}
sort.Slice(observerRows, func(i, j int) bool {
+4
View File
@@ -909,6 +909,10 @@ type ObserverResp struct {
ClockSkewSeconds interface{} `json:"clock_skew_seconds"`
ClockSkewCount24h int `json:"clock_skew_count_24h"`
ClockLastNaiveAt interface{} `json:"clock_last_naive_at"`
// Issue #1290: firmware 1.16 `repeat` flag — true=repeater,
// false=listener-only. Drives the UI badge on observers list +
// node detail page. Defaults to true for legacy observers.
CanRelay bool `json:"can_relay"`
}
type ObserverListResponse struct {
+31
View File
@@ -82,6 +82,9 @@ func Apply(rw *sql.DB, logf Logger) error {
if err := ensureObserverNaiveClockColumns(rw, logf); err != nil {
return fmt.Errorf("ensure observers naive-clock columns: %w", err)
}
if err := ensureObserverCanRelayColumn(rw, logf); err != nil {
return fmt.Errorf("ensure observers.can_relay: %w", err)
}
return nil
}
@@ -138,6 +141,12 @@ func AssertReady(ro *sql.DB) error {
mustCol("observers", "clock_skew_seconds")
mustCol("observers", "clock_skew_count_24h")
mustCol("observers", "clock_last_naive_at")
// Issue #1290: firmware 1.16 publishes a `repeat: on|off` flag in
// the MQTT /status JSON. Ingestor persists it as can_relay; server
// reads it to filter listener-only observers out of the path-hop
// disambiguator candidate set. Default 1 preserves prior behavior
// for legacy observers that never sent the field.
mustCol("observers", "can_relay")
if len(missing) > 0 {
return fmt.Errorf("schema not migrated by ingestor; restart ingestor first. missing: %s",
@@ -522,3 +531,25 @@ func ensureObserverNaiveClockColumns(rw *sql.DB, logf Logger) error {
}
return nil
}
// ensureObserverCanRelayColumn adds the can_relay column to observers.
// Firmware 1.16 publishes a `repeat: on|off` flag in the MQTT /status
// JSON (#1290); the ingestor parses it and writes 0/1 here. The server's
// path-hop disambiguator (cmd/server/store.go pm.resolveWithContext)
// excludes observers with can_relay=0 from the candidate set. Default 1
// preserves prior behavior for legacy observers (no repeat field).
func ensureObserverCanRelayColumn(rw *sql.DB, logf Logger) error {
has, err := TableHasColumn(rw, "observers", "can_relay")
if err != nil {
return err
}
if has {
return nil
}
// PREFLIGHT: async=true reason="single-column ALTER on observers (low-cardinality, ~1k rows in prod); DEFAULT 1 is a constant so SQLite does the rewrite as a metadata-only schema update, no row scan"
if _, err := rw.Exec("ALTER TABLE observers ADD COLUMN can_relay INTEGER DEFAULT 1"); err != nil {
return err
}
logf("[dbschema] added can_relay column to observers")
return nil
}
+1 -1
View File
@@ -676,7 +676,7 @@
</tr></thead>
<tbody>
${observers.map(o => `<tr>
<td data-value="${escapeHtml((o.observer_name || o.observer_id || '').toLowerCase())}" style="font-weight:600">${escapeHtml(o.observer_name || o.observer_id)}</td>
<td data-value="${escapeHtml((o.observer_name || o.observer_id || '').toLowerCase())}" style="font-weight:600">${escapeHtml(o.observer_name || o.observer_id)}${o.can_relay === false ? ' <span class="badge-listener" title="Firmware reported repeat:off — excluded from path-hop disambiguator (#1290)">listener</span>' : (o.can_relay === true ? ' <span class="badge-repeater" title="Firmware reported repeat:on — eligible as a path hop">repeater</span>' : '')}</td>
<td data-value="${escapeHtml((o.iata || '').toLowerCase())}">${o.iata ? escapeHtml(o.iata) : '—'}</td>
<td data-value="${o.packetCount || 0}">${o.packetCount}</td>
<td data-value="${o.avgSnr != null ? Number(o.avgSnr) : ''}">${o.avgSnr != null ? Number(o.avgSnr).toFixed(1) + ' dB' : '—'}</td>
+4
View File
@@ -180,6 +180,10 @@ window.ObserverDetailNaiveBanner = {
<div class="stat-label">Status</div>
<div class="stat-value"><span class="health-dot ${statusCls}"></span> ${statusLabel}</div>
</div>
<div class="stat-card">
<div class="stat-label">Relay</div>
<div class="stat-value">${obs.can_relay === false ? '<span class="badge-listener" title="Firmware reported repeat:off — excluded from path-hop disambiguator (#1290)">listener</span>' : '<span class="badge-repeater" title="Firmware reported repeat:on (or unknown) — eligible as a path hop">repeater</span>'}</div>
</div>
<div class="stat-card">
<div class="stat-label">Region</div>
<div class="stat-value">${obs.iata ? '<span class="badge-region">' + escapeHtml(obs.iata) + '</span>' : '—'}</div>
+1 -1
View File
@@ -297,7 +297,7 @@ window.ObserversSummary = (function () {
const shape = h.cls === 'health-green' ? '●' : h.cls === 'health-yellow' ? '▲' : '✕';
return `<tr style="cursor:pointer" tabindex="0" role="row" data-action="navigate" data-value="#/observers/${encodeURIComponent(o.id)}" onclick="location.hash='#/observers/${encodeURIComponent(o.id)}'">
<td><span class="health-dot ${h.cls}" title="${h.label}">${shape}</span> ${h.label}</td>
<td class="mono">${escapeHtml(o.name || o.id)}${window.ObserversNaiveChip.render(o)}</td>
<td class="mono">${escapeHtml(o.name || o.id)}${window.ObserversNaiveChip.render(o)}${o.can_relay === false ? ' <span class="badge-listener" title="Firmware reported repeat:off — listener-only; excluded from path-hop disambiguator (issue #1290)">listener</span>' : ' <span class="badge-repeater" title="Firmware reported repeat:on (or unknown) — eligible as a path hop">repeater</span>'}</td>
<td>${o.iata ? `<span class="badge-region">${o.iata}</span>` : '—'}</td>
<td>${timeAgo(o.last_seen)}</td>
<td>${o.last_packet_at ? timeAgo(o.last_packet_at) : '<span class="text-muted">—</span>'}</td>
+18
View File
@@ -1146,6 +1146,24 @@ body.scroll-locked { overflow: hidden; }
font-size: 10px; font-weight: 700; font-family: var(--mono);
background: var(--nav-bg); color: var(--nav-text); letter-spacing: .5px;
}
/* Issue #1290: listener vs repeater badge on observers list + detail.
* Driven by ObserverResp.can_relay (false = firmware reported repeat:off).
* Server-side disambiguator excludes listener observers from path-hop
* candidates; this badge surfaces the same hint to operators. Colors use
* existing theme vars so the badge tracks the active palette. */
.badge-listener,
.badge-repeater {
display: inline-block; padding: 1px 5px; border-radius: 4px;
font-size: 10px; font-weight: 700; font-family: var(--mono);
letter-spacing: .5px; margin-left: 4px; vertical-align: middle;
}
.badge-listener {
background: var(--transport-badge-bg, #f59e0b20);
color: var(--transport-badge-fg, #d97706);
}
.badge-repeater {
background: var(--nav-bg); color: var(--nav-text);
}
/* Observer IATA pill rendered inline next to observer name on packets (#1188).
* Visually similar to .badge-region but distinct so the row badge and the
* inline-with-observer badge can be styled independently in future themes. */