fix(ingest): strip NUL bytes from observer status

internal/ingest/ingest.go

@@ -374,7 +374,7 @@ func (w *Worker) handleMessageContext(parent context.Context, msg mqtt.Message)
 	// iata_codes.iata is CHAR(3); anything else would fail the DB insert
 	// downstream, so reject malformed topic segments here instead.
 	if !isValidIATA(iata) {
-		w.log.Warn("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic())
+		w.log.Debug("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic())
 		return
 	}
 

internal/ingest/logging_test.go

@@ -22,14 +22,15 @@ func (m malformedTopicMessage) MessageID() uint16 { return 0 }
 func (m malformedTopicMessage) Payload() []byte   { return nil }
 func (m malformedTopicMessage) Ack()              {}
 
-func TestMalformedIATAWarnNamesSource(t *testing.T) {
+func TestMalformedIATALogNamesSource(t *testing.T) {
 	for _, format := range []string{"json", "text"} {
 		for _, iata := range []string{"XX", "XX\r\nforged=true"} {
 			t.Run(format+"/"+strconv.Quote(iata), func(t *testing.T) {
 				var output bytes.Buffer
-				var handler slog.Handler = slog.NewJSONHandler(&output, nil)
+				opts := &slog.HandlerOptions{Level: slog.LevelDebug}
+				var handler slog.Handler = slog.NewJSONHandler(&output, opts)
 				if format == "text" {
-					handler = slog.NewTextHandler(&output, nil)
+					handler = slog.NewTextHandler(&output, opts)
 				}
 				previous := slog.Default()
 				slog.SetDefault(slog.New(handler))
@@ -38,23 +39,23 @@ func TestMalformedIATAWarnNamesSource(t *testing.T) {
 				topic := "meshcore/" + iata + "/0011223344556677/packets"
 				worker.handleMessage(malformedTopicMessage{topic: topic})
 				if bytes.Count(output.Bytes(), []byte("\n")) != 1 || bytes.ContainsRune(output.Bytes(), '\r') {
-					t.Fatalf("warning did not remain one escaped record: %q", output.String())
+					t.Fatalf("log did not remain one escaped record: %q", output.String())
 				}
 				if format == "json" {
 					var record map[string]any
 					if err := json.Unmarshal(output.Bytes(), &record); err != nil {
 						t.Fatal(err)
 					}
 					if record["iata"] != iata || record["topic"] != topic || record["msg"] != "dropped packet with malformed IATA" {
-						t.Fatalf("warning lost its source: %v", record)
+						t.Fatalf("log lost its source: %v", record)
 					}
 				} else {
 					expectedIATA, expectedTopic := iata, topic
 					if strings.ContainsAny(iata, "\r\n") {
 						expectedIATA, expectedTopic = strconv.Quote(iata), strconv.Quote(topic)
 					}
 					if !strings.Contains(output.String(), "iata="+expectedIATA) || !strings.Contains(output.String(), "topic="+expectedTopic) {
-						t.Fatalf("warning lost its source: %q", output.String())
+						t.Fatalf("log lost its source: %q", output.String())
 					}
 				}
 			})

internal/ingest/packet.go

@@ -732,14 +732,14 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
 	}
 	heardAt, err := parseHeardAt(envelope.Timestamp)
 	if err != nil {
-		w.log.Warn(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err)
+		w.log.Debug(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err)
 		heardAt = time.Now().UTC()
 	} else {
 		// clamp to server time if offset is suspicious (> 30 min drift)
 		now := time.Now().UTC()
 		diff := heardAt.UTC().Sub(now)
 		if diff > 30*time.Minute || diff < -30*time.Minute {
-			w.log.Warn(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[:8]))
 			heardAt = now
 		}
 	}

internal/ingest/status.go

@@ -4,10 +4,12 @@
 package ingest
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -70,6 +72,57 @@ func (s statusStats) usable() bool {
 	return s.NoiseFloor != 0 || s.TxAirSecs != 0 || s.RxAirSecs != 0
 }
 
+// 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 {
+	var dirty []string
+	for name, s := range fields {
+		if strings.Contains(*s, "\x00") {
+			*s = strings.ReplaceAll(*s, "\x00", "")
+			dirty = append(dirty, name)
+		}
+	}
+	slices.Sort(dirty)
+	return dirty
+}
+
+// stripJSONNULs drops \u0000 from a JSON document's strings, which jsonb rejects.
+// Payloads without one are returned untouched.
+func stripJSONNULs(raw []byte) ([]byte, bool) {
+	if !bytes.Contains(raw, []byte(`\u0000`)) {
+		return raw, false
+	}
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
+	var v any
+	if err := dec.Decode(&v); err != nil {
+		return raw, false
+	}
+	out, err := json.Marshal(dropNULs(v))
+	if err != nil {
+		return raw, false
+	}
+	return out, true
+}
+
+func dropNULs(v any) any {
+	switch v := v.(type) {
+	case string:
+		return strings.ReplaceAll(v, "\x00", "")
+	case []any:
+		for i := range v {
+			v[i] = dropNULs(v[i])
+		}
+	case map[string]any:
+		out := make(map[string]any, len(v))
+		for k, e := range v {
+			out[strings.ReplaceAll(k, "\x00", "")] = dropNULs(e)
+		}
+		return out
+	}
+	return v
+}
+
 // handleStatus processes a /status message and fans out an observerStatus event.
 func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) {
 	var envelope struct {
@@ -90,6 +143,15 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 		w.log.Warn(fmt.Sprintf("invalid pubkey hex in status from %s", pubkeyHex), "error", err)
 		return
 	}
+	dirty := stripNULs(map[string]*string{
+		"source": &envelope.ObserverType, "client_version": &envelope.SoftwareVersion,
+		"model": &envelope.HardwareModel, "firmware_version": &envelope.FirmwareVersion,
+		"origin": &envelope.DisplayName, "radio": &envelope.RadioString,
+	})
+	raw, rawDirty := stripJSONNULs(raw)
+	if len(dirty) > 0 || rawDirty {
+		w.log.Warn(fmt.Sprintf("stripped NUL bytes from status from %s", pubkeyHex), "fields", dirty, "metadata", rawDirty)
+	}
 	id, _, err := w.db.UpsertObserver(ctx, pubkey)
 	if err != nil {
 		w.log.Error(fmt.Sprintf("db: upsert observer failed in status from %s", pubkeyHex), "error", err)

internal/ingest/status_test.go

@@ -33,3 +33,42 @@ func TestStatusStatsUsable(t *testing.T) {
 		})
 	}
 }
+
+func TestStripNULs(t *testing.T) {
+	fw := "v1.17.1-d929643\x00\x00\x00"
+	name := "Canadaverse\x00 RemoteTerm"
+	model := "Heltec Wireless Paper"
+	got := stripNULs(map[string]*string{"firmware_version": &fw, "origin": &name, "model": &model})
+	if fw != "v1.17.1-d929643" || name != "Canadaverse RemoteTerm" || model != "Heltec Wireless Paper" {
+		t.Errorf("stripped to %q %q %q", fw, name, model)
+	}
+	if len(got) != 2 || got[0] != "firmware_version" || got[1] != "origin" {
+		t.Errorf("reported fields %v, want [firmware_version origin]", got)
+	}
+}
+
+func TestStripJSONNULs(t *testing.T) {
+	clean := []byte(`{"origin":"a","stats":{"uptime_secs":12}}`)
+	if out, changed := stripJSONNULs(clean); changed || string(out) != string(clean) {
+		t.Errorf("clean payload rewritten: %s", out)
+	}
+
+	dirty := []byte(`{"firmware_version":"v1.17\u0000\u0000","extra":{"list":["x\u0000y"],"n":1.50}}`)
+	out, changed := stripJSONNULs(dirty)
+	if !changed {
+		t.Fatal("expected payload to be cleaned")
+	}
+	var v struct {
+		FirmwareVersion string `json:"firmware_version"`
+		Extra           struct {
+			List []string    `json:"list"`
+			N    json.Number `json:"n"`
+		} `json:"extra"`
+	}
+	if err := json.Unmarshal(out, &v); err != nil {
+		t.Fatal(err)
+	}
+	if v.FirmwareVersion != "v1.17" || v.Extra.List[0] != "xy" || v.Extra.N != "1.50" {
+		t.Errorf("cleaned to %s", out)
+	}
+}