feat(ingestor): RF environment samples from mobile clients (#1906)

> **Stacked on #1905.** This branch contains #1905's commits; review and
merge that one first. The diff unique to this PR is the
`client_rf_samples` table, its handler, the delta query and its
retention.

## What

Everything CoreDrive RX records today is anchored to a *packet*. But a
drive also passes through RF conditions that exist whether or not a
packet arrives: the noise floor, how busy the channel is, how many
receptions fail CRC. The radio measures all three and was never asked.

This samples the companion's own counters along the GPS track and stores
them, so the server can render a noise-floor map, a channel-utilisation
map and a CRC-error-rate map. A fixed observer cannot produce those — it
measures one point forever.

**Zero airtime:** `CMD_GET_STATS` is a local Bluetooth query to the
attached radio. Nothing is transmitted.

## Design points worth knowing

- **Absolutes are stored; deltas are derived at query time.** A lost or
reordered sample then costs one interval rather than corrupting a
running total. `ClientRfDeltas` breaks the chain whenever `uptime_secs`
fails to increase — that is the reboot and counter-wrap detector.
- **Absent is not zero, end to end.** Firmware predating the
`recv_errors` field cannot count CRC errors at all, and a stored `0`
would read downstream as "a perfectly clean channel" — the opposite of
"we don't know". Presence/absence is preserved through the app parser,
the wire payload, a nullable column, and the delta view, which returns
`nil` rather than `0` when either endpoint is unknown. Each of those
five layers has its own test.
- **`sampled_at` is millisecond precision, and it is load-bearing.**
SQLite compares these strings lexicographically and `.` (0x2E) sorts
before `Z` (0x5A), so a second-resolution retention cutoff would delete
rows *inside* the window. The prune formats its cutoff with the same
layout.

## Performance justification (touches the ingest hot path)

- One INSERT per sample, gated behind an opt-in flag that defaults off.
Sample rate is 15 s while moving and 5 min while parked, so roughly 240
rows per hour per active driver.
- The delta query is a single `LAG(...) OVER` pass with no nested query
inside the loop, so it cannot deadlock the single writer connection.
Window functions are already used elsewhere in this codebase.
- Retention has its own key and index (`sampled_at`); without it the
table would grow unbounded, so `config.example.json` documents it
inline.

## Safety for existing deployments

Opt-in and default off on both sides (`clientRfSamples.enabled`, and
`rfSampler` in the app). The coverage path is untouched —
`Publisher.buildPayload` is byte-identical and a record with no `kind`
field still routes to `/packets` unchanged.

The MQTT dispatch was reshaped so that **anything on `meshcore/client/…`
returns from that branch in every config state**, with the enable-gates
inside rather than in the topic match. Previously a disabled gate let
the message fall through to the observer path, where `parts[1]` — the
literal string `client` — was read as a region and the phone's pubkey
registered as an observer. The blacklist check now also runs ahead of
the sub-topic switch, so it covers every present and future client
sub-topic.

## Testing

Full ingestor suite green. Notable coverage: a `/rf` message with the
gate off writes nothing anywhere and does not fall through; a sample
missing `uptime_secs` is rejected rather than stored as an unusable row;
two samples 40 ms apart remain two rows; and the retention test seeds a
row with a non-zero millisecond component inside the cutoff second,
which is the only row that distinguishes a correct cutoff from an
RFC3339 one.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

cmd/ingestor/client_reception.go

@@ -250,6 +250,50 @@ func (s *Store) InsertClientReception(r *ClientReception) (bool, error) {
 	return n > 0, nil
 }
 
+// ClientRfSample is one RF environment sample from a mobile client: the
+// attached LoRa radio's own counters (noise floor, RX/TX airtime, CRC errors,
+// packet totals), paired with the GPS point where the sample was taken. Every
+// counter is an absolute cumulative value the radio reported; Task 5 derives
+// deltas at query time. RecvErrors stays nil on firmware that cannot count
+// CRC errors — never storing 0, which would read as a perfectly clean channel.
+// Errors is the exception: per the firmware stats frame, it is an error-flags
+// bitmask, not a counter, so it is deliberately excluded from ClientRfDeltas.
+type ClientRfSample struct {
+	RxPubkey, SampledAt, IngestedAt                                                    string
+	Lat, Lon                                                                           float64
+	PosAccM                                                                            *float64
+	Stationary                                                                         bool
+	UptimeSecs                                                                         int64
+	BatteryMV, QueueLen, Errors                                                        *int
+	NoiseFloor, LastRSSI                                                               *int
+	LastSNR                                                                            *float64
+	TxAirSecs, RxAirSecs, Recv, Sent, FloodRx, DirectRx, FloodTx, DirectTx, RecvErrors *int64
+}
+
+// InsertClientRfSample writes one RF environment sample. Idempotent via
+// UNIQUE(rx_pubkey, sampled_at). Every counter is stored as an absolute; the
+// server derives deltas, so a lost or reordered sample costs one interval
+// rather than corrupting a running total.
+func (s *Store) InsertClientRfSample(o *ClientRfSample) (bool, error) {
+	res, err := s.db.Exec(`
+		INSERT INTO client_rf_samples
+			(rx_pubkey, sampled_at, ingested_at, lat, lon, pos_acc_m, stationary,
+			 uptime_secs, battery_mv, queue_len, errors, noise_floor, last_rssi,
+			 last_snr, tx_air_secs, rx_air_secs, recv, sent, flood_rx, direct_rx,
+			 flood_tx, direct_tx, recv_errors)
+		VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+		ON CONFLICT(rx_pubkey, sampled_at) DO NOTHING`,
+		o.RxPubkey, o.SampledAt, o.IngestedAt, o.Lat, o.Lon, o.PosAccM, boolToInt(o.Stationary),
+		o.UptimeSecs, o.BatteryMV, o.QueueLen, o.Errors, o.NoiseFloor, o.LastRSSI,
+		o.LastSNR, o.TxAirSecs, o.RxAirSecs, o.Recv, o.Sent, o.FloodRx, o.DirectRx,
+		o.FloodTx, o.DirectTx, o.RecvErrors)
+	if err != nil {
+		return false, err
+	}
+	n, _ := res.RowsAffected()
+	return n > 0, nil
+}
+
 // ClientRxObservation is one decodable packet a mobile client heard, whether or
 // not it was attributable to a directly-heard node. Diagnostic only: this table
 // is never a coverage source (see client_receptions for that).

cmd/ingestor/client_reception_test.go

@@ -76,11 +76,22 @@ func TestPruneClientRxObservationsUsesIndex(t *testing.T) {
 // TestPruneOldClientRxObservations verifies the diagnostic observations reaper
 // deletes rows older than the window and keeps recent ones, and that days=0
 // disables it — mirroring TestPruneOldClientReceptions for the sibling table.
+// It also covers the millisecond-boundary case an RFC3339 (no-millisecond)
+// cutoff would get wrong, mirroring TestPruneOldClientRfSamples: rx_at is
+// stored via rxTimeMillisLayout, and a row 500ms after the cutoff instant —
+// chronologically newer, so it must survive — has a string form like
+// "...T10:00:00.500Z" that sorts lexicographically BEFORE a bare
+// "...T10:00:00Z" cutoff (since '.' is 0x2E and 'Z' is 0x5A). An
+// RFC3339-formatted cutoff would therefore wrongly delete it; a
+// millisecond-formatted cutoff (what PruneOldClientRxObservations uses today)
+// correctly keeps it.
 func TestPruneOldClientRxObservations(t *testing.T) {
 	s := newTestStore(t)
 	now := time.Now().UTC()
 	recent := now.AddDate(0, 0, -1).Format(rxTimeMillisLayout)
 	old := now.AddDate(0, 0, -40).Format(rxTimeMillisLayout)
+	cutoffInstant := now.AddDate(0, 0, -7)
+	boundary := cutoffInstant.Add(500 * time.Millisecond).Format(rxTimeMillisLayout)
 
 	mk := func(rxAt, pktHash string) *ClientRxObservation {
 		return &ClientRxObservation{
@@ -94,6 +105,9 @@ func TestPruneOldClientRxObservations(t *testing.T) {
 	if _, err := s.InsertClientRxObservation(mk(old, "hash-old")); err != nil {
 		t.Fatal(err)
 	}
+	if _, err := s.InsertClientRxObservation(mk(boundary, "hash-boundary")); err != nil {
+		t.Fatal(err)
+	}
 
 	if n, _ := s.PruneOldClientRxObservations(0); n != 0 {
 		t.Fatalf("days=0 must be a no-op, got %d", n)
@@ -107,13 +121,18 @@ func TestPruneOldClientRxObservations(t *testing.T) {
 	}
 	var remaining int
 	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations`).Scan(&remaining)
-	if remaining != 1 {
-		t.Fatalf("expected 1 observation remaining (recent), got %d", remaining)
+	if remaining != 2 {
+		t.Fatalf("expected 2 observations remaining (recent + boundary), got %d", remaining)
 	}
-	var remainingHash string
-	s.db.QueryRow(`SELECT pkt_hash FROM client_rx_observations`).Scan(&remainingHash)
-	if remainingHash != "hash-recent" {
-		t.Fatalf("expected the recent row to survive, got pkt_hash=%q", remainingHash)
+	var recentSurvived int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations WHERE pkt_hash = ?`, "hash-recent").Scan(&recentSurvived)
+	if recentSurvived != 1 {
+		t.Fatalf("the recent row must survive the prune; got %d", recentSurvived)
+	}
+	var boundarySurvived int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rx_observations WHERE pkt_hash = ?`, "hash-boundary").Scan(&boundarySurvived)
+	if boundarySurvived != 1 {
+		t.Fatalf("boundary row (500ms after cutoff instant) must survive; an RFC3339 (no-ms) cutoff would wrongly delete it because '.' < 'Z' lexicographically — got %d", boundarySurvived)
 	}
 }
 

cmd/ingestor/client_rf_sample.go

@@ -0,0 +1,189 @@
+package main
+
+import (
+	"log"
+	"strings"
+	"time"
+)
+
+// handleClientRfSample processes one RF environment sample from
+// meshcore/client/{PUBLIC_KEY}/rf. The topic pubkey is the identity (ACL-bound
+// by the broker); origin_id in the payload is informational and MUST NOT be
+// used as a fallback — that would defeat the trust model.
+func handleClientRfSample(store *Store, tag, rxPubkey string, msg map[string]interface{}) {
+	rxPubkey = strings.ToLower(strings.TrimSpace(rxPubkey))
+	if !clientPubkeyRe.MatchString(rxPubkey) {
+		log.Printf("MQTT [%s] rf: invalid pubkey %.8q, dropping", tag, rxPubkey)
+		return
+	}
+	gps, ok := msg["gps"].(map[string]interface{})
+	if !ok {
+		return // a sample without a position is not a map point
+	}
+	lat, latOK := toFloat64(gps["lat"])
+	lon, lonOK := toFloat64(gps["lon"])
+	if !latOK || !lonOK || lat < -90 || lat > 90 || lon < -180 || lon > 180 {
+		return
+	}
+	uptime, uptimeOK := toFloat64(msg["uptime_secs"])
+	if !uptimeOK {
+		// uptime_secs is the only reboot/wrap detector the delta query has. A
+		// missing value would silently default to 0, making every future sample
+		// look like uptime never increases — ClientRfDeltas would then return
+		// nothing forever while the table looks full. The app always sends this
+		// field, so absence means a malformed publisher; reject it the same way
+		// a missing GPS fix is rejected.
+		log.Printf("MQTT [%s] rf: missing uptime_secs, dropping", tag)
+		return
+	}
+
+	rxTime, _ := resolveRxTimeCore(msg, tag)
+	sampledAt := rxTime.Format(rxTimeMillisLayout)
+
+	o := &ClientRfSample{
+		RxPubkey:   rxPubkey,
+		SampledAt:  sampledAt,
+		IngestedAt: time.Now().UTC().Format(time.RFC3339),
+		Lat:        lat, Lon: lon,
+		UptimeSecs: int64(uptime),
+	}
+	if acc, ok := toFloat64(gps["acc_m"]); ok {
+		o.PosAccM = &acc
+	}
+	if v, ok := msg["stationary"].(bool); ok {
+		o.Stationary = v
+	}
+	// Optional numeric fields: absent stays NULL. recv_errors in particular must
+	// never default to 0 — that would read as a clean channel on firmware that
+	// does not report it.
+	o.BatteryMV = optInt(msg, "battery_mv")
+	o.QueueLen = optInt(msg, "queue_len")
+	o.Errors = optInt(msg, "errors")
+	o.NoiseFloor = optInt(msg, "noise_floor")
+	o.LastRSSI = optInt(msg, "last_rssi")
+	o.LastSNR = optFloat(msg, "last_snr")
+	o.TxAirSecs = optInt64(msg, "tx_air_secs")
+	o.RxAirSecs = optInt64(msg, "rx_air_secs")
+	o.Recv = optInt64(msg, "recv")
+	o.Sent = optInt64(msg, "sent")
+	o.FloodRx = optInt64(msg, "flood_rx")
+	o.DirectRx = optInt64(msg, "direct_rx")
+	o.FloodTx = optInt64(msg, "flood_tx")
+	o.DirectTx = optInt64(msg, "direct_tx")
+	o.RecvErrors = optInt64(msg, "recv_errors")
+
+	if _, err := store.InsertClientRfSample(o); err != nil {
+		log.Printf("MQTT [%s] rf sample insert: %v", tag, err)
+	}
+}
+
+// optInt / optInt64 / optFloat return nil when the key is absent, so a missing
+// counter becomes SQL NULL rather than a misleading zero.
+func optInt(m map[string]interface{}, k string) *int {
+	if f, ok := toFloat64(m[k]); ok {
+		v := int(f)
+		return &v
+	}
+	return nil
+}
+
+func optInt64(m map[string]interface{}, k string) *int64 {
+	if f, ok := toFloat64(m[k]); ok {
+		v := int64(f)
+		return &v
+	}
+	return nil
+}
+
+func optFloat(m map[string]interface{}, k string) *float64 {
+	if f, ok := toFloat64(m[k]); ok {
+		return &f
+	}
+	return nil
+}
+
+// ClientRfDelta is one consecutive pair of samples from the same radio. Each
+// *Delta field is nil when either endpoint of the pair is NULL (the counter
+// is unsupported on that firmware) — a plain int64 defaulting to 0 there
+// would be indistinguishable from a measured zero, the exact conflation the
+// NULL-vs-zero rule on the underlying columns exists to prevent.
+type ClientRfDelta struct {
+	At           string
+	Lat, Lon     float64
+	Stationary   bool
+	WallMillis   int64
+	RxAirDelta   *int64
+	TxAirDelta   *int64
+	RecvDelta    *int64
+	RecvErrDelta *int64
+}
+
+// ClientRfDeltas returns consecutive-sample deltas for one radio. Pairs where
+// uptime_secs did not increase are skipped: that is a reboot (or a counter
+// wrap), and subtracting across it would produce a large negative or a bogus
+// spike. Absolutes stay in the table; only this view does arithmetic.
+func (s *Store) ClientRfDeltas(rxPubkey, from, to string) ([]ClientRfDelta, error) {
+	rxPubkey = strings.ToLower(strings.TrimSpace(rxPubkey))
+	rows, err := s.db.Query(`
+		SELECT sampled_at, lat, lon, stationary, uptime_secs,
+		       rx_air_secs, tx_air_secs, recv, recv_errors,
+		       LAG(sampled_at)  OVER w AS prev_at,
+		       LAG(uptime_secs) OVER w AS prev_uptime,
+		       LAG(rx_air_secs) OVER w AS prev_rx,
+		       LAG(tx_air_secs) OVER w AS prev_tx,
+		       LAG(recv)        OVER w AS prev_recv,
+		       LAG(recv_errors) OVER w AS prev_errs
+		FROM client_rf_samples
+		WHERE rx_pubkey = ? AND sampled_at >= ? AND sampled_at < ?
+		WINDOW w AS (ORDER BY sampled_at)
+		ORDER BY sampled_at`, rxPubkey, from, to)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []ClientRfDelta
+	for rows.Next() {
+		var at string
+		var lat, lon float64
+		var stationary int
+		var uptime int64
+		var rx, tx, recv, errs *int64
+		var prevAt *string
+		var prevUptime, prevRx, prevTx, prevRecv, prevErrs *int64
+		if err := rows.Scan(&at, &lat, &lon, &stationary, &uptime, &rx, &tx, &recv, &errs,
+			&prevAt, &prevUptime, &prevRx, &prevTx, &prevRecv, &prevErrs); err != nil {
+			log.Printf("client_rf_samples delta scan: %v", err)
+			continue
+		}
+		if prevAt == nil || prevUptime == nil || uptime <= *prevUptime {
+			continue // first row of the window, or a reboot/wrap — no delta
+		}
+		t1, err1 := time.Parse(time.RFC3339, *prevAt)
+		t2, err2 := time.Parse(time.RFC3339, at)
+		if err1 != nil || err2 != nil {
+			continue
+		}
+		d := ClientRfDelta{
+			At: at, Lat: lat, Lon: lon, Stationary: stationary == 1,
+			WallMillis: t2.Sub(t1).Milliseconds(),
+		}
+		d.RxAirDelta = deltaOf(rx, prevRx)
+		d.TxAirDelta = deltaOf(tx, prevTx)
+		d.RecvDelta = deltaOf(recv, prevRecv)
+		d.RecvErrDelta = deltaOf(errs, prevErrs)
+		out = append(out, d)
+	}
+	return out, rows.Err()
+}
+
+func deltaOf(cur, prev *int64) *int64 {
+	if cur == nil || prev == nil {
+		return nil
+	}
+	d := *cur - *prev
+	if d < 0 { // a wrapped individual counter even though uptime_secs looked monotonic
+		d = 0
+	}
+	return &d
+}

cmd/ingestor/client_rf_sample_test.go

@@ -0,0 +1,169 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+func TestHandleClientRfSample(t *testing.T) {
+	s := newTestStore(t)
+	msg := map[string]interface{}{
+		"type": "RF_SAMPLE", "timestamp": "2026-08-17T10:00:00.000Z",
+		"gps":         map[string]interface{}{"lat": 51.2, "lon": 4.4, "acc_m": 8.0},
+		"stationary":  false,
+		"uptime_secs": 84213.0, "noise_floor": -119.0, "rx_air_secs": 20877.0,
+	}
+	handleClientRfSample(s, "test", "aa11", msg)
+
+	var n int
+	if err := s.db.QueryRow(`SELECT COUNT(*) FROM client_rf_samples`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("rows = %d, want 1", n)
+	}
+
+	var sampledAt string
+	var recvErrorsNull bool
+	if err := s.db.QueryRow(`SELECT sampled_at, recv_errors IS NULL FROM client_rf_samples`).Scan(&sampledAt, &recvErrorsNull); err != nil {
+		t.Fatal(err)
+	}
+	// sampled_at must be stored at rxTimeMillisLayout precision, not
+	// resolveRxTime's second-resolution RFC3339 — Task 6's prune cutoff
+	// compares lexicographically against this string.
+	if _, err := time.Parse(rxTimeMillisLayout, sampledAt); err != nil {
+		t.Errorf("sampled_at = %q, want rxTimeMillisLayout format: %v", sampledAt, err)
+	}
+	// The payload omits recv_errors; it must stay NULL, never default to 0
+	// (which would read as a clean channel on firmware that can't count CRC
+	// errors).
+	if !recvErrorsNull {
+		t.Error("recv_errors should be NULL when omitted from the payload")
+	}
+}
+
+func TestHandleClientRfSampleRejects(t *testing.T) {
+	s := newTestStore(t)
+	base := func() map[string]interface{} {
+		return map[string]interface{}{
+			"timestamp":   "2026-08-17T10:00:00.000Z",
+			"gps":         map[string]interface{}{"lat": 51.2, "lon": 4.4},
+			"uptime_secs": 1.0,
+		}
+	}
+
+	handleClientRfSample(s, "test", "NOT-HEX!", base()) // bad topic pubkey
+
+	noGPS := base()
+	delete(noGPS, "gps")
+	handleClientRfSample(s, "test", "aa11", noGPS)
+
+	badGPS := base()
+	badGPS["gps"] = map[string]interface{}{"lat": 999.0, "lon": 4.4}
+	handleClientRfSample(s, "test", "aa11", badGPS)
+
+	noUptime := base()
+	delete(noUptime, "uptime_secs")
+	handleClientRfSample(s, "test", "aa11", noUptime)
+
+	var n int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rf_samples`).Scan(&n)
+	if n != 0 {
+		t.Errorf("rows = %d, want 0 — all four must be rejected", n)
+	}
+}
+
+// TestRfDeltaBreaksOnReboot seeds the first two samples through
+// handleClientRfSample (not a direct InsertClientRfSample) so the
+// handler-written rxTimeMillisLayout format is what ClientRfDeltas actually
+// parses back — a regression that reverted the handler to second-resolution
+// resolveRxTime would leave a direct-insert-seeded test green while breaking
+// this round trip for real.
+func TestRfDeltaBreaksOnReboot(t *testing.T) {
+	s := newTestStore(t)
+	seed := func(ts string, uptime, rxAir float64) {
+		handleClientRfSample(s, "test", "aa11", map[string]interface{}{
+			"timestamp":   ts,
+			"gps":         map[string]interface{}{"lat": 51.2, "lon": 4.4},
+			"uptime_secs": uptime,
+			"rx_air_secs": rxAir,
+		})
+	}
+	seed("2026-08-17T10:00:00.000Z", 1000, 500)
+	seed("2026-08-17T10:00:15.000Z", 1015, 512) // +12 s of RX air over 15 s
+	seed("2026-08-17T10:00:30.000Z", 10, 3)     // rebooted: uptime dropped
+
+	// Bounds are compared lexicographically against millisecond-precision
+	// sampled_at values, so they must themselves be in the millisecond layout
+	// — "10:00:00Z" would lexicographically exclude "10:00:00.000Z" (`.` < `Z`).
+	deltas, err := s.ClientRfDeltas("aa11", "2026-08-17T00:00:00.000Z", "2026-08-18T00:00:00.000Z")
+	if err != nil {
+		t.Fatalf("deltas: %v", err)
+	}
+	if len(deltas) != 1 {
+		t.Fatalf("deltas = %d, want 1 (the reboot pair must be skipped)", len(deltas))
+	}
+	if deltas[0].RxAirDelta == nil || *deltas[0].RxAirDelta != 12 || deltas[0].WallMillis != 15000 {
+		t.Errorf("delta = %v over %dms, want 12 over 15000ms", deltas[0].RxAirDelta, deltas[0].WallMillis)
+	}
+}
+
+// TestRfDeltaNilWhenEitherEndpointMissingCounter proves the NULL-vs-zero
+// distinction survives into the delta view: a counter unsupported by
+// firmware (NULL on either endpoint) must produce a nil *int64 delta, never
+// a 0 that a CRC-error-rate map would render as "clean channel".
+func TestRfDeltaNilWhenEitherEndpointMissingCounter(t *testing.T) {
+	s := newTestStore(t)
+	seed := func(ts string, uptime float64, recvErrors *int64) {
+		msg := map[string]interface{}{
+			"timestamp":   ts,
+			"gps":         map[string]interface{}{"lat": 51.2, "lon": 4.4},
+			"uptime_secs": uptime,
+		}
+		if recvErrors != nil {
+			msg["recv_errors"] = float64(*recvErrors)
+		}
+		handleClientRfSample(s, "test", "aa11", msg)
+	}
+	five := int64(5)
+	seed("2026-08-17T10:00:00.000Z", 1000, nil) // firmware without the counter
+	seed("2026-08-17T10:00:15.000Z", 1015, &five)
+
+	deltas, err := s.ClientRfDeltas("aa11", "2026-08-17T00:00:00.000Z", "2026-08-18T00:00:00.000Z")
+	if err != nil {
+		t.Fatalf("deltas: %v", err)
+	}
+	if len(deltas) != 1 {
+		t.Fatalf("deltas = %d, want 1", len(deltas))
+	}
+	if deltas[0].RecvErrDelta != nil {
+		t.Errorf("RecvErrDelta = %v, want nil (prev recv_errors was NULL)", *deltas[0].RecvErrDelta)
+	}
+}
+
+// TestClientRfDeltasNormalizesPubkeyCase proves ClientRfDeltas matches the
+// stored lowercase rx_pubkey even when a caller passes uppercase hex —
+// mirroring the strings.ToLower(rx) normalization in
+// cmd/server/rx_dashboard.go's queryCoverageFiltered. Without it, a URL path
+// pubkey pasted in uppercase silently returns zero rows.
+func TestClientRfDeltasNormalizesPubkeyCase(t *testing.T) {
+	s := newTestStore(t)
+	seed := func(ts string, uptime, rxAir float64) {
+		handleClientRfSample(s, "test", "aa11", map[string]interface{}{
+			"timestamp":   ts,
+			"gps":         map[string]interface{}{"lat": 51.2, "lon": 4.4},
+			"uptime_secs": uptime,
+			"rx_air_secs": rxAir,
+		})
+	}
+	seed("2026-08-17T10:00:00.000Z", 1000, 500)
+	seed("2026-08-17T10:00:15.000Z", 1015, 512)
+
+	deltas, err := s.ClientRfDeltas("AA11", "2026-08-17T00:00:00.000Z", "2026-08-18T00:00:00.000Z")
+	if err != nil {
+		t.Fatalf("deltas: %v", err)
+	}
+	if len(deltas) != 1 {
+		t.Fatalf("deltas = %d, want 1 for uppercase pubkey lookup", len(deltas))
+	}
+}

cmd/ingestor/config.go

@@ -57,6 +57,7 @@ type Config struct {
 	Runtime              *RuntimeConfig              `json:"runtime,omitempty"`
 	ClientRxCoverage     *ClientRxCoverageConfig     `json:"clientRxCoverage,omitempty"`
 	ClientRxObservations *ClientRxObservationsConfig `json:"clientRxObservations,omitempty"`
+	ClientRfSamples      *ClientRfSamplesConfig      `json:"clientRfSamples,omitempty"`
 	GeoFilter            *GeoFilterConfig            `json:"geo_filter,omitempty"`
 	// PathTrust configures the minimum path-hash prefix length trusted as
 	// mapping/topology evidence (issue #1784).
@@ -162,6 +163,16 @@ func (c *Config) ClientRxObservationsEnabled() bool {
 	return c.ClientRxObservations != nil && c.ClientRxObservations.Enabled
 }
 
+// ClientRfSamplesConfig controls the opt-in RF environment sample stream.
+type ClientRfSamplesConfig struct {
+	Enabled bool `json:"enabled"`
+}
+
+// ClientRfSamplesEnabled reports whether RF samples are recorded. Default false.
+func (c *Config) ClientRfSamplesEnabled() bool {
+	return c.ClientRfSamples != nil && c.ClientRfSamples.Enabled
+}
+
 // RetentionConfig controls how long stale nodes are kept before being moved to inactive_nodes.
 type RetentionConfig struct {
 	NodeDays     int `json:"nodeDays"`
@@ -184,6 +195,10 @@ type RetentionConfig struct {
 	// client_rx_observations table; 0 disables. Shorter than clientRxDays —
 	// this table is diagnostic, not archival.
 	ClientRxObsDays int `json:"clientRxObsDays"`
+	// ClientRfDays is the retention window (by sampled_at) for the
+	// client_rf_samples table; 0 disables. Bounds the table the opt-in RF
+	// sample stream would otherwise grow without limit.
+	ClientRfDays int `json:"clientRfDays"`
 }
 
 // PacketDaysOrZero returns the configured retention.packetDays or 0
@@ -221,6 +236,14 @@ func (c *Config) ClientRxObsDaysOrZero() int {
 	return 0
 }
 
+// ClientRfDaysOrZero returns retention.clientRfDays or 0 (disabled).
+func (c *Config) ClientRfDaysOrZero() int {
+	if c.Retention != nil && c.Retention.ClientRfDays > 0 {
+		return c.Retention.ClientRfDays
+	}
+	return 0
+}
+
 // MetricsConfig controls observer metrics collection.
 type MetricsConfig struct {
 	SampleIntervalSec int `json:"sampleIntervalSec"`

cmd/ingestor/db.go

@@ -377,6 +377,39 @@ func applySchema(db *sql.DB) error {
 		CREATE INDEX IF NOT EXISTS idx_cro_hash      ON client_rx_observations(pkt_hash, rx_at);
 		CREATE INDEX IF NOT EXISTS idx_cro_forwarder ON client_rx_observations(forwarder, rx_at);
 		CREATE INDEX IF NOT EXISTS idx_cro_scope     ON client_rx_observations(scope_name, rx_at);
+
+		-- RF environment samples from mobile clients: radio counters paired with
+		-- a GPS point. Absolutes only — deltas are computed at query time, and a
+		-- decrease in uptime_secs (reboot) or any counter (wrap) breaks the chain.
+		CREATE TABLE IF NOT EXISTS client_rf_samples (
+			id           INTEGER PRIMARY KEY AUTOINCREMENT,
+			rx_pubkey    TEXT NOT NULL,
+			sampled_at   TEXT NOT NULL,
+			ingested_at  TEXT NOT NULL,
+			lat          REAL NOT NULL,
+			lon          REAL NOT NULL,
+			pos_acc_m    REAL,
+			stationary   INTEGER NOT NULL DEFAULT 0,
+			uptime_secs  INTEGER NOT NULL,
+			battery_mv   INTEGER,
+			queue_len    INTEGER,
+			errors       INTEGER,
+			noise_floor  INTEGER,
+			last_rssi    INTEGER,
+			last_snr     REAL,
+			tx_air_secs  INTEGER,
+			rx_air_secs  INTEGER,
+			recv         INTEGER,
+			sent         INTEGER,
+			flood_rx     INTEGER,
+			direct_rx    INTEGER,
+			flood_tx     INTEGER,
+			direct_tx    INTEGER,
+			recv_errors  INTEGER,
+			UNIQUE(rx_pubkey, sampled_at)
+		);
+		CREATE INDEX IF NOT EXISTS idx_crf_prune ON client_rf_samples(sampled_at);
+		CREATE INDEX IF NOT EXISTS idx_crf_track ON client_rf_samples(rx_pubkey, sampled_at);
 	`
 	if _, err := db.Exec(schema); err != nil {
 		return fmt.Errorf("base schema: %w", err)
@@ -1965,6 +1998,14 @@ func nilIfEmpty(s string) interface{} {
 	return s
 }
 
+// boolToInt converts a bool to SQLite's 0/1 INTEGER representation.
+func boolToInt(b bool) int {
+	if b {
+		return 1
+	}
+	return 0
+}
+
 // scopeNameForDB encodes PacketData scope semantics for DB storage:
 // non-transport-scoped → nil (SQL NULL); transport-scoped → pointer to ScopeName
 // (may be "" for unknown region, "#name" for matched region).

cmd/ingestor/db_test.go

@@ -3285,3 +3285,215 @@ func TestInsertClientRxObservation(t *testing.T) {
 // failure mode isn't reachable from a black-box test of applySchema without
 // contorting production code. The fix itself (db.go, transport_codes_v1
 // block) still stands, matching the observers_clock_naive_v1 pattern.
+
+func TestInsertClientRfSample(t *testing.T) {
+	s := newTestStore(t)
+	posAcc := 12.5
+	snr := -6.75
+	errs := 7
+	nf := -119
+	rssi := -92
+	battery := 4021
+	queueLen := 3
+	txAir := int64(341)
+	rxAir := int64(20877)
+	recv := int64(18422)
+	sent := int64(290)
+	floodRx := int64(17110)
+	directRx := int64(1312)
+	floodTx := int64(212)
+	directTx := int64(78)
+	o := &ClientRfSample{
+		RxPubkey: "aa11", SampledAt: "2026-08-17T10:00:00.000Z", IngestedAt: "2026-08-17T10:00:01Z",
+		Lat: 51.2, Lon: 4.4, PosAccM: &posAcc, Stationary: true, UptimeSecs: 84213,
+		BatteryMV: &battery, QueueLen: &queueLen, Errors: &errs, NoiseFloor: &nf, LastRSSI: &rssi,
+		LastSNR: &snr, TxAirSecs: &txAir, RxAirSecs: &rxAir, Recv: &recv, Sent: &sent,
+		FloodRx: &floodRx, DirectRx: &directRx, FloodTx: &floodTx, DirectTx: &directTx,
+	}
+	ins, err := s.InsertClientRfSample(o)
+	if err != nil || !ins {
+		t.Fatalf("insert: ins=%v err=%v", ins, err)
+	}
+	ins, err = s.InsertClientRfSample(o)
+	if err != nil || ins {
+		t.Fatalf("duplicate must be idempotent: ins=%v err=%v", ins, err)
+	}
+
+	// Every column gets a distinct, non-symmetric value and is read back, so a
+	// positional transposition between any two same-typed neighbours in the
+	// 23-column INSERT (e.g. flood_rx/direct_rx, or recv/sent) is detectable —
+	// row-counting alone would pass either way.
+	var gotLat, gotLon, gotPosAccM, gotLastSNR float64
+	var gotStationary, gotErrors, gotNoiseFloor, gotLastRSSI, gotBatteryMV, gotQueueLen int
+	var gotUptimeSecs, gotTxAirSecs, gotRxAirSecs, gotRecv, gotSent, gotFloodRx, gotDirectRx, gotFloodTx, gotDirectTx int64
+	var gotRecvErrors *int64
+	if err := s.db.QueryRow(`
+		SELECT lat, lon, pos_acc_m, stationary, uptime_secs, battery_mv, queue_len,
+		       errors, noise_floor, last_rssi, last_snr, tx_air_secs, rx_air_secs,
+		       recv, sent, flood_rx, direct_rx, flood_tx, direct_tx, recv_errors
+		FROM client_rf_samples WHERE sampled_at = ?`, o.SampledAt).Scan(
+		&gotLat, &gotLon, &gotPosAccM, &gotStationary, &gotUptimeSecs, &gotBatteryMV, &gotQueueLen,
+		&gotErrors, &gotNoiseFloor, &gotLastRSSI, &gotLastSNR, &gotTxAirSecs, &gotRxAirSecs,
+		&gotRecv, &gotSent, &gotFloodRx, &gotDirectRx, &gotFloodTx, &gotDirectTx, &gotRecvErrors); err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if gotLat != 51.2 || gotLon != 4.4 {
+		t.Errorf("lat/lon = %v/%v, want 51.2/4.4 (a lat<->lon swap would pass with symmetric values)", gotLat, gotLon)
+	}
+	if gotPosAccM != 12.5 {
+		t.Errorf("pos_acc_m = %v, want 12.5", gotPosAccM)
+	}
+	if gotStationary != 1 {
+		t.Errorf("stationary = %d, want 1", gotStationary)
+	}
+	if gotUptimeSecs != 84213 {
+		t.Errorf("uptime_secs = %d, want 84213", gotUptimeSecs)
+	}
+	if gotBatteryMV != 4021 {
+		t.Errorf("battery_mv = %d, want 4021", gotBatteryMV)
+	}
+	if gotQueueLen != 3 {
+		t.Errorf("queue_len = %d, want 3", gotQueueLen)
+	}
+	if gotErrors != 7 {
+		t.Errorf("errors = %d, want 7", gotErrors)
+	}
+	if gotNoiseFloor != -119 {
+		t.Errorf("noise_floor = %d, want -119", gotNoiseFloor)
+	}
+	if gotLastRSSI != -92 {
+		t.Errorf("last_rssi = %d, want -92", gotLastRSSI)
+	}
+	if gotLastSNR != -6.75 {
+		t.Errorf("last_snr = %v, want -6.75", gotLastSNR)
+	}
+	if gotTxAirSecs != 341 {
+		t.Errorf("tx_air_secs = %d, want 341", gotTxAirSecs)
+	}
+	if gotRxAirSecs != 20877 {
+		t.Errorf("rx_air_secs = %d, want 20877", gotRxAirSecs)
+	}
+	if gotRecv != 18422 {
+		t.Errorf("recv = %d, want 18422", gotRecv)
+	}
+	if gotSent != 290 {
+		t.Errorf("sent = %d, want 290", gotSent)
+	}
+	if gotFloodRx != 17110 {
+		t.Errorf("flood_rx = %d, want 17110", gotFloodRx)
+	}
+	if gotDirectRx != 1312 {
+		t.Errorf("direct_rx = %d, want 1312", gotDirectRx)
+	}
+	if gotFloodTx != 212 {
+		t.Errorf("flood_tx = %d, want 212", gotFloodTx)
+	}
+	if gotDirectTx != 78 {
+		t.Errorf("direct_tx = %d, want 78", gotDirectTx)
+	}
+	if gotRecvErrors != nil {
+		t.Errorf("recv_errors = %v, want NULL on firmware that cannot count them", gotRecvErrors)
+	}
+
+	// A second sample, at a different sampled_at, proves recv_errors round-trips
+	// non-NULL too — the first sample only covers the NULL direction.
+	re := int64(55)
+	o2 := &ClientRfSample{
+		RxPubkey: "aa11", SampledAt: "2026-08-17T10:00:15.000Z", IngestedAt: "2026-08-17T10:00:16Z",
+		Lat: 51.2, Lon: 4.4, Stationary: true, UptimeSecs: 84228, RecvErrors: &re,
+	}
+	if ins, err = s.InsertClientRfSample(o2); err != nil || !ins {
+		t.Fatalf("insert second sample: ins=%v err=%v", ins, err)
+	}
+	var gotRecvErrors2 *int64
+	if err := s.db.QueryRow(`SELECT recv_errors FROM client_rf_samples WHERE sampled_at = ?`, o2.SampledAt).
+		Scan(&gotRecvErrors2); err != nil {
+		t.Fatalf("read second sample: %v", err)
+	}
+	if gotRecvErrors2 == nil || *gotRecvErrors2 != 55 {
+		t.Errorf("recv_errors = %v, want 55", gotRecvErrors2)
+	}
+}
+
+// TestPruneClientRfSamplesUsesIndex verifies the RF-sample reaper's
+// DELETE ... WHERE sampled_at < ? seeks idx_crf_prune rather than
+// full-scanning under the writer lock, mirroring
+// TestPruneClientRxObservationsUsesIndex for the sibling table. May pass
+// immediately since idx_crf_prune already exists (Task 4) — in that case
+// this is a regression guard.
+func TestPruneClientRfSamplesUsesIndex(t *testing.T) {
+	s := newTestStore(t)
+	rows, err := s.db.Query(`EXPLAIN QUERY PLAN DELETE FROM client_rf_samples WHERE sampled_at < ?`, "2026-01-01T00:00:00Z")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	plan := ""
+	for rows.Next() {
+		var id, parent, notused int
+		var detail string
+		if err := rows.Scan(&id, &parent, &notused, &detail); err != nil {
+			t.Fatal(err)
+		}
+		plan += detail + "\n"
+	}
+	if !strings.Contains(plan, "idx_crf_prune") {
+		t.Fatalf("retention DELETE should use idx_crf_prune, plan was:\n%s", plan)
+	}
+}
+
+// TestPruneOldClientRfSamples verifies the RF-sample reaper deletes rows
+// older than the window and keeps recent ones, and that days=0 disables it —
+// mirroring TestPruneOldClientRxObservations for the sibling table. It also
+// covers the case an RFC3339 (no-millisecond) cutoff gets wrong: sampled_at
+// is stored via rxTimeMillisLayout, and a row sampled 500ms after the cutoff
+// instant — chronologically newer, so it must survive — has a string form
+// like "...T10:00:00.500Z" that sorts lexicographically BEFORE a bare
+// "...T10:00:00Z" cutoff (since '.' is 0x2E and 'Z' is 0x5A). An
+// RFC3339-formatted cutoff would therefore wrongly delete it; a
+// millisecond-formatted cutoff correctly keeps it.
+func TestPruneOldClientRfSamples(t *testing.T) {
+	s := newTestStore(t)
+	now := time.Now().UTC()
+	recent := now.AddDate(0, 0, -1).Format(rxTimeMillisLayout)
+	old := now.AddDate(0, 0, -40).Format(rxTimeMillisLayout)
+	cutoffInstant := now.AddDate(0, 0, -7)
+	boundary := cutoffInstant.Add(500 * time.Millisecond).Format(rxTimeMillisLayout)
+
+	mk := func(sampledAt string) *ClientRfSample {
+		return &ClientRfSample{
+			RxPubkey: "aa11", SampledAt: sampledAt, IngestedAt: sampledAt,
+			Lat: 51.2, Lon: 4.4, UptimeSecs: 100,
+		}
+	}
+	if _, err := s.InsertClientRfSample(mk(recent)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.InsertClientRfSample(mk(old)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.InsertClientRfSample(mk(boundary)); err != nil {
+		t.Fatal(err)
+	}
+
+	if n, _ := s.PruneOldClientRfSamples(0); n != 0 {
+		t.Fatalf("days=0 must be a no-op, got %d", n)
+	}
+	n, err := s.PruneOldClientRfSamples(7)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Fatalf("expected 1 old sample pruned, got %d", n)
+	}
+	var remaining int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rf_samples`).Scan(&remaining)
+	if remaining != 2 {
+		t.Fatalf("expected 2 samples remaining (recent + boundary), got %d", remaining)
+	}
+	var boundarySurvived int
+	s.db.QueryRow(`SELECT COUNT(*) FROM client_rf_samples WHERE sampled_at = ?`, boundary).Scan(&boundarySurvived)
+	if boundarySurvived != 1 {
+		t.Fatalf("boundary row (500ms after cutoff instant) must survive; an RFC3339 (no-ms) cutoff would wrongly delete it because '.' < 'Z' lexicographically — got %d", boundarySurvived)
+	}
+}

cmd/ingestor/main.go

@@ -302,6 +302,17 @@ func main() {
 		}
 	}
 
+	// Diagnostic client_rf_samples retention: separate window, independent
+	// of clientRxDays/clientRxObsDays. 0 = disabled.
+	clientRfDays := cfg.ClientRfDaysOrZero()
+	if clientRfDays > 0 {
+		if n, err := store.PruneOldClientRfSamples(clientRfDays); err != nil {
+			log.Printf("[prune] client_rf_samples: %v", err)
+		} else if n > 0 {
+			log.Printf("[prune] startup pruned %d client_rf_samples older than %d days", n, clientRfDays)
+		}
+	}
+
 	vacuumPages := cfg.IncrementalVacuumPages()
 	store.RunIncrementalVacuum(vacuumPages)
 
@@ -367,10 +378,11 @@ func main() {
 	}
 
 	// Daily ticker for client-RX coverage retention (#1727), reused for the
-	// diagnostic client_rx_observations retention (Task 6) rather than
-	// starting a second ticker — the two flags are independent (0 disables
-	// each separately), so the ticker itself must run when either is set.
-	if clientRxDays > 0 || clientRxObsDays > 0 {
+	// diagnostic client_rx_observations and client_rf_samples retention
+	// (Task 6) rather than starting a second ticker — the three flags are
+	// independent (0 disables each separately), so the ticker itself must
+	// run when any is set.
+	if clientRxDays > 0 || clientRxObsDays > 0 || clientRfDays > 0 {
 		clientRxRetentionTicker := time.NewTicker(24 * time.Hour)
 		go func() {
 			for range clientRxRetentionTicker.C {
@@ -388,6 +400,13 @@ func main() {
 						store.RunIncrementalVacuum(vacuumPages)
 					}
 				}
+				if clientRfDays > 0 {
+					if n, err := store.PruneOldClientRfSamples(clientRfDays); err != nil {
+						log.Printf("[prune] client_rf_samples: %v", err)
+					} else if n > 0 {
+						store.RunIncrementalVacuum(vacuumPages)
+					}
+				}
 			}
 		}()
 		if clientRxDays > 0 {
@@ -396,6 +415,9 @@ func main() {
 		if clientRxObsDays > 0 {
 			log.Printf("[prune] auto-prune enabled: client_rx_observations older than %d days will be removed daily", clientRxObsDays)
 		}
+		if clientRfDays > 0 {
+			log.Printf("[prune] auto-prune enabled: client_rf_samples older than %d days will be removed daily", clientRfDays)
+		}
 	}
 
 	// Hourly WAL checkpoint to prevent unbounded WAL growth.
@@ -650,30 +672,39 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message,
 		return
 	}
 
-	// Mobile client RX coverage: dedicated topic meshcore/client/{PUBLIC_KEY}/packets.
-	// A roaming companion reports where it directly heard a node; handled in isolation
-	// from the observer/observations path. EMQX ACL binds parts[2] to the client's own key.
+	// Mobile client topics: meshcore/client/{PUBLIC_KEY}/packets (RX coverage)
+	// and meshcore/client/{PUBLIC_KEY}/rf (RF environment samples). A roaming
+	// companion reports where it directly heard a node, or its own radio's
+	// counters; both are handled in isolation from the observer/observations
+	// path. EMQX ACL binds parts[2] to the client's own key.
 	//
 	// The topic match and the enable-gate MUST be separate: matching on
 	// parts[1]=="client" always returns from this branch, whatever the config
-	// says. The gate only decides drop-vs-handle. Previously the gate sat
-	// inside the topic match, so a disabled gate made the whole condition
-	// false and fell through to the observer path below, where parts[1]
-	// ("client") would be taken as a region and the companion pubkey as an
-	// observer id — silently poisoning the region list, and worse now that
-	// fullRfLog multiplies client-topic volume.
+	// says. The gate only decides drop-vs-handle per sub-topic. Previously the
+	// gate sat inside the topic match, so a disabled gate made the whole
+	// condition false and fell through to the observer path below, where
+	// parts[1] ("client") would be taken as a region and the companion pubkey
+	// as an observer id — silently poisoning the region list, and worse now
+	// that fullRfLog multiplies client-topic volume.
 	if len(parts) >= 4 && parts[1] == "client" {
-		if !cfg.ClientRxCoverageEnabled() || parts[3] != "packets" {
-			return
-		}
 		// The observer blacklist (checked below on the observer path) only runs
 		// there, so a blacklisted operator could otherwise skirt it via the
-		// client topic (#1). Enforce it here before any coverage write.
+		// client topic (#1). Enforce it here before any client-topic write, for
+		// every sub-topic.
 		if cfg.IsObserverBlacklisted(parts[2]) {
 			log.Printf("MQTT [%s] client %.8s blacklisted, dropping", tag, parts[2])
 			return
 		}
-		handleClientPacket(store, cfg, tag, parts[2], msg, channelKeys, regionKeys)
+		switch parts[3] {
+		case "packets":
+			if cfg.ClientRxCoverageEnabled() {
+				handleClientPacket(store, cfg, tag, parts[2], msg, channelKeys, regionKeys)
+			}
+		case "rf":
+			if cfg.ClientRfSamplesEnabled() {
+				handleClientRfSample(store, tag, parts[2], msg)
+			}
+		}
 		return
 	}
 

cmd/ingestor/maintenance.go

@@ -104,6 +104,29 @@ func (s *Store) PruneOldClientRxObservations(days int) (int64, error) {
 	return n, nil
 }
 
+// PruneOldClientRfSamples deletes RF environment sample rows older than
+// `days` (by sampled_at). Diagnostic, not archival, so it gets its own
+// window, independent of clientRxDays/clientRxObsDays. 0 disables. Owned by
+// the ingestor writer (#1283). sampled_at on this table is stored at
+// millisecond precision via rxTimeMillisLayout, not RFC3339 — the cutoff
+// must be formatted the same way so lexicographic comparison against
+// stored values stays correct.
+func (s *Store) PruneOldClientRfSamples(days int) (int64, error) {
+	if days <= 0 {
+		return 0, nil
+	}
+	cutoff := time.Now().UTC().AddDate(0, 0, -days).Format(rxTimeMillisLayout)
+	res, err := s.db.Exec(`DELETE FROM client_rf_samples WHERE sampled_at < ?`, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("prune client_rf_samples: %w", err)
+	}
+	n, _ := res.RowsAffected()
+	if n > 0 {
+		log.Printf("[prune] deleted %d client_rf_samples older than %d days", n, days)
+	}
+	return n, nil
+}
+
 // SoftDeleteBlacklistedObservers marks observers in the blacklist as
 // inactive=1 so they are hidden from API responses. Owned by ingestor
 // per #1287. Runs once at startup.

cmd/ingestor/rf_gate_test.go

@@ -0,0 +1,136 @@
+package main
+
+import "testing"
+
+// clientRfMsg builds a valid RF environment sample message on the dedicated
+// topic meshcore/client/<pubkey>/rf.
+func clientRfMsg() *mockMessage {
+	payload := []byte(`{"timestamp":"2026-06-09T12:00:00Z","uptime_secs":100.0,"noise_floor":-119.0,"gps":{"lat":51.05,"lon":3.72,"acc_m":8.0}}`)
+	return &mockMessage{topic: "meshcore/client/" + testCompanionPK + "/rf", payload: payload}
+}
+
+// clientRfMsgWithRaw is clientRfMsg plus a "raw" field (same advert hex as
+// clientCoverageMsg). The real app never sends "raw" on /rf, but the
+// fall-through guard must be structural — independent of payload shape — so
+// this proves the dispatch itself (not payload luck) is what keeps a /rf
+// message with an incidental "raw" field from being decoded as an ordinary
+// observer packet were the gate ever to fall through.
+func clientRfMsgWithRaw() *mockMessage {
+	advertHex := "11451000D818206D3AAC152C8A91F89957E6D30CA51F36E28790228971C473B755F244F718754CF5EE4A2FD58D944466E42CDED140C66D0CC590183E32BAF40F112BE8F3F2BDF6012B4B2793C52F1D36F69EE054D9A05593286F78453E56C0EC4A3EB95DDA2A7543FCCC00B939CACC009278603902FC12BCF84B706120526F6F6620536F6C6172"
+	payload := []byte(`{"raw":"` + advertHex + `","direction":"rx","timestamp":"2026-06-09T12:00:00Z","origin":"MyMob","uptime_secs":100.0,"gps":{"lat":51.05,"lon":3.72,"acc_m":8.0}}`)
+	return &mockMessage{topic: "meshcore/client/" + testCompanionPK + "/rf", payload: payload}
+}
+
+func clientRfSampleCount(t *testing.T, s *Store) int {
+	t.Helper()
+	var n int
+	if err := s.db.QueryRow(`SELECT COUNT(*) FROM client_rf_samples`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+// TestClientRfSamplesEnabledDefault verifies the gate helper defaults OFF for
+// nil/absent config and is only true when explicitly enabled.
+func TestClientRfSamplesEnabledDefault(t *testing.T) {
+	if (&Config{}).ClientRfSamplesEnabled() {
+		t.Fatal("nil ClientRfSamples must report disabled")
+	}
+	if (&Config{ClientRfSamples: &ClientRfSamplesConfig{Enabled: false}}).ClientRfSamplesEnabled() {
+		t.Fatal("Enabled:false must report disabled")
+	}
+	if !(&Config{ClientRfSamples: &ClientRfSamplesConfig{Enabled: true}}).ClientRfSamplesEnabled() {
+		t.Fatal("Enabled:true must report enabled")
+	}
+}
+
+// TestClientRfSamplesGateOn drives handleMessage with the feature ON: the
+// /rf topic message must be dispatched and write exactly one row.
+func TestClientRfSamplesGateOn(t *testing.T) {
+	store := newTestStore(t)
+	source := MQTTSource{Name: "test"}
+	cfg := &Config{ClientRfSamples: &ClientRfSamplesConfig{Enabled: true}}
+
+	handleMessage(store, "test", source, clientRfMsg(), nil, nil, cfg)
+
+	if n := clientRfSampleCount(t, store); n != 1 {
+		t.Fatalf("feature ON: expected 1 client_rf_samples row, got %d", n)
+	}
+}
+
+// TestClientRfSamplesGateOffDoesNotFallThroughToObserverPath is the
+// dispatch-shape regression test for the /rf topic, mirroring
+// TestClientRxCoverageGateOffDoesNotFallThroughToObserverPath. With the
+// feature OFF, a meshcore/client/<pubkey>/rf message must be dropped
+// outright — never fall through to the observer packet path, which would
+// take parts[1] ("client") as a region and parts[2] (the companion pubkey)
+// as an observer id.
+func TestClientRfSamplesGateOffDoesNotFallThroughToObserverPath(t *testing.T) {
+	store := newTestStore(t)
+	source := MQTTSource{Name: "test"}
+	cfg := &Config{} // ClientRfSamples nil ⇒ disabled
+
+	handleMessage(store, "test", source, clientRfMsgWithRaw(), nil, nil, cfg)
+
+	if n := clientRfSampleCount(t, store); n != 0 {
+		t.Fatalf("feature OFF: expected 0 client_rf_samples rows, got %d", n)
+	}
+	var observerRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE id = ?`, testCompanionPK).Scan(&observerRows); err != nil {
+		t.Fatal(err)
+	}
+	if observerRows != 0 {
+		t.Fatalf("feature OFF: the companion pubkey must not be registered as an observer, got %d rows", observerRows)
+	}
+	var clientRegionRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM observers WHERE iata = 'client'`).Scan(&clientRegionRows); err != nil {
+		t.Fatal(err)
+	}
+	if clientRegionRows != 0 {
+		t.Fatalf("feature OFF: no observer should be registered under a bogus 'client' region, got %d rows", clientRegionRows)
+	}
+	var txRows int
+	if err := store.db.QueryRow(`SELECT COUNT(*) FROM transmissions`).Scan(&txRows); err != nil {
+		t.Fatal(err)
+	}
+	if txRows != 0 {
+		t.Fatalf("feature OFF: the client-topic sample must not be ingested as an ordinary observer packet, got %d transmissions rows", txRows)
+	}
+}
+
+// TestClientRfSamplesBlacklistedDropped verifies a blacklisted operator
+// cannot skirt the observer blacklist via the /rf topic, mirroring
+// TestClientRxCoverageBlacklistedDropped.
+func TestClientRfSamplesBlacklistedDropped(t *testing.T) {
+	store := newTestStore(t)
+	source := MQTTSource{Name: "test"}
+	cfg := &Config{
+		ClientRfSamples:   &ClientRfSamplesConfig{Enabled: true},
+		ObserverBlacklist: []string{testCompanionPK},
+	}
+
+	handleMessage(store, "test", source, clientRfMsg(), nil, nil, cfg)
+
+	if n := clientRfSampleCount(t, store); n != 0 {
+		t.Fatalf("blacklisted companion: expected 0 client_rf_samples rows, got %d", n)
+	}
+}
+
+// TestClientPacketsGateUnaffectedByRfGate verifies the two client sub-topic
+// gates are independent: /rf disabled must not block /packets when
+// clientRxCoverage is enabled (and vice versa is covered by the existing
+// coverage_gate_test.go suite).
+func TestClientPacketsGateUnaffectedByRfGate(t *testing.T) {
+	store := newTestStore(t)
+	source := MQTTSource{Name: "test"}
+	cfg := &Config{
+		ClientRxCoverage: &ClientRxCoverageConfig{Enabled: true},
+		ClientRfSamples:  &ClientRfSamplesConfig{Enabled: false},
+	}
+
+	handleMessage(store, "test", source, clientCoverageMsg(), nil, nil, cfg)
+
+	if n := clientReceptionCount(t, store); n != 1 {
+		t.Fatalf("expected 1 client_receptions row with clientRxCoverage on / clientRfSamples off, got %d", n)
+	}
+}

config.example.json

@@ -14,7 +14,8 @@
     "packetDays": 30,
     "clientRxDays": 30,
     "clientRxObsDays": 30,
-    "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes."
+    "clientRfDays": 30,
+    "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. clientRfDays: RF environment sample rows (client_rf_samples, by sampled_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRfSamples table; unset means unbounded growth at roughly 240 rows/hour per driver. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes."
   },
   "db": {
     "vacuumOnStartup": false,
@@ -377,6 +378,8 @@
   "_comment_clientRxCoverage": "Opt-in mobile client-RX coverage (corescope-rx companions publishing GPS-tagged receptions to meshcore/client/<pubkey>/packets). Default OFF: when disabled the ingestor writes no client_receptions, the /api/rx-coverage|rx-leaderboard|nodes/{pubkey}/rx-coverage endpoints 404, and the UI hides the Coverage dashboard + Reach overlay. Set enabled=true to turn it on. SINGLE FLAG, BOTH PROCESSES: the ingestor and server each parse this same config.json, so this one clientRxCoverage.enabled entry gates both the ingest write path and the read endpoints — set it once, not per-process. TRUST: the feature requires an ACL-capable broker binding meshcore/client/{pubkey}/packets to that publisher; without ACLs the companion GPS is spoofable (see docs/client-rx-coverage.md). Retention: see retention.clientRxDays. Companion app + setup: https://github.com/efiten/corescope-rx.",
   "clientRxObservations": { "enabled": false },
   "_comment_clientRxObservations": "Opt-in diagnostic RF observations: every decodable packet a companion heard, whether or not it is attributable to a directly-heard node (client_rx_observations — never a coverage source; see clientRxCoverage for that). Default OFF. TOP-LEVEL FLAG: this is its own top-level Config field, NOT nested inside clientRxCoverage — config loading is plain json.Unmarshal with no DisallowUnknownFields, so nesting it under clientRxCoverage is silently ignored. Gated behind clientRxCoverage being reachable in the ingest control flow, but the JSON key sits alongside it. Requires the companion app's fullRfLog flag to actually upload unattributable packets, or this table stays empty even when enabled — see corescope-rx README. Retention: see retention.clientRxObsDays.",
+  "clientRfSamples": { "enabled": false },
+  "_comment_clientRfSamples": "Opt-in RF environment samples: a mobile client's attached companion radio's own counters (noise floor, RX/TX airtime, CRC errors, packet totals) along its GPS track, published on meshcore/client/<pubkey>/rf and stored in client_rf_samples — feeds a noise-floor map, a channel-utilisation map, and a CRC-error-rate map. Default OFF, independent of clientRxCoverage and clientRxObservations: a deployment can run any combination. TOP-LEVEL FLAG, same as clientRxObservations. Requires an ACL-capable broker binding meshcore/client/{pubkey}/rf to that publisher — see docs/client-rf-samples.md. Retention: see retention.clientRfDays; leaving it unset means unbounded growth.",
   "_comment_channelKeys": "Hex keys for decrypting channel messages. Key name = channel display name. public channel key is well-known.",
   "_comment_hashChannels": "Channel names whose keys are derived via SHA256. Key = SHA256(name)[:16]. Listed here so the ingestor can auto-derive keys.",
   "hashRegions": [

docs/client-rf-samples.md

@@ -0,0 +1,138 @@
+# Client RF Environment Samples
+
+Crowdsourced RF environment data from mobile clients: the same phone that samples
+[client RX coverage](client-rx-coverage.md) also reads its attached MeshCore companion radio's own
+counters — noise floor, RX/TX airtime, CRC errors, packet totals — along the GPS track, and publishes
+them on a separate topic. Unlike a fixed observer, which measures one point forever, a roaming
+companion builds a noise-floor map, a channel-utilisation map and a CRC-error-rate map as it moves.
+
+## Enabling RF samples (operators)
+
+Off by default. To turn it on:
+
+1. In CoreScope's `config.json`, set `"clientRfSamples": { "enabled": true }` and restart the
+   ingestor. Independent of `clientRxCoverage` — a deployment can run one, both, or neither.
+2. **Required: an ACL-capable broker**, same as coverage. Bind `meshcore/client/{PUBLIC_KEY}/rf` so
+   each client may publish only under its own pubkey. The ingestor already subscribes under
+   `meshcore/#`.
+3. Optionally set `retention.clientRfDays` to bound the table.
+
+## MQTT topic & payload
+
+Topic: `meshcore/client/{PUBLIC_KEY}/rf` — `{PUBLIC_KEY}` is the companion's pubkey.
+
+```json
+{
+  "type": "RF_SAMPLE",
+  "timestamp": "2026-08-17T10:00:00.000Z",
+  "gps": { "lat": 51.2, "lon": 4.4, "acc_m": 8.0 },
+  "stationary": false,
+  "uptime_secs": 84213,
+  "battery_mv": 3950,
+  "queue_len": 0,
+  "errors": 0,
+  "noise_floor": -119,
+  "last_rssi": -92,
+  "last_snr": -7.0,
+  "tx_air_secs": 120,
+  "rx_air_secs": 20877,
+  "recv": 4310,
+  "sent": 12,
+  "flood_rx": 3980,
+  "direct_rx": 330,
+  "flood_tx": 10,
+  "direct_tx": 2,
+  "recv_errors": 5
+}
+```
+
+- The discriminator is the `gps` object, same as coverage: a sample without `gps` (or with an
+  out-of-range `lat`/`lon`) is dropped.
+- `uptime_secs` is required. It is the only reboot/wrap detector the delta query has — see
+  [Deltas](#deltas--reboot-and-wrap-handling) below. A sample missing it is dropped and logged, the
+  same as a missing GPS fix, rather than silently defaulting to 0.
+- `stationary` is optional, defaults to `false` when absent.
+- Every counter besides `uptime_secs` is optional. Absent stays SQL `NULL`, never `0` — see
+  [recv_errors](#recv_errors-absent-vs-zero) below.
+- `errors` is not a counter despite sitting among them here: per the firmware stats frame it is an
+  error-flags bitmask, so it is stored as reported but deliberately excluded from `ClientRfDeltas`.
+- Subscription: the ingestor's default subscription (`meshcore/#`) already covers this topic. Sources
+  configured with an explicit topic list must add `meshcore/client/+/rf`.
+
+## Trust
+
+Identity = the companion pubkey, taken from the `{PUBLIC_KEY}` topic segment — never from a
+payload-supplied `origin_id`, which would defeat the ACL trust model. The ingestor rejects any topic
+pubkey that is not lowercase hex before writing (same `clientPubkeyRe` used for coverage), and the
+observer blacklist is enforced before any write, so a blacklisted operator cannot contribute RF
+samples any more than it can contribute coverage.
+
+## `recv_errors`: absent vs. zero
+
+`recv_errors` counts CRC errors. Firmware predating the counter cannot report it at all, and that
+case must stay `NULL` — storing `0` would read as "a perfectly clean channel," the opposite of "we
+don't know." The app is expected to omit the field entirely (not send a fabricated `0`) on firmware
+that doesn't support it, and the handler never collapses an absent optional field to a zero.
+
+## Storage — `client_rf_samples` (ingestor-owned)
+
+```
+client_rf_samples(
+  id, rx_pubkey, sampled_at, ingested_at, lat, lon, pos_acc_m, stationary,
+  uptime_secs, battery_mv, queue_len, errors, noise_floor, last_rssi, last_snr,
+  tx_air_secs, rx_air_secs, recv, sent, flood_rx, direct_rx, flood_tx, direct_tx,
+  recv_errors,
+  UNIQUE(rx_pubkey, sampled_at))   -- idempotent re-ingest
+```
+
+Every counter is stored as an absolute cumulative value, exactly as the radio reports it — the
+handler does no arithmetic. `errors` is the exception: per the firmware stats frame it is an
+error-flags bitmask, not a counter, so it is deliberately excluded from `ClientRfDeltas` — a future
+consumer must not wire it into a rate calculation. `sampled_at` is stored at **millisecond** precision (the
+`rxTimeMillisLayout` format shared with `client_rx_observations.rx_at`), not the second-resolution
+RFC3339 used by `client_receptions.rx_at`. This matters for two reasons: samples can arrive faster
+than once per second along a moving track, and Task 6's retention prune compares the cutoff
+lexicographically against these stored strings — a second-resolution cutoff against
+millisecond-resolution values would delete rows it should keep.
+
+Retention: `retention.clientRfDays` bounds the table by `sampled_at`; `0` disables it (Task 6).
+
+## Deltas — reboot and wrap handling
+
+`Store.ClientRfDeltas(rxPubkey, from, to)` derives consecutive-sample deltas (RX/TX airtime, recv,
+recv_errors, wall-clock milliseconds) for one radio over a time range, via a `LAG(...) OVER (ORDER BY
+sampled_at)` window query. Every counter in the table is cumulative, so a delta is only meaningful
+between two samples from the *same uninterrupted uptime*:
+
+- A pair is skipped whenever `uptime_secs` does not strictly increase between consecutive samples.
+  That is a reboot (uptime reset near 0) or a counter wrap, and subtracting across it would produce a
+  large negative or a bogus spike. `uptime_secs` has whole-second granularity, so sampling faster than
+  1 Hz means roughly every other pair gets skipped this way too (equal, not decreased, `uptime_secs`)
+  — that is expected, not a bug: strict-increase is deliberately correct for reboot detection, but it
+  means `ClientRfDeltas` effectively assumes callers sample at ≥1 s intervals if they want every
+  interval represented.
+- The absolute values always stay in `client_rf_samples`; only the delta view drops the
+  cross-reboot pair. No row is ever deleted or modified because of this check.
+- Each `*Delta` field on `ClientRfDelta` (`RxAirDelta`, `TxAirDelta`, `RecvDelta`, `RecvErrDelta`) is a
+  `*int64`, **nil** when either endpoint of the pair is NULL — i.e. the underlying counter is
+  unsupported on that firmware. A delta of `0` there would be indistinguishable from a measured zero,
+  which for `RecvErrDelta` in particular would render as "clean channel" on a radio that simply can't
+  count CRC errors. Consumers must treat nil as unknown, not zero.
+- When both endpoints of a pair *are* present, the per-metric delta additionally guards against
+  `cur < prev` (covers a wrapped individual counter even when `uptime_secs` itself looks monotonic) by
+  clamping to `0` rather than going negative.
+- `from`/`to` are compared **lexicographically** against millisecond-precision `sampled_at` strings, so
+  callers must pass bounds in the same `rxTimeMillisLayout` format
+  (`2006-01-02T15:04:05.000Z07:00`). A bound like `10:00:00Z` would lexicographically exclude a sample
+  stored as `10:00:00.500Z` (`.` sorts before `Z`), silently narrowing the range.
+- A phone clock running more than 14h ahead has its `timestamp` clamped to ingest time by
+  `resolveRxTimeCore` (same clamp used across the client-RX paths), which compresses `WallMillis` for a
+  buffered batch uploaded in one burst — several samples with distinct on-device timestamps can collapse
+  onto ingest-time values seconds apart. `sampled_at` can never be zero-spaced (the UNIQUE constraint
+  drops an exact collision), so this is a wrong-rate hazard, not a divide-by-zero: a consumer computing
+  a rate from `*Delta / WallMillis` must sanity-bound `WallMillis` before dividing.
+
+## Configurable values (future customizer)
+
+`retention.clientRfDays` is the only tunable so far; no color/threshold customization applies to this
+feature yet (no map rendering built on it in this task).