feat: ingest observer /neighbors report as confirmed scope evidence (#1865)

The ESP32 observer firmware now emits a periodic /neighbors report carrying
the observer's own configured region scopes (`self`) plus, for each zero-hop
neighbor, the scopes fetched via an OTA scope query. This records that
CONFIRMED configuration into a new nodes.configured_scope column, kept
strictly separate from the existing inferred `default_scope` (observed advert
transport scope) and transported_scopes (transmissions.scope_name).

Provenance is modelled explicitly rather than overloading default_scope:
default_scope is overwritten on every advert observation, so writing neighbor
scopes there would let the next inferred observation clobber a confirmed
value. A dedicated configured_scope (+ configured_scope_at) column preserves
the distinction and structurally satisfies the report contract.

Report semantics honored:
- Only neighbors with status=="responded" update configured_scope. A timeout
  is NOT evidence the scopes were cleared, so it never writes.
- Absence of a neighbor is never a signal: the report is size-capped and
  truncates by ordering, so missing != gone — no deletes ever happen.
- A responded neighbor with empty scopes is a valid "no scopes configured"
  statement and IS stored (the handler gates on status, not emptiness).
- self scopes are keyed by origin_id (the observer node pubkey) and need no
  OTA query. Report pubkeys are uppercase; nodes.public_key is lowercase hex,
  so keys are lowercased before the UPDATE. Unknown neighbors are a no-op
  until a later advert creates the node.
- Out-of-order reports can't clobber newer data (last-write-wins on
  configured_scope_at).

Changes:
- dbschema: additive ensureConfiguredScopeColumns migration on nodes +
  inactive_nodes (marker nodes_configured_scope_v1), asserted via mustCol.
- ingestor: handleNeighborsReport dispatch on topic
  meshcore/<region>/<observer_id>/neighbors (analogous to /status);
  Store.UpdateNodeConfiguredScope writer.
- server: PRAGMA-detect configured_scope (hasConfiguredScope) and expose it +
  configured_scope_at on the node read path.
- UI: node-detail (nodes.js + live.js) shows a "Configured scope" row marked
  confirmed, with last-confirmed timestamp, distinct from the observed scope.
- tests: handleNeighborsReport (responded writes, timeout/absence never
  clears, empty-responded stored, unknown no-op) + last-write-wins.

Topic format meshcore/<region>/<observer_id>/neighbors is assumed by analogy
to the /status topic; noted for reviewer confirmation against the firmware.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
SaarMesh-Bot
2026-07-26 06:48:20 +00:00
co-authored by Claude
parent c7f0c6931f
commit ae2c9e9111
7 changed files with 314 additions and 0 deletions
+36
View File
@@ -1625,6 +1625,42 @@ func (s *Store) UpdateNodeDefaultScope(pubkey, scope string) error {
return err
}
// UpdateNodeConfiguredScope records the region scopes a node has CONFIGURED,
// as concrete evidence from an observer /neighbors report (#1865). Unlike
// UpdateNodeDefaultScope (inferred, overwritten on every observation), this is
// only called for the observer's own `self` scopes and for neighbors whose OTA
// scope query returned status="responded" — so the caller, not this method,
// gates on status. An empty scope IS a valid "responded, no scopes configured"
// statement and is stored; a timeout must simply not reach this method.
//
// reportedAt is the report envelope timestamp (ISO-8601). It is stored in
// configured_scope_at and used for last-write-wins: an out-of-order older
// report must not clobber a newer confirmed value. A blank reportedAt skips
// the ordering guard (always writes).
func (s *Store) UpdateNodeConfiguredScope(pubkey, scope, reportedAt string) error {
if pubkey == "" {
return nil
}
// Last-write-wins: skip if the stored confirmation is newer-or-equal.
if reportedAt != "" {
var curAt sql.NullString
row := s.db.QueryRow(`SELECT configured_scope_at FROM nodes WHERE public_key = ?`, pubkey)
if row.Scan(&curAt) == nil && curAt.Valid && curAt.String >= reportedAt {
return nil
}
}
if _, err := s.db.Exec(
`UPDATE nodes SET configured_scope = ?, configured_scope_at = ? WHERE public_key = ?`,
scope, reportedAt, pubkey); err != nil {
return err
}
// Mirror to inactive_nodes (node may be there if recently moved by retention).
_, err := s.db.Exec(
`UPDATE inactive_nodes SET configured_scope = ?, configured_scope_at = ? WHERE public_key = ?`,
scope, reportedAt, pubkey)
return err
}
// RecordNaiveSkew is called when resolveRxTime() clamps a packet's envelope
// timestamp because the observer is emitting a zone-less local-time string
// off from UTC by more than 15 min (issue #1478). Stamps the observer's
+160
View File
@@ -0,0 +1,160 @@
package main
// Tests for #1865: ingest the observer /neighbors report as concrete evidence
// of configured region scopes. Covers handleNeighborsReport dispatch semantics
// and the UpdateNodeConfiguredScope store method (status gating, case folding,
// last-write-wins, and the "absence/timeout is not a signal" contract).
import (
"database/sql"
"path/filepath"
"testing"
)
// seedNode inserts a node into both nodes and inactive_nodes (lowercase key).
func seedNode(t *testing.T, store *Store, pubkey string) {
t.Helper()
if _, err := store.db.Exec(`INSERT INTO nodes (public_key, name) VALUES (?, ?)`, pubkey, "n_"+pubkey[:4]); err != nil {
t.Fatalf("seed node %s: %v", pubkey, err)
}
if _, err := store.db.Exec(`INSERT INTO inactive_nodes (public_key, name) VALUES (?, ?)`, pubkey, "n_"+pubkey[:4]); err != nil {
t.Fatalf("seed inactive node %s: %v", pubkey, err)
}
}
func configuredScope(t *testing.T, store *Store, pubkey string) (sql.NullString, sql.NullString) {
t.Helper()
var sc, at sql.NullString
if err := store.db.QueryRow(
`SELECT configured_scope, configured_scope_at FROM nodes WHERE public_key = ?`, pubkey,
).Scan(&sc, &at); err != nil {
t.Fatalf("read configured_scope for %s: %v", pubkey, err)
}
return sc, at
}
func openNeighborsStore(t *testing.T) *Store {
t.Helper()
store, err := OpenStore(filepath.Join(t.TempDir(), "test.db"))
if err != nil {
t.Fatalf("OpenStore: %v", err)
}
t.Cleanup(func() { store.Close() })
return store
}
func TestHandleNeighborsReportWritesSelfAndResponded(t *testing.T) {
store := openNeighborsStore(t)
// Report keys are UPPERCASE; nodes.public_key is lowercase hex.
const originUpper = "FEEDCA4AD4E2AE615AAAB3CB73FAEC6EF0C7AF4D410F5C58A70FC0F724B7C933"
const respUpper = "B0D17C59FCF580592F8FB78B67D2F0CE9E9187EF3483A765BDFF1D7947A5109C"
const timeoutUpper = "0CE5EA7CFA3AB01D11810EF56B73DD899CD6C58644D6A6832A5C1AE89AFC5E25"
originLower := "feedca4ad4e2ae615aaab3cb73faec6ef0c7af4d410f5c58a70fc0f724b7c933"
respLower := "b0d17c59fcf580592f8fb78b67d2f0ce9e9187ef3483a765bdff1d7947a5109c"
timeoutLower := "0ce5ea7cfa3ab01d11810ef56b73dd899cd6c58644d6a6832a5c1ae89afc5e25"
seedNode(t, store, originLower)
seedNode(t, store, respLower)
seedNode(t, store, timeoutLower)
// Pre-existing confirmed scope on the timeout node — a timeout must NOT clobber it.
if err := store.UpdateNodeConfiguredScope(timeoutLower, "eu", "2026-07-24T00:00:00Z"); err != nil {
t.Fatal(err)
}
report := map[string]interface{}{
"timestamp": "2026-07-25T10:46:14.000000+00:00",
"origin_id": originUpper,
"self": map[string]interface{}{"scopes": "*"},
"neighbors": []interface{}{
map[string]interface{}{"pubkey": timeoutUpper, "scopes": "", "status": "timeout"},
map[string]interface{}{"pubkey": respUpper, "scopes": "de,eu", "status": "responded"},
},
}
handleNeighborsReport(store, "test", "obs-topic-id", report)
// self scopes written under the lowercased origin_id.
if sc, _ := configuredScope(t, store, originLower); !sc.Valid || sc.String != "*" {
t.Errorf("self configured_scope = %v, want '*'", sc)
}
// responded neighbor written, lowercased.
if sc, _ := configuredScope(t, store, respLower); !sc.Valid || sc.String != "de,eu" {
t.Errorf("responded configured_scope = %v, want 'de,eu'", sc)
}
// timeout neighbor untouched — prior confirmed value survives.
if sc, _ := configuredScope(t, store, timeoutLower); !sc.Valid || sc.String != "eu" {
t.Errorf("timeout node configured_scope = %v, want prior 'eu' (must not be cleared)", sc)
}
}
func TestHandleNeighborsReportUnknownNeighborIsNoop(t *testing.T) {
store := openNeighborsStore(t)
// No node seeded for this pubkey — UPDATE must match no row (no insert, no error).
report := map[string]interface{}{
"timestamp": "2026-07-25T10:46:14Z",
"neighbors": []interface{}{
map[string]interface{}{"pubkey": "aa" + "00000000000000000000000000000000000000000000000000000000000000"[2:], "scopes": "de", "status": "responded"},
},
}
handleNeighborsReport(store, "test", "obs", report)
var n int
if err := store.db.QueryRow(`SELECT COUNT(*) FROM nodes`).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Errorf("nodes count = %d, want 0 (report must not create nodes)", n)
}
}
func TestHandleNeighborsReportRespondedEmptyScopeIsStored(t *testing.T) {
store := openNeighborsStore(t)
pk := "cc00000000000000000000000000000000000000000000000000000000000001"
seedNode(t, store, pk)
// A responded query with empty scopes is a valid "no scopes configured".
report := map[string]interface{}{
"timestamp": "2026-07-25T10:46:14Z",
"neighbors": []interface{}{
map[string]interface{}{"pubkey": pk, "scopes": "", "status": "responded"},
},
}
handleNeighborsReport(store, "test", "obs", report)
sc, at := configuredScope(t, store, pk)
if !sc.Valid || sc.String != "" {
t.Errorf("responded-empty configured_scope = %v, want stored empty string", sc)
}
if !at.Valid || at.String == "" {
t.Errorf("configured_scope_at = %v, want the report timestamp", at)
}
}
func TestUpdateNodeConfiguredScopeLastWriteWins(t *testing.T) {
store := openNeighborsStore(t)
pk := "dd00000000000000000000000000000000000000000000000000000000000001"
seedNode(t, store, pk)
if err := store.UpdateNodeConfiguredScope(pk, "eu", "2026-07-25T12:00:00Z"); err != nil {
t.Fatal(err)
}
// Older report must NOT clobber the newer confirmed value.
if err := store.UpdateNodeConfiguredScope(pk, "stale", "2026-07-24T00:00:00Z"); err != nil {
t.Fatal(err)
}
if sc, _ := configuredScope(t, store, pk); sc.String != "eu" {
t.Errorf("configured_scope = %q, want 'eu' (older report must not overwrite)", sc.String)
}
// Newer report updates.
if err := store.UpdateNodeConfiguredScope(pk, "de", "2026-07-26T00:00:00Z"); err != nil {
t.Fatal(err)
}
if sc, _ := configuredScope(t, store, pk); sc.String != "de" {
t.Errorf("configured_scope = %q, want 'de' (newer report should update)", sc.String)
}
// inactive_nodes mirrored.
var inactive sql.NullString
if err := store.db.QueryRow(`SELECT configured_scope FROM inactive_nodes WHERE public_key = ?`, pk).Scan(&inactive); err != nil {
t.Fatal(err)
}
if inactive.String != "de" {
t.Errorf("inactive_nodes.configured_scope = %q, want 'de'", inactive.String)
}
}
+62
View File
@@ -595,6 +595,17 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message,
return
}
// Neighbors report topic: meshcore/<region>/<observer_id>/neighbors (#1865).
// The ESP32 observer firmware emits a periodic neighbor report carrying its
// own configured region scopes (`self`) plus, for each zero-hop neighbor,
// the scopes fetched via an OTA scope query. Like /status this is observer
// metadata (region-independent), so the per-source packet IATA filter below
// does not apply.
if len(parts) >= 4 && parts[3] == "neighbors" {
handleNeighborsReport(store, tag, parts[2], msg)
return
}
// Status topic: meshcore/<region>/<observer_id>/status
// Per-source IATA filter does NOT apply here — observer metadata (noise_floor, battery, etc.)
// is region-independent and should be accepted from all observers regardless of
@@ -1487,6 +1498,57 @@ func init() {
}
}
// handleNeighborsReport ingests an observer /neighbors report (#1865) and
// records CONFIRMED region scopes into nodes.configured_scope:
// - the observer's own scopes from `self`, keyed by origin_id (the observer
// node pubkey), which need no OTA query and are always trusted; and
// - each neighbor whose OTA scope query returned status=="responded".
//
// Per the report contract: neighbors with any other status (e.g. "timeout")
// are skipped — a failed query is NOT evidence the scopes were cleared — and a
// missing neighbor is never a signal (the report is 10 KB-capped and truncates
// by ordering, so absent != gone). Report pubkeys are uppercase; nodes.public_key
// is lowercase hex, so keys are lowercased before the UPDATE. Unknown neighbors
// are a no-op (the UPDATE matches no row) until a later advert creates the node.
func handleNeighborsReport(store *Store, tag string, observerID string, msg map[string]interface{}) {
reportedAt, _ := msg["timestamp"].(string)
// self: the observer's own configured scopes.
originID, _ := msg["origin_id"].(string)
if originID == "" {
originID = observerID
}
originID = strings.ToLower(originID)
if self, ok := msg["self"].(map[string]interface{}); ok && originID != "" {
if sc, ok := self["scopes"].(string); ok {
if err := store.UpdateNodeConfiguredScope(originID, sc, reportedAt); err != nil {
log.Printf("MQTT [%s] neighbors self scope error: %v", tag, err)
}
}
}
// neighbors[]: only status=="responded" carries usable scope evidence.
neighbors, _ := msg["neighbors"].([]interface{})
for _, raw := range neighbors {
n, ok := raw.(map[string]interface{})
if !ok {
continue
}
if status, _ := n["status"].(string); status != "responded" {
continue
}
pubkey, _ := n["pubkey"].(string)
pubkey = strings.ToLower(pubkey)
if pubkey == "" {
continue
}
scopes, _ := n["scopes"].(string)
if err := store.UpdateNodeConfiguredScope(pubkey, scopes, reportedAt); err != nil {
log.Printf("MQTT [%s] neighbors scope error for %.8s: %v", tag, pubkey, err)
}
}
}
// shouldUpdateDefaultScope returns true when the packet carries a transport
// scope whose region key matched (#1534). Without the ScopeName non-empty
// guard, transport-scoped adverts from non-matching regions would overwrite
+15
View File
@@ -39,6 +39,7 @@ type DB struct {
hasObsRawHex bool // observations table has raw_hex column (#881)
hasScopeName bool // transmissions.scope_name column exists (#899)
hasDefaultScope bool // nodes.default_scope column exists (#899)
hasConfiguredScope bool // nodes.configured_scope column exists (#1865)
hasMultibyteSupCols bool // nodes/inactive_nodes have multibyte_sup/multibyte_evidence (#903)
hasLastSeen bool // transmissions.last_seen column exists (#1690)
@@ -139,6 +140,8 @@ func (db *DB) detectSchema() {
switch colName {
case "default_scope":
db.hasDefaultScope = true
case "configured_scope":
db.hasConfiguredScope = true
case "multibyte_sup":
db.hasMultibyteSupCols = true
}
@@ -153,6 +156,10 @@ func (db *DB) nodeSelectCols() string {
if db.hasDefaultScope {
cols += ", default_scope"
}
// #1865: confirmed scopes appended after default_scope; scan order must match.
if db.hasConfiguredScope {
cols += ", configured_scope, configured_scope_at"
}
return cols
}
@@ -2241,11 +2248,15 @@ func (db *DB) scanNodeRow(rows *sql.Rows) map[string]interface{} {
var temperatureC sql.NullFloat64
var foreign sql.NullInt64
var defaultScope sql.NullString
var configuredScope, configuredScopeAt sql.NullString
scanArgs := []interface{}{&pk, &name, &role, &lat, &lon, &lastSeen, &firstSeen, &advertCount, &batteryMv, &temperatureC, &foreign}
if db.hasDefaultScope {
scanArgs = append(scanArgs, &defaultScope)
}
if db.hasConfiguredScope {
scanArgs = append(scanArgs, &configuredScope, &configuredScopeAt)
}
if err := rows.Scan(scanArgs...); err != nil {
return nil
}
@@ -2276,6 +2287,10 @@ func (db *DB) scanNodeRow(rows *sql.Rows) map[string]interface{} {
if db.hasDefaultScope {
m["default_scope"] = nullStr(defaultScope)
}
if db.hasConfiguredScope {
m["configured_scope"] = nullStr(configuredScope)
m["configured_scope_at"] = nullStr(configuredScopeAt)
}
return m
}
+39
View File
@@ -73,6 +73,9 @@ func Apply(rw *sql.DB, logf Logger) error {
if err := ensureDefaultScopeColumns(rw, logf); err != nil {
return fmt.Errorf("ensure default_scope: %w", err)
}
if err := ensureConfiguredScopeColumns(rw, logf); err != nil {
return fmt.Errorf("ensure configured_scope: %w", err)
}
if err := ensureObservationsRawHexColumn(rw, logf); err != nil {
return fmt.Errorf("ensure observations.raw_hex: %w", err)
}
@@ -144,6 +147,9 @@ func AssertReady(ro *sql.DB) error {
mustCol("transmissions", "scope_name")
mustCol("nodes", "default_scope")
mustCol("inactive_nodes", "default_scope")
// #1865: confirmed region scopes from the observer /neighbors report.
mustCol("nodes", "configured_scope")
mustCol("inactive_nodes", "configured_scope")
mustCol("observations", "raw_hex")
// Multi-byte capability cache (#1324 follow-up; PR #903 surface).
// Owned by ingestor — server reads these for O(1) /api/nodes
@@ -429,6 +435,39 @@ func ensureDefaultScopeColumns(rw *sql.DB, logf Logger) error {
return nil
}
// ensureConfiguredScopeColumns adds nodes.configured_scope +
// nodes.configured_scope_at (and mirrors on inactive_nodes) for #1865.
// Unlike default_scope (inferred from observed advert transport scope, and
// overwritten on every observation), configured_scope holds the region scopes
// a node has CONFIGURED, taken as concrete evidence from the observer
// /neighbors report — written only for the observer's own `self` scopes and
// for neighbors whose OTA scope query returned status="responded". The
// server PRAGMA-detects configured_scope as hasConfiguredScope.
func ensureConfiguredScopeColumns(rw *sql.DB, logf Logger) error {
if err := ensureMigrationsTable(rw); err != nil {
return err
}
for _, table := range []string{"nodes", "inactive_nodes"} {
for _, col := range []string{"configured_scope", "configured_scope_at"} {
has, err := TableHasColumn(rw, table, col)
if err != nil {
return fmt.Errorf("inspect %s.%s: %w", table, col, err)
}
if has {
continue
}
if _, err := rw.Exec(fmt.Sprintf(`ALTER TABLE %s ADD COLUMN %s TEXT DEFAULT NULL`, table, col)); err != nil {
return fmt.Errorf("alter %s add %s: %w", table, col, err)
}
logf("[dbschema] added %s column to %s (#1865)", col, table)
}
}
if _, err := rw.Exec(`INSERT OR IGNORE INTO _migrations (name) VALUES ('nodes_configured_scope_v1')`); err != nil {
return fmt.Errorf("record nodes_configured_scope_v1: %w", err)
}
return nil
}
// ensureObservationsRawHexColumn adds observations.raw_hex (#881).
// Source of truth lives here per #1321 (was previously cmd/ingestor/db.go only):
// the server PRAGMA-detects this column as hasObsRawHex.
+1
View File
@@ -2556,6 +2556,7 @@
<table style="font-size:12px;width:100%;border-collapse:collapse;">
<tr><td style="color:var(--text-muted);padding:4px 8px 4px 0;">Last Seen</td><td>${lastSeen}</td></tr>
<tr><td style="color:var(--text-muted);padding:4px 8px 4px 0;">Adverts</td><td>${n.advert_count || 0}</td></tr>
${'configured_scope' in n && n.configured_scope !== null ? `<tr><td style="color:var(--text-muted);padding:4px 8px 4px 0;" title="Region scopes this node has configured, confirmed via an observer /neighbors report (status=responded) — concrete evidence (#1865).${n.configured_scope_at ? ' Last confirmed ' + escapeHtml(String(n.configured_scope_at)) + '.' : ''}">Configured scope <span style="color:var(--status-green,#2ecc71)" aria-label="confirmed">✓</span></td><td>${n.configured_scope === '' ? '<span style="color:var(--text-muted)">none configured</span>' : `<code style="color:var(--link-color)">${escapeHtml(n.configured_scope)}</code>`}</td></tr>` : ''}
${'default_scope' in n ? `<tr><td style="color:var(--text-muted);padding:4px 8px 4px 0;">Scope</td><td>${n.default_scope === null ? '<span style="color:var(--text-muted)">—</span>'
: n.default_scope === '' ? '<span style="color:var(--text-muted)">unknown scope</span>'
: `<code style="color:var(--link-color)">${escapeHtml(n.default_scope)}</code>`
+1
View File
@@ -674,6 +674,7 @@
return `<tr id="row-bridge-score" data-bridge-score="${b.toFixed(4)}"><td title="${btooltip}">Bridge score <span style="color:var(--text-muted);cursor:help" aria-label="help">ⓘ</span></td><td><span style="display:inline-block;vertical-align:middle;width:80px;height:8px;background:var(--bg-secondary,#333);border-radius:4px;overflow:hidden;margin-right:6px"><span style="display:block;width:${bbarWidth}%;height:100%;background:${bcolor}"></span></span><span style="color:${bcolor};font-weight:600">${bpct}%</span> <span style="color:var(--text-muted);font-size:11px;margin-left:4px">${blabel}</span></td></tr>`;
})() : ''}
${(n.role === 'repeater' || n.role === 'room') && Array.isArray(n.transported_scopes) && n.transported_scopes.length ? `<tr id="row-transported-scopes"><td title="Distinct region scopes (transmissions.scope_name) of all non-advert packets in which this repeater appears as a path hop. Shows which regions' traffic this repeater has carried (#1751).">Transported scopes</td><td><span style="display:inline-flex;flex-wrap:wrap;gap:3px;vertical-align:middle">${n.transported_scopes.map(sc => '<span class="badge">' + escapeHtml(String(sc)) + '</span>').join('')}</span></td></tr>` : ''}
${'configured_scope' in n && n.configured_scope !== null ? `<tr id="row-configured-scope"><td title="Region scopes this node has CONFIGURED, confirmed via an observer /neighbors report (status=responded) — concrete evidence, distinct from observed default scope and transported scopes (#1865).${n.configured_scope_at ? ' Last confirmed ' + escapeHtml(String(n.configured_scope_at)) + '.' : ''}">Configured scope <span style="color:var(--status-green,#2ecc71)" aria-label="confirmed">✓</span></td><td>${n.configured_scope === '' ? '<span style="color:var(--text-muted)">none configured</span>' : `<code style="color:var(--link-color)">${escapeHtml(n.configured_scope)}</code>`}</td></tr>` : ''}
<tr><td>First Seen</td><td>${renderNodeTimestampHtml(n.first_seen)}</td></tr>
<tr><td>Total Packets</td><td>${stats.totalTransmissions || stats.totalPackets || n.advert_count || 0}${stats.totalObservations && stats.totalObservations !== (stats.totalTransmissions || stats.totalPackets) ? ' <span class="text-muted" style="font-size:0.85em">(seen ' + stats.totalObservations + '×)</span>' : ''}</td></tr>
<tr><td>Packets Today</td><td>${stats.packetsToday || 0}</td></tr>