fix(ingest): validate topic keys and harden against malformed input

A short topic pubkey panicked a lane worker and killed the process; keys must
now be 32 bytes and lane workers recover. Also reject non-finite radio values
before airtime, strip NULs from advert names and neighbour scopes, guard the
MQTT client pointer, and never stream a broker copy of a hearing as a repeat.

internal/hub/hub.go

@@ -243,8 +243,8 @@ func (h *Hub) RepeatsWanted() bool {
 	return h.repeatClients.Load() > 0
 }
 
-// MarkSent records every hearing's path and reports whether it is new within the TTL,
-// so broker copies and same-path duplicates are never streamed as repeats.
+// MarkSent records every hearing's path and reports whether it is a later, new path for a
+// packet the observer already heard, so broker copies and same-path duplicates never stream.
 func (h *Hub) MarkSent(packetHash, observerID, path []byte) bool {
 	return h.sent.mark(packetHash, observerID, path, time.Now())
 }

internal/hub/hub_test.go

@@ -516,20 +516,23 @@ func TestSentPaths(t *testing.T) {
 	start := time.Now()
 	s := newSentPaths(time.Minute, 3)
 	hash, observer := []byte{1, 2}, []byte{3}
-	if !s.mark(hash, observer, []byte{0xaa}, start) {
-		t.Fatal("first hearing refused")
+	if s.mark(hash, observer, []byte{0xaa}, start) {
+		t.Fatal("first hearing counted as a repeat")
 	}
 	if s.mark(hash, observer, []byte{0xaa}, start.Add(time.Second)) {
-		t.Fatal("exact copy within the window allowed")
+		t.Fatal("broker copy of the first hearing counted as a repeat")
 	}
 	if !s.mark(hash, observer, []byte{0xaa, 0xbb}, start.Add(time.Second)) {
 		t.Fatal("new path refused")
 	}
-	if !s.mark(hash, []byte{4}, []byte{0xaa}, start.Add(time.Second)) {
-		t.Fatal("other observer refused")
+	if s.mark(hash, observer, []byte{0xaa, 0xbb}, start.Add(time.Second)) {
+		t.Fatal("exact copy within the window allowed")
 	}
-	if !s.mark(hash, observer, []byte{0xaa}, start.Add(time.Minute)) {
-		t.Fatal("copy after the window refused")
+	if s.mark(hash, []byte{4}, []byte{0xaa}, start.Add(time.Second)) {
+		t.Fatal("other observer's first hearing counted as a repeat")
+	}
+	if s.mark(hash, observer, []byte{0xcc}, start.Add(time.Minute)) {
+		t.Fatal("hearing after the window not treated as a first hearing")
 	}
 
 	s = newSentPaths(time.Hour, 3)
@@ -539,7 +542,7 @@ func TestSentPaths(t *testing.T) {
 	if len(s.at) != 3 || len(s.order) != 3 {
 		t.Fatalf("size bound: %d keys, %d queued", len(s.at), len(s.order))
 	}
-	if !s.mark(hash, observer, []byte{0}, start) {
+	if _, ok := s.at[s.key(hash, observer, []byte{0})]; ok {
 		t.Fatal("oldest key not evicted at the size bound")
 	}
 }

internal/hub/sent.go

@@ -34,18 +34,29 @@ func newSentPaths(ttl time.Duration, max int) *sentPaths {
 	return &sentPaths{seed: maphash.MakeSeed(), ttl: ttl, max: max, at: make(map[uint64]time.Time)}
 }
 
-// mark records the hearing and reports whether it was not already recorded within the TTL.
+// mark records the hearing and reports whether it is a new path for a (packet, observer) already
+// heard over another one. Keying the first hearing by pair, not by which worker won the insert,
+// means a broker copy of it is never a repeat however the two workers interleave.
 func (s *sentPaths) mark(packetHash, observerID, path []byte, now time.Time) bool {
+	pair, hearing := s.key(packetHash, observerID), s.key(packetHash, observerID, path)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	firstHearing := s.add(pair, now)
+	return s.add(hearing, now) && !firstHearing
+}
+
+func (s *sentPaths) key(parts ...[]byte) uint64 {
 	var h maphash.Hash
 	h.SetSeed(s.seed)
-	for _, part := range [][]byte{packetHash, observerID, path} {
+	for _, part := range parts {
 		h.WriteByte(byte(len(part)))
 		h.Write(part)
 	}
-	key := h.Sum64()
+	return h.Sum64()
+}
 
-	s.mu.Lock()
-	defer s.mu.Unlock()
+// add records key unless already present within the TTL and reports whether it was new.
+func (s *sentPaths) add(key uint64, now time.Time) bool {
 	for len(s.order) > 0 && (now.Sub(s.order[0].at) >= s.ttl || len(s.order) >= s.max) {
 		delete(s.at, s.order[0].key)
 		s.order = s.order[1:]

internal/ingest/ingest.go

@@ -39,11 +39,13 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/binary"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"log/slog"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -226,7 +228,7 @@ type Worker struct {
 	hub              *hub.Hub
 	keys             ChannelKeyStore
 	scopes           ScopeStore
-	client           mqtt.Client
+	client           atomic.Pointer[mqtt.Client] // read by /brokers while Start sets it
 	capabilities     capabilityCache
 	onNodeUpsert     func(ctx context.Context, nodeID uuid.UUID)
 	onObserverUpsert func(ctx context.Context, observerID uuid.UUID)
@@ -289,9 +291,10 @@ func (w *Worker) Start(ctx context.Context) {
 			w.log.Warn("connection lost, will reconnect", "error", err)
 		})
 
-	w.client = mqtt.NewClient(opts)
+	client := mqtt.NewClient(opts)
+	w.client.Store(&client)
 
-	token := w.client.Connect()
+	token := client.Connect()
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
 	connected := token.Done()
@@ -305,7 +308,7 @@ func (w *Worker) Start(ctx context.Context) {
 		case <-ticker.C:
 			reportDrops()
 		case <-ctx.Done():
-			w.client.Disconnect(500)
+			client.Disconnect(500)
 			return
 		}
 	}
@@ -317,10 +320,11 @@ func (w *Worker) BrokerName() string {
 }
 
 func (w *Worker) IsConnected() bool {
-	if w.client == nil {
+	c := w.client.Load()
+	if c == nil {
 		return false
 	}
-	return w.client.IsConnected()
+	return (*c).IsConnected()
 }
 
 func (w *Worker) SetCacheInvalidators(onNode, onObserver func(ctx context.Context, id uuid.UUID)) {
@@ -356,6 +360,15 @@ func isValidIATA(s string) bool {
 	return true
 }
 
+// isValidPubkeyHex reports whether s is a 32-byte public key in hex.
+func isValidPubkeyHex(s string) bool {
+	if len(s) != 64 {
+		return false
+	}
+	_, err := hex.DecodeString(s)
+	return err == nil
+}
+
 // handleMessage dispatches incoming MQTT messages by subtopic.
 // Each message is processed with a 30s timeout to prevent slow DB calls
 // from holding a processing worker indefinitely.
@@ -377,6 +390,10 @@ func (w *Worker) handleMessageContext(parent context.Context, msg mqtt.Message)
 		w.log.Debug("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic())
 		return
 	}
+	if !isValidPubkeyHex(pubkeyHex) {
+		w.log.Debug("dropped message with malformed observer pubkey", "pubkey", pubkeyHex, "topic", msg.Topic())
+		return
+	}
 
 	// Drop packets from IATAs outside the configured geographic filter.
 	if w.cfg.AllowedIATAs != nil {

internal/ingest/mqtt_client_test.go

@@ -0,0 +1,30 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package ingest
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/hub"
+)
+
+// Run with -race: /brokers reads the client while Start creates it.
+func TestIsConnectedWhileStarting(t *testing.T) {
+	w := New(Config{BrokerName: "test", URL: "tcp://127.0.0.1:1"}, &stubDB{}, hub.New(), &stubKeys{}, &stubScopes{})
+	ctx, cancel := context.WithCancel(context.Background())
+	stopped := make(chan struct{})
+	go func() { w.Start(ctx); close(stopped) }()
+	deadline := time.Now().Add(100 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		w.IsConnected()
+	}
+	cancel()
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("worker did not stop")
+	}
+}

internal/ingest/neighbors.go

@@ -55,7 +55,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra
 	// self.scopes is always known (it's the observer's own config, not an OTA
 	// query), so this is an unconditional write -- unlike the per-neighbor
 	// scopes below, there's no "query failed" case to protect against here.
-	if err := w.db.UpdateObserverRegionScope(ctx, observerID, report.Self.Scopes); err != nil {
+	if err := w.db.UpdateObserverRegionScope(ctx, observerID, cleanText(report.Self.Scopes)); err != nil {
 		w.log.Error(fmt.Sprintf("db: update observer region scope failed for %s", pubkeyHex), "error", err)
 	}
 
@@ -94,7 +94,8 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra
 		// was last known.
 		var regionScope *string
 		if n.Status == "responded" {
-			regionScope = &n.Scopes
+			scope := cleanText(n.Scopes)
+			regionScope = &scope
 		}
 
 		if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, neighborNodeID, iata, &snr, regionScope); err != nil {

internal/ingest/packet.go

@@ -427,7 +427,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 
 			var name *string
 			if hasName {
-				n := strings.ToValidUTF8(appData.Name, "\uFFFD")
+				n := cleanText(appData.Name)
 				name = &n
 				if n != "" {
 					summary = name
@@ -747,7 +747,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 		now := time.Now().UTC()
 		diff := heardAt.UTC().Sub(now)
 		if diff > 30*time.Minute || diff < -30*time.Minute {
-			w.log.Debug(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex[:8]))
+			w.log.Debug(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex))
 			heardAt = now
 		}
 	}

internal/ingest/queue.go

@@ -5,6 +5,8 @@ package ingest
 
 import (
 	"context"
+	"log/slog"
+	"runtime/debug"
 	"strings"
 	"sync"
 
@@ -33,7 +35,7 @@ func newMessageQueue(workers, capacity, maxBytes int, handle func(context.Contex
 				if ctx.Err() != nil {
 					return
 				}
-				handle(ctx, m)
+				handleSafely(ctx, m, handle)
 				q.mu.Lock()
 				q.bytes -= len(m.Topic()) + len(m.Payload())
 				q.mu.Unlock()
@@ -44,6 +46,16 @@ func newMessageQueue(workers, capacity, maxBytes int, handle func(context.Contex
 	return q
 }
 
+// handleSafely keeps one bad message from killing the lane, and with it the process.
+func handleSafely(ctx context.Context, m mqtt.Message, handle func(context.Context, mqtt.Message)) {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("ingest handler panicked", "component", "ingest", "topic", m.Topic(), "panic", r, "stack", string(debug.Stack()))
+		}
+	}()
+	handle(ctx, m)
+}
+
 func (q *messageQueue) enqueue(m mqtt.Message) bool {
 	// Keep status and packets from each observer in arrival order.
 	parts := strings.SplitN(m.Topic(), "/", 4)

internal/ingest/repeats_test.go

@@ -123,18 +123,32 @@ func TestHandlePacket_PathMarkedRightAfterInsert(t *testing.T) {
 	packet.Path = []byte{0x11}
 	packet.PathLength = 1
 	hash := packet.PacketHash()
-	var fresh bool
+	var repeat bool
 	r.db.dbHook = func() {
 		// stubDB.UpsertObserver returns uuid.Nil, so that is the observer id ingest marks with
-		fresh = r.w.hub.MarkSent(hash[:], uuid.Nil[:], packet.Path)
+		repeat = r.w.hub.MarkSent(hash[:], uuid.Nil[:], []byte{0x22})
 	}
 	r.db.observationInserted = true
 	r.w.handlePacket(r.ctx, "YOW", "0102", packetEnvelope(r.t, packet))
-	if fresh {
+	if !repeat {
 		t.Fatal("path was not marked right after the insert")
 	}
 }
 
+// A broker copy can lose the insert race yet mark its path first; it is still the first hearing.
+func TestHandlePacket_BrokerCopyMarkedFirstIsNotARepeat(t *testing.T) {
+	r := newRepeatHarness(t, true)
+	if got := r.hear(false, 0x11); len(got) != 0 {
+		t.Fatalf("broker copy streamed as a repeat: %s", got)
+	}
+	if got := r.hear(true, 0x11); len(got) != 1 || strings.Contains(string(got[0]), "isRepeat") {
+		t.Fatalf("inserted hearing: %s", got)
+	}
+	if got := r.hear(false, 0x22, 0x11); len(got) != 1 {
+		t.Fatalf("later path sent %d events", len(got))
+	}
+}
+
 func TestHandlePacket_PathsRecordedWithoutSubscribers(t *testing.T) {
 	r := newRepeatHarness(t, false)
 	if got := r.hear(true, 0x11); len(got) != 1 {

internal/ingest/sanitize_test.go

@@ -0,0 +1,102 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package ingest
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/meshcore-go/meshcore-go"
+)
+
+type statusCaptureDB struct {
+	*stubDB
+	status []UpdateObserverStatusParams
+}
+
+func (s *statusCaptureDB) UpdateObserverStatus(_ context.Context, p UpdateObserverStatusParams) (uuid.UUID, error) {
+	s.status = append(s.status, p)
+	return uuid.Nil, nil
+}
+
+func TestHandleStatus_RejectsInsaneRadio(t *testing.T) {
+	for _, tc := range []struct {
+		radio    string
+		freq, bw float32
+	}{
+		{"910.5,62.5,7,5", 910.5, 62.5},
+		{"910.5,NaN,7,5", 910.5, 0},
+		{"Inf,62.5,7,5", 0, 62.5},
+		{"910.5,1e-40,7,5", 910.5, 0},
+		{"-910.5,-62.5,7,5", 0, 0},
+		{"1e9,1e9,7,5", 0, 0},
+	} {
+		t.Run(tc.radio, func(t *testing.T) {
+			w, base := newTestWorker()
+			db := &statusCaptureDB{stubDB: base}
+			w.db = db
+			w.handleStatus(context.Background(), strings.Repeat("01", 32), []byte(`{"radio":"`+tc.radio+`"}`))
+			if len(db.status) != 1 {
+				t.Fatal("status not stored")
+			}
+			if got := db.status[0]; got.RadioFreqMHz != tc.freq || got.RadioBWKHz != tc.bw {
+				t.Fatalf("radio stored as %v MHz / %v kHz, want %v / %v", got.RadioFreqMHz, got.RadioBWKHz, tc.freq, tc.bw)
+			}
+		})
+	}
+}
+
+func TestHandlePacket_AdvertNameNULsStripped(t *testing.T) {
+	w, base := newTestWorker()
+	db := &frameCaptureDB{stubDB: base}
+	w.db = db
+	data := append([]byte{meshcore.AdvertTypeRepeater | meshcore.AdvertNameMask}, "Base\x00Camp\x00\x00"...)
+	w.handlePacket(context.Background(), "YOW", "0102", packetEnvelope(t, buildAdvertPacketWithData(t, data, false)))
+	if len(db.packets) != 1 || bytes.Contains(db.packets[0].ParsedPayload, []byte(`\u0000`)) {
+		t.Fatalf("parsed payload kept NULs: %s", db.packets[0].ParsedPayload)
+	}
+	if base.upsertNodeCalls != 1 || base.upsertNodeParams.Name != "BaseCamp" {
+		t.Fatalf("node name stored as %q", base.upsertNodeParams.Name)
+	}
+}
+
+type neighborCaptureDB struct {
+	*stubDB
+	selfScope string
+	scopes    []string
+}
+
+func (s *neighborCaptureDB) GetNodeByPubkey(_ context.Context, pubkey []byte) (uuid.UUID, error) {
+	if len(pubkey) == 0 {
+		return uuid.Nil, errors.New("not found")
+	}
+	return uuid.UUID{pubkey[0]}, nil
+}
+
+func (s *neighborCaptureDB) UpdateObserverRegionScope(_ context.Context, _ uuid.UUID, scope string) error {
+	s.selfScope = scope
+	return nil
+}
+
+func (s *neighborCaptureDB) UpsertNodeNeighbor(_ context.Context, _, _ uuid.UUID, _ string, _ *float32, scope *string) error {
+	if scope != nil {
+		s.scopes = append(s.scopes, *scope)
+	}
+	return nil
+}
+
+func TestHandleNeighbors_ScopeNULsStripped(t *testing.T) {
+	w, base := newTestWorker()
+	db := &neighborCaptureDB{stubDB: base}
+	w.db = db
+	report := `{"self":{"scopes":"#ott\u0000"},"neighbors":[{"pubkey":"` + strings.Repeat("02", 32) + `","status":"responded","scopes":"#y\u0000ow"}]}`
+	w.handleNeighbors(context.Background(), "YOW", strings.Repeat("01", 32), []byte(report))
+	if db.selfScope != "#ott" || len(db.scopes) != 1 || db.scopes[0] != "#yow" {
+		t.Fatalf("scopes stored as %q %q", db.selfScope, db.scopes)
+	}
+}

internal/ingest/side_effects.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/MeshCore-Beacon/beacon-server/internal/api"
@@ -106,7 +105,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 		}
 		params := UpsertNodeParams{
 			PublicKey:       advert.PublicKey.PublicKeyBytes(),
-			Name:            strings.ToValidUTF8(advert.AppData().Name, "\uFFFD"),
+			Name:            cleanText(advert.AppData().Name),
 			NodeType:        advert.Type(),
 			Latitude:        lat,
 			Longitude:       lon,
@@ -181,7 +180,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 		evt := nodeUpdateEvent{
 			NodeID:       nodeID.String(),
 			PublicKey:    pubkeyHex,
-			Name:         advert.AppData().Name,
+			Name:         cleanText(advert.AppData().Name),
 			NodeType:     advert.Type(),
 			NodeTypeName: api.NodeTypeName(int16(advert.Type())),
 			IATA:         iata,
@@ -230,8 +229,8 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 				ChannelID:   result.ChannelID,
 				ChannelHash: hex.EncodeToString(channelHashBytes),
 				PacketHash:  hex.EncodeToString(packetHash),
-				SenderName:  strings.ReplaceAll(strings.ToValidUTF8(result.Payload.Sender, "\uFFFD"), "\x00", ""),
-				Content:     strings.ReplaceAll(strings.ToValidUTF8(result.Payload.Text, "\uFFFD"), "\x00", ""),
+				SenderName:  cleanText(result.Payload.Sender),
+				Content:     cleanText(result.Payload.Text),
 				SentAt:      time.Unix(int64(result.Payload.Timestamp), 0).UnixMilli(),
 			}
 			w.broadcast(hub.EventChannelMessage, iata, 0, fmt.Sprintf("%02x", grpTxt.ChannelHash), evt)

internal/ingest/status.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/MeshCore-Beacon/beacon-server/internal/hub"
+	"github.com/MeshCore-Beacon/beacon-server/internal/lora"
 )
 
 // UpdateObserverStatusParams carries the fields parsed from a /status message.
@@ -72,6 +73,25 @@ func (s statusStats) usable() bool {
 	return s.NoiseFloor != 0 || s.TxAirSecs != 0 || s.RxAirSecs != 0
 }
 
+// LoRa radios span 137 MHz sub-GHz to 2.5 GHz SX128x.
+const (
+	minRadioFreqMHz = 100
+	maxRadioFreqMHz = 3000
+)
+
+// parseRadioFloat rejects NaN, Inf and anything outside [lo, hi]; observers send free text and
+// one bad value would poison every airtime the observer is later costed with.
+func parseRadioFloat(s string, lo, hi float64) (float64, error) {
+	f, err := strconv.ParseFloat(s, 32)
+	if err != nil {
+		return 0, err
+	}
+	if !(f >= lo && f <= hi) {
+		return 0, fmt.Errorf("radio value %q outside %g..%g", s, lo, hi)
+	}
+	return f, nil
+}
+
 // stripNULs removes NUL bytes, which Postgres text columns reject, and returns the
 // names of the fields that had any.
 func stripNULs(fields map[string]*string) []string {
@@ -86,6 +106,11 @@ func stripNULs(fields map[string]*string) []string {
 	return dirty
 }
 
+// cleanText makes over-the-air text storable: invalid UTF-8 replaced, NULs dropped.
+func cleanText(s string) string {
+	return strings.ReplaceAll(strings.ToValidUTF8(s, "�"), "\x00", "")
+}
+
 // stripJSONNULs drops \u0000 from a JSON document's strings, which jsonb rejects.
 // Payloads without one are returned untouched.
 func stripJSONNULs(raw []byte) ([]byte, bool) {
@@ -197,13 +222,13 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 	if len(radio) != 4 {
 		w.log.Warn(fmt.Sprintf("missing or malformed radio params in status from %s, skipping radio fields", pubkeyHex))
 	} else {
-		freq, err := strconv.ParseFloat(radio[0], 32)
+		freq, err := parseRadioFloat(radio[0], minRadioFreqMHz, maxRadioFreqMHz)
 		if err != nil {
 			w.log.Warn(fmt.Sprintf("error parsing radio freq in status from %s", pubkeyHex), "error", err)
 		} else {
 			params.RadioFreqMHz = float32(freq)
 		}
-		bw, err := strconv.ParseFloat(radio[1], 32)
+		bw, err := parseRadioFloat(radio[1], lora.MinBWKHz, lora.MaxBWKHz)
 		if err != nil {
 			w.log.Warn(fmt.Sprintf("error parsing radio bw in status from %s", pubkeyHex), "error", err)
 		} else {

internal/ingest/topic_test.go

@@ -0,0 +1,79 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package ingest
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/google/uuid"
+)
+
+type observerCountDB struct {
+	*frameCaptureDB
+	observers int
+}
+
+func (s *observerCountDB) UpsertObserver(context.Context, []byte, string) (uuid.UUID, string, error) {
+	s.observers++
+	return uuid.Nil, "", nil
+}
+
+func TestTopicPubkeyMustBe32Bytes(t *testing.T) {
+	var envelope map[string]string
+	if err := json.Unmarshal(packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16))), &envelope); err != nil {
+		t.Fatal(err)
+	}
+	envelope["timestamp"] = time.Now().UTC().Add(-time.Hour).Format(time.RFC3339) // takes the clamp log path
+	body, err := json.Marshal(envelope)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{"ab", strings.Repeat("ab", 31), strings.Repeat("ab", 33), strings.Repeat("zz", 32)} {
+		for _, sub := range []string{"packets", "status", "neighbors"} {
+			t.Run(sub+"/"+key, func(t *testing.T) {
+				w, base := newTestWorker()
+				db := &observerCountDB{frameCaptureDB: &frameCaptureDB{stubDB: base}}
+				w.db = db
+				w.handleMessage(queuedTestMessage{malformedTopicMessage{"meshcore/YOW/" + key + "/" + sub}, body})
+				if db.observers != 0 || len(db.packets) != 0 {
+					t.Fatalf("malformed pubkey processed: %d observers, %d packets", db.observers, len(db.packets))
+				}
+			})
+		}
+	}
+	w, base := newTestWorker()
+	db := &observerCountDB{frameCaptureDB: &frameCaptureDB{stubDB: base}}
+	w.db = db
+	w.handleMessage(queuedTestMessage{malformedTopicMessage{"meshcore/YOW/" + strings.Repeat("AB", 32) + "/packets"}, body})
+	if len(db.packets) != 1 {
+		t.Fatal("valid pubkey dropped")
+	}
+}
+
+func TestQueueSurvivesHandlerPanic(t *testing.T) {
+	done := make(chan struct{})
+	q := newMessageQueue(1, 8, 4096, func(_ context.Context, m mqtt.Message) {
+		if string(m.Payload()) == "0" {
+			panic("bad message")
+		}
+		close(done)
+	})
+	t.Cleanup(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_ = q.close(ctx)
+	})
+	q.enqueue(queueMessage("aa", "packets", 0))
+	q.enqueue(queueMessage("aa", "packets", 1))
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("lane stopped after a panic")
+	}
+}

internal/lora/airtime.go

@@ -12,6 +12,10 @@ const (
 	maxSF = 12
 	minCR = 5
 	maxCR = 8
+
+	// MinBWKHz and MaxBWKHz bracket every LoRa bandwidth (7.8 kHz sub-GHz to 1625 kHz SX128x).
+	MinBWKHz = 7
+	MaxBWKHz = 1625
 )
 
 // PreambleSymbols mirrors MeshCore's preambleLengthForSF: 32 symbols at SF<=8, 16 above.
@@ -34,7 +38,7 @@ func FrameLength(transportCodes bool, pathLen, payloadLen int) int {
 // TimeOnAirMs is the Semtech LoRa time-on-air (RadioLib getTimeOnAir: CRC on, explicit header,
 // LDRO auto at t_sym >= 16 ms). ok is false when any parameter is outside the costable range.
 func TimeOnAirMs(frameLen, sf int, bwKHz float64, cr int) (ms float64, ok bool) {
-	if frameLen <= 0 || sf < minSF || sf > maxSF || bwKHz <= 0 || cr < minCR || cr > maxCR {
+	if frameLen <= 0 || sf < minSF || sf > maxSF || !(bwKHz >= MinBWKHz && bwKHz <= MaxBWKHz) || cr < minCR || cr > maxCR {
 		return 0, false
 	}
 	tSym := math.Exp2(float64(sf)) / bwKHz

internal/lora/airtime_test.go

@@ -104,6 +104,10 @@ func TestTimeOnAirMsNotCostable(t *testing.T) {
 		{"sf below range", 50, 6, 62.5, 5},
 		{"sf above range", 50, 13, 62.5, 5},
 		{"bw zero", 50, 7, 0, 5},
+		{"bw NaN", 50, 7, math.NaN(), 5},
+		{"bw infinite", 50, 7, math.Inf(1), 5},
+		{"bw denormal", 50, 12, 1e-40, 5},
+		{"bw above range", 50, 7, 1e9, 5},
 		{"cr below range", 50, 7, 62.5, 4},
 		{"cr above range", 50, 7, 62.5, 9},
 		{"empty frame", 0, 7, 62.5, 5},