port(upstream): 26 clean upstream fixes — prune batching, /ws limits, observer liveness, watchdog race

.github/workflows/deploy.yml

@@ -86,6 +86,9 @@ jobs:
       - name: Staging disk-monitor unit tests (issue #1684)
         run: bash scripts/staging/test-disk-monitor.sh
 
+      - name: QA SQL parameter-binding unit tests (issue #1977)
+        run: bash qa/scripts/test-blacklist-sql.sh
+
       - name: Lint CSS variables (issue #1128)
         run: |
           set -e
@@ -100,6 +103,7 @@ jobs:
           node test-confidence-indicator.js
           node test-1659-analytics-warmup.js
           node test-channels-merge-1498-unit.js
+          node test-packets-local-channels.js
           node test-issue-1518-home-url.js
           node test-channel-decrypt-insecure-context.js
           node test-live-region-filter.js

channel-rainbow.json

@@ -295,5 +295,28 @@
   "#bogota": "6d0864985b64350ce4cbfebf4979e970",
   "#peru": "7e6fc347bf29a4c128ac3156865bd521",
   "#lima": "5f167ce354eca08ab742463df10ef255",
-  "Public": "8b3387e9c5cdea6ac9e5edbaa115cd72"
+  "Public": "8b3387e9c5cdea6ac9e5edbaa115cd72",
+  "#Public": "7534fe292203ccdce93db79377d4b566",
+  "#newbie": "ff2ecc59f59dbf051a10d880f0b4308d",
+  "#power": "c4d23a51329aeaee767234644d52cff8",
+  "#garden": "56ff6fe3500872c2765dbc178350005c",
+  "#jokes": "abe610d61dee508c0ac18793c36c41c3",
+  "#space": "5caa27568ba7a216bcc09fcb69c77ea1",
+  "#path": "29eed6c759d96346b30c869e23fd86f8",
+  "#hamradio": "83c8b01997654265938da8765cbc7db9",
+  "#amrron": "68ec397a1dd4f3416911d00211ed0ef6",
+  "#family": "7d1b805d4bf059c2ef06ed221794663b",
+  "#testing": "cde5e82cf515647dcb547a79a4f065d1",
+  "#wicked": "05e990d48943ee03c943c41dea0ddfe1",
+  "#gay": "3a612fa3bfc9f3b439cc230b01cbbc00",
+  "#t": "7b5336e85005be805b496bc27c722ff3",
+  "#doom": "7034910637d6b9b6c2cc3bcd8dfe9bb5",
+  "#empty": "3351cfa0dad99e1777cce210c983b561",
+  "#puppy": "c20cf85aab9561e6e65ac4a19a852198",
+  "#naked": "f423f23ddcf9e472c2ace38dce7591bc",
+  "#cruising": "550857053360e3492755251f9142e74c",
+  "#nightowls": "1411c5c3732d3df1b855644f78c47417",
+  "#worldcup": "d5786cc7bcee5a4828bbb20c3825c5a9",
+  "#nofilter": "cab31f589959debcd1721b7d252259b1",
+  "#politics": "64cc9fdaad67f03f23b4cec8c9a4468f"
 }

cmd/ingestor/config.go

@@ -144,7 +144,13 @@ func (c *Config) ClientRxCoverageEnabled() bool {
 type RetentionConfig struct {
 	NodeDays     int `json:"nodeDays"`
 	ObserverDays int `json:"observerDays"`
-	MetricsDays  int `json:"metricsDays"`
+	// ObserverPurgeDays hard-deletes observers that observerDays already
+	// soft-deleted, once they are older than this and no observation,
+	// metric or dropped-packet row still references them; 0 disables.
+	// Only meaningful above observerDays and packetDays — below those the
+	// reference guards keep every candidate row anyway.
+	ObserverPurgeDays int `json:"observerPurgeDays"`
+	MetricsDays       int `json:"metricsDays"`
 	// PacketDays is the retention window for transmissions (#1283).
 	// Ownership moved from cmd/server to cmd/ingestor; 0 disables.
 	PacketDays int `json:"packetDays"`
@@ -163,6 +169,15 @@ func (c *Config) PacketDaysOrZero() int {
 	return 0
 }
 
+// ObserverPurgeDaysOrZero returns the configured retention.observerPurgeDays
+// or 0 (disabled) if not set.
+func (c *Config) ObserverPurgeDaysOrZero() int {
+	if c.Retention != nil && c.Retention.ObserverPurgeDays > 0 {
+		return c.Retention.ObserverPurgeDays
+	}
+	return 0
+}
+
 // ClientRxDaysOrZero returns the configured retention.clientRxDays or 0
 // (disabled) if not set.
 func (c *Config) ClientRxDaysOrZero() int {

cmd/ingestor/db.go

@@ -1413,8 +1413,68 @@ 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, canRelay interface{}
+	model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta)
+
+	_, err := s.stmtUpsertObserver.Exec(
+		id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
+		name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
+	)
+	if err != nil {
+		s.Stats.WriteErrors.Add(1)
+		return err
+	}
+	s.Stats.ObserverUpserts.Add(1)
+
+	// Reactivate if this observer was previously marked inactive
+	s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id)
+	return nil
+}
+
+// UpsertObserverRetained applies the metadata from a RETAINED status message
+// without treating it as a sign of life. The broker replays retained messages
+// on every subscribe, so an ingestor restart would otherwise stamp last_seen
+// with the restart time for every observer that ever published one — making
+// dead observers permanently un-ageable by RemoveStaleObservers, and undoing
+// any inactive flag it did manage to set.
+//
+// Deliberately narrower than UpsertObserverAt: it updates metadata columns on
+// an existing row only. It does not advance last_seen, does not clear
+// inactive, does not bump packet_count, and does not INSERT — a retained-only
+// observer the analyzer has never heard from live describes a past that may be
+// months old and does not belong in the list. A live message from the same
+// observer arrives moments later and creates the row through the normal path.
+func (s *Store) UpsertObserverRetained(id, name, iata string, meta *ObserverMeta) error {
+	normalizedIATA := strings.TrimSpace(strings.ToUpper(iata))
+	model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay := observerMetaColumns(meta)
+
+	_, err := s.db.Exec(`
+		UPDATE observers SET
+			name = COALESCE(?, name),
+			iata = COALESCE(?, iata),
+			model = COALESCE(?, model),
+			firmware = COALESCE(?, firmware),
+			client_version = COALESCE(?, client_version),
+			radio = COALESCE(?, radio),
+			battery_mv = COALESCE(?, battery_mv),
+			uptime_secs = COALESCE(?, uptime_secs),
+			noise_floor = COALESCE(?, noise_floor),
+			can_relay = COALESCE(?, can_relay),
+			can_relay_seen = CASE WHEN ? IS NULL THEN can_relay_seen ELSE 1 END
+		WHERE id = ?`,
+		name, normalizedIATA, model, firmware, clientVersion, radio,
+		batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay, id,
+	)
+	if err != nil {
+		s.Stats.WriteErrors.Add(1)
+		return err
+	}
+	return nil
+}
+
+// observerMetaColumns flattens an *ObserverMeta into the driver args the
+// observer upserts bind. A nil field stays nil so the COALESCE in the SQL
+// leaves the existing column untouched (#1290).
+func observerMetaColumns(meta *ObserverMeta) (model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay interface{}) {
 	if meta != nil {
 		if meta.Model != nil {
 			model = *meta.Model
@@ -1449,20 +1509,7 @@ func (s *Store) UpsertObserverAt(id, name, iata string, meta *ObserverMeta, last
 			}
 		}
 	}
-
-	_, err := s.stmtUpsertObserver.Exec(
-		id, name, normalizedIATA, lastSeen, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
-		name, normalizedIATA, ingestNow, lastSeen, model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay, canRelay,
-	)
-	if err != nil {
-		s.Stats.WriteErrors.Add(1)
-		return err
-	}
-	s.Stats.ObserverUpserts.Add(1)
-
-	// Reactivate if this observer was previously marked inactive
-	s.db.Exec(`UPDATE observers SET inactive = 0 WHERE id = ? AND inactive = 1`, id)
-	return nil
+	return model, firmware, clientVersion, radio, batteryMv, uptimeSecs, noiseFloor, canRelay
 }
 
 // Close checkpoints the WAL and closes the database.
@@ -1839,6 +1886,42 @@ func (s *Store) RemoveStaleObservers(observerDays int) (int64, error) {
 	return removed, nil
 }
 
+// PurgeStaleObservers hard-deletes rows RemoveStaleObservers already soft-deleted,
+// once they are older than purgeDays and nothing references them any more. It is
+// the second stage of observer retention: the soft-delete hides the observer,
+// this reclaims the row after its packets have aged out via packetDays.
+//
+// observations.observer_idx is a bare rowid with no foreign key, so deleting a
+// still-referenced observer silently orphans history — packets_v stops resolving
+// the observer and the packets are mis-attributed. The three NOT EXISTS guards
+// are what make the delete safe; they are correctness, not defensive padding.
+// Each is an index seek per candidate row (observers is O(100)), so the whole
+// statement stays cheap enough for the daily retention tick.
+//
+// purgeDays <= 0 disables the purge (the default).
+func (s *Store) PurgeStaleObservers(purgeDays int) (int64, error) {
+	if purgeDays <= 0 {
+		return 0, nil // disabled
+	}
+	cutoff := time.Now().UTC().AddDate(0, 0, -purgeDays).Format(time.RFC3339)
+	// Tagged for /api/perf writer-lock visibility (#1340).
+	result, err := s.instrumentedExec("purge_observers", `
+		DELETE FROM observers
+		WHERE inactive = 1
+		  AND last_seen < ?
+		  AND NOT EXISTS (SELECT 1 FROM observations o WHERE o.observer_idx = observers.rowid)
+		  AND NOT EXISTS (SELECT 1 FROM observer_metrics m WHERE m.observer_id = observers.id)
+		  AND NOT EXISTS (SELECT 1 FROM dropped_packets d WHERE d.observer_id = observers.id)`, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("purge stale observers: %w", err)
+	}
+	purged, _ := result.RowsAffected()
+	if purged > 0 {
+		log.Printf("Purged %d inactive observer(s) with no remaining data (not seen in %d days)", purged, purgeDays)
+	}
+	return purged, nil
+}
+
 // DroppedPacket holds data for a packet rejected during ingest.
 type DroppedPacket struct {
 	Hash         string

cmd/ingestor/main.go

@@ -201,14 +201,10 @@ func main() {
 		// half-open TCP socket and re-dial when paho.IsConnected==true
 		// but no messages have flowed past the stall threshold. Throttled
 		// per source by the watchdog itself (forceReconnectThrottle).
-		// Disconnect(250) gives in-flight publishes 250ms to drain;
-		// Connect() returns immediately and paho's reconnect machinery
-		// takes over from there. Captured-by-value `client` is the same
-		// pointer used everywhere else for this source.
-		liveness.ForceReconnectFn = func() {
-			client.Disconnect(250)
-			client.Connect()
-		}
+		// Captured-by-value `client` is the same pointer used everywhere
+		// else for this source. See buildForceReconnectFn for why this is
+		// NOT simply "Disconnect(250) then Connect()".
+		liveness.ForceReconnectFn = buildForceReconnectFn(client, tag)
 		// PR #1216 r2 item 3: tag collisions used to log.Fatalf, which
 		// killed the entire ingestor over one config typo and recreated
 		// the #1212 total-ingest-stop class this PR exists to prevent.
@@ -264,6 +260,14 @@ func main() {
 	observerDays := cfg.ObserverDaysOrDefault()
 	store.RemoveStaleObservers(observerDays)
 
+	// Observer purge: second stage, hard-deletes long-inactive observers whose
+	// packets have already aged out. Always runs after the soft-delete so a row
+	// crossing both thresholds is finalised in a single pass. 0 = disabled.
+	observerPurgeDays := cfg.ObserverPurgeDaysOrZero()
+	if _, err := store.PurgeStaleObservers(observerPurgeDays); err != nil {
+		log.Printf("[prune] error: %v", err)
+	}
+
 	// Metrics retention: prune old metrics on startup
 	metricsDays := cfg.MetricsRetentionDays()
 	store.PruneOldMetrics(metricsDays)
@@ -323,9 +327,11 @@ func main() {
 	go func() {
 		time.Sleep(90 * time.Second) // stagger after metrics prune
 		store.RemoveStaleObservers(observerDays)
+		store.PurgeStaleObservers(observerPurgeDays)
 		store.RunIncrementalVacuum(vacuumPages)
 		for range observerRetentionTicker.C {
 			store.RemoveStaleObservers(observerDays)
+			store.PurgeStaleObservers(observerPurgeDays)
 			store.RunIncrementalVacuum(vacuumPages)
 		}
 	}()
@@ -554,6 +560,54 @@ func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions {
 	return opts
 }
 
+// buildForceReconnectFn builds the watchdog's forced-reconnect action for a
+// source (#1335, hardened against a race found while investigating a 100+
+// minute reconnect failure).
+//
+// paho's own client.IsConnected() — used as liveness.IsConnectedFn — reports
+// true not only when genuinely connected but for the ENTIRE time paho's
+// background AutoReconnect/ConnectRetry loop is retrying (status
+// reconnecting/connecting). So the watchdog's LivenessStalled classification
+// (IsConnected==true, no messages) fires just as often for "paho is actively,
+// correctly retrying a still-down broker" as it does for the true #1335
+// half-open-TCP case. Naively doing Disconnect(250) then Connect() in the
+// first case is actively harmful: paho's Disconnecting() must block until the
+// CURRENT in-flight connection attempt plus its backoff sleep unwind (up to
+// ConnectTimeout+MaxReconnectInterval, tens of seconds) before status
+// actually reaches `disconnected`. Disconnect(250) returns to the caller
+// after the 250ms quiesce regardless, so the following Connect() usually runs
+// while status is still the transitional `disconnecting` state — paho then
+// returns an error token (silently discarded by the old code) AND, because
+// Disconnect() was called at all, tears down paho's own retry loop for good
+// ("user requested no auto reconnection"). The client is left with nothing
+// retrying until the watchdog's next trigger fires, which can repeat the same
+// race — compounding into very long outages.
+//
+// client.IsConnectionOpen() (unlike IsConnected()) is strictly status ==
+// connected — never true while paho is reconnecting/connecting — so it
+// reliably distinguishes "genuinely connected, maybe half-open" (safe to
+// Disconnect then Connect; Disconnecting() does not need to wait on any
+// in-flight retry loop from status connected, so it completes well within
+// the 250ms quiesce) from "paho is already retrying on its own" (must NOT
+// call Disconnect; Connect() alone is a safe no-op per paho when a retry is
+// already under way, and properly starts a fresh attempt on the rare
+// occasion status has actually settled to disconnected).
+func buildForceReconnectFn(client mqtt.Client, tag string) func() {
+	return func() {
+		if client.IsConnectionOpen() {
+			client.Disconnect(250)
+		}
+		// Connect() resolves synchronously (Error() readable immediately,
+		// no Wait() needed) for both error returns and the "already
+		// retrying, treated as a safe no-op" success case — only a genuine
+		// fresh connection attempt leaves the token pending in the
+		// background, and we must not block this call on that.
+		if token := client.Connect(); token.Error() != nil {
+			log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %v", tag, token.Error())
+		}
+	}
+}
+
 func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, channelKeys map[string]string, regionKeys map[string][]byte, cfg *Config) {
 	// Liveness watchdog (#1212): record receipt before any processing so a
 	// slow handler still counts as "source is alive". Cheap atomic store.
@@ -632,6 +686,22 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message,
 		name, _ := msg["origin"].(string)
 		iata := parts[1]
 		meta := extractObserverMeta(msg)
+		// A replayed status message is the broker handing us the observer's
+		// last published snapshot — it is not evidence the observer is alive
+		// now. Stamping last_seen from it resurrects dead observers, so the
+		// replay path only refreshes metadata.
+		//
+		// Two ways to recognise one: the retain flag (our own subscribe), and
+		// the payload itself (a replay that reached us through the mosquitto
+		// bridge, where the flag does not survive the hop — see
+		// statusIsLiveness).
+		if m.Retained() || !statusIsLiveness(msg, time.Now().UTC()) {
+			if err := store.UpsertObserverRetained(observerID, name, iata, meta); err != nil {
+				log.Printf("MQTT [%s] retained observer status error: %v", tag, err)
+			}
+			log.Print(formatStatusLog(tag, firstNonEmpty(name, observerID), iata))
+			return
+		}
 		// observer.last_seen is "when did the analyzer last hear from this
 		// observer" — fundamentally an ingest-time question. Passing "" makes
 		// UpsertObserverAt use time.Now(), independent of the envelope timestamp

cmd/ingestor/main_test.go

@@ -105,13 +105,14 @@ func TestUnixTime(t *testing.T) {
 
 // mockMessage implements mqtt.Message for testing handleMessage
 type mockMessage struct {
-	topic   string
-	payload []byte
+	topic    string
+	payload  []byte
+	retained bool
 }
 
 func (m *mockMessage) Duplicate() bool  { return false }
 func (m *mockMessage) Qos() byte        { return 0 }
-func (m *mockMessage) Retained() bool   { return false }
+func (m *mockMessage) Retained() bool   { return m.retained }
 func (m *mockMessage) Topic() string     { return m.topic }
 func (m *mockMessage) MessageID() uint16 { return 0 }
 func (m *mockMessage) Payload() []byte   { return m.payload }