diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 2806126..7f67480 100644 --- a/internal/hub/hub.go +++ b/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()) } diff --git a/internal/hub/hub_test.go b/internal/hub/hub_test.go index 94be968..bb8fc7a 100644 --- a/internal/hub/hub_test.go +++ b/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") } } diff --git a/internal/hub/sent.go b/internal/hub/sent.go index 45ac801..f6ae3ed 100644 --- a/internal/hub/sent.go +++ b/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:] diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 9ca3584..cf00b0c 100644 --- a/internal/ingest/ingest.go +++ b/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 { diff --git a/internal/ingest/mqtt_client_test.go b/internal/ingest/mqtt_client_test.go new file mode 100644 index 0000000..096dd89 --- /dev/null +++ b/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") + } +} diff --git a/internal/ingest/neighbors.go b/internal/ingest/neighbors.go index 6579560..3972c65 100644 --- a/internal/ingest/neighbors.go +++ b/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 { diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 8f56c81..e875492 100644 --- a/internal/ingest/packet.go +++ b/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 } } diff --git a/internal/ingest/queue.go b/internal/ingest/queue.go index ff7e6d5..6a537c5 100644 --- a/internal/ingest/queue.go +++ b/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) diff --git a/internal/ingest/repeats_test.go b/internal/ingest/repeats_test.go index 178b8f0..2ad85a9 100644 --- a/internal/ingest/repeats_test.go +++ b/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 { diff --git a/internal/ingest/sanitize_test.go b/internal/ingest/sanitize_test.go new file mode 100644 index 0000000..dc26986 --- /dev/null +++ b/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) + } +} diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index c92095f..3299027 100644 --- a/internal/ingest/side_effects.go +++ b/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) diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 8dde612..2e6d8cd 100644 --- a/internal/ingest/status.go +++ b/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 { diff --git a/internal/ingest/topic_test.go b/internal/ingest/topic_test.go new file mode 100644 index 0000000..452c7fe --- /dev/null +++ b/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") + } +} diff --git a/internal/lora/airtime.go b/internal/lora/airtime.go index 1bd8a15..5abd63c 100644 --- a/internal/lora/airtime.go +++ b/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 diff --git a/internal/lora/airtime_test.go b/internal/lora/airtime_test.go index 795357b..d67d898 100644 --- a/internal/lora/airtime_test.go +++ b/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},