feat(p987): report why frames fail to decode

p987/main.go

@@ -23,6 +23,9 @@ func main() {
 		database.Init()
 	}
 
+	// Started before the subscription so the first frames are counted.
+	service.InitDecodeReporting()
+
 	mqtt.SetMessageHandler(service.HandleInboundMessage)
 	if err := mqtt.Init(context.Background()); err != nil {
 		logger.SugarLogger.Fatalf("Failed to initialize MQTT: %v", err)

p987/service/decode_report.go

@@ -0,0 +1,128 @@
+package service
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gaucho-racing/mapache/p987/pkg/logger"
+)
+
+// A frame that produces no signals never reaches the live path — it is
+// stored with a status in its metadata and dropped. That is correct
+// behaviour but it made "frames arrive, nothing shows up live" impossible
+// to diagnose without querying ClickHouse, so decode outcomes are counted
+// here and reported.
+//
+// Frames arrive at a few hundred per second, so a line per failure would
+// bury everything else and out-write the disk. Each distinct
+// (bus, can id, reason) logs once when first seen — a new unknown id is
+// visible immediately — and repeats are rolled up on an interval.
+
+const decodeReportInterval = 30 * time.Second
+
+// maxReportedKinds bounds the summary line. A bus carrying dozens of
+// undecodable ids should not produce an unbounded log entry.
+const maxReportedKinds = 12
+
+type decodeKey struct {
+	bus    string
+	canID  int
+	status string
+}
+
+type decodeStat struct {
+	total       int64
+	sinceReport int64
+}
+
+var (
+	decodeMu      sync.Mutex
+	decodeStats   = map[decodeKey]*decodeStat{}
+	decodedOK     int64
+	decodedFailed int64
+)
+
+// NoteDecodeOutcome records one frame's decode result.
+func NoteDecodeOutcome(bus string, canID int, status, note string) {
+	decodeMu.Lock()
+	if status == statusOK {
+		decodedOK++
+		decodeMu.Unlock()
+		return
+	}
+	decodedFailed++
+
+	key := decodeKey{bus: bus, canID: canID, status: status}
+	stat, ok := decodeStats[key]
+	if !ok {
+		stat = &decodeStat{}
+		decodeStats[key] = stat
+	}
+	first := stat.total == 0
+	stat.total++
+	stat.sinceReport++
+	decodeMu.Unlock()
+
+	if first {
+		logger.SugarLogger.Warnf("[DECODE] %s 0x%X %s: %s", bus, canID, status, note)
+	}
+}
+
+// InitDecodeReporting starts the rollup. Runs for the life of the process.
+func InitDecodeReporting() {
+	go func() {
+		ticker := time.NewTicker(decodeReportInterval)
+		defer ticker.Stop()
+		for range ticker.C {
+			reportDecodeOutcomes()
+		}
+	}()
+}
+
+func reportDecodeOutcomes() {
+	decodeMu.Lock()
+	ok, failed := decodedOK, decodedFailed
+	decodedOK, decodedFailed = 0, 0
+
+	type entry struct {
+		key   decodeKey
+		count int64
+	}
+	entries := make([]entry, 0, len(decodeStats))
+	for k, s := range decodeStats {
+		if s.sinceReport > 0 {
+			entries = append(entries, entry{k, s.sinceReport})
+			s.sinceReport = 0
+		}
+	}
+	decodeMu.Unlock()
+
+	if ok == 0 && failed == 0 {
+		return
+	}
+
+	// Loudest offenders first — that is what you act on.
+	sort.Slice(entries, func(i, j int) bool { return entries[i].count > entries[j].count })
+
+	var b strings.Builder
+	fmt.Fprintf(&b, "[DECODE] %s: %d decoded, %d undecodable", decodeReportInterval, ok, failed)
+	if len(entries) > 0 {
+		b.WriteString(" —")
+		for i, e := range entries {
+			if i == maxReportedKinds {
+				fmt.Fprintf(&b, " (+%d more kinds)", len(entries)-i)
+				break
+			}
+			fmt.Fprintf(&b, " %s/0x%X %s ×%d", e.key.bus, e.key.canID, e.key.status, e.count)
+		}
+	}
+
+	if failed > 0 {
+		logger.SugarLogger.Warnln(b.String())
+		return
+	}
+	logger.SugarLogger.Infoln(b.String())
+}

p987/service/message.go

@@ -16,6 +16,15 @@ import (
 	mapache "github.com/gaucho-racing/mapache/mapache-go/v3"
 )
 
+// Decode outcome statuses, recorded in the stored frame's metadata and
+// counted by the decode reporter.
+const (
+	statusOK               = "ok"
+	statusUnknownCANID     = "unknown_can_id"
+	statusDecodeError      = "decode_error"
+	statusInvalidTimestamp = "invalid_timestamp"
+)
+
 // headerSize is the relay's wire format: u64 BE microsecond timestamp
 // followed by a u16 BE upload key, then the raw CAN payload.
 const headerSize = 10
@@ -86,13 +95,10 @@ func ProcessFrame(vehicleID, bus string, canID, timestamp int, data []byte) (mod
 
 	switch {
 	case !IsValidProducedAt(timestamp):
-		logger.SugarLogger.Warnf("Frame with invalid timestamp: vehicle=%s bus=%s can_id=0x%X ts=%d decoded=%s",
-			vehicleID, bus, canID, timestamp, producedAt.UTC().Format(time.RFC3339Nano))
-		meta = MustJSON(map[string]any{
-			"status": "invalid_timestamp",
-			"note": fmt.Sprintf("ts=%d (%s) is before %s", timestamp,
-				producedAt.UTC().Format(time.RFC3339Nano), minValidProducedAt.UTC().Format(time.RFC3339Nano)),
-		})
+		note := fmt.Sprintf("ts=%d (%s) is before %s", timestamp,
+			producedAt.UTC().Format(time.RFC3339Nano), minValidProducedAt.UTC().Format(time.RFC3339Nano))
+		NoteDecodeOutcome(bus, canID, statusInvalidTimestamp, note)
+		meta = MustJSON(map[string]any{"status": statusInvalidTimestamp, "note": note})
 	default:
 		decoded, meta = decodeFrame(bus, canID, data)
 	}
@@ -130,18 +136,17 @@ func ProcessFrame(vehicleID, bus string, canID, timestamp int, data []byte) (mod
 func decodeFrame(bus string, canID int, data []byte) ([]mapache.Signal, []byte) {
 	messageStruct := model.GetMessage(bus, canID)
 	if messageStruct == nil {
-		return nil, MustJSON(map[string]any{
-			"status": "unknown_can_id",
-			"note":   fmt.Sprintf("no decoder registered for can id 0x%X on bus %s", canID, bus),
-		})
+		note := fmt.Sprintf("no decoder registered for can id 0x%X on bus %s", canID, bus)
+		NoteDecodeOutcome(bus, canID, statusUnknownCANID, note)
+		return nil, MustJSON(map[string]any{"status": statusUnknownCANID, "note": note})
 	}
 	if err := messageStruct.FillFromBytes(data); err != nil {
-		return nil, MustJSON(map[string]any{
-			"status": "decode_error",
-			"note":   err.Error(),
-		})
+		note := err.Error()
+		NoteDecodeOutcome(bus, canID, statusDecodeError, note)
+		return nil, MustJSON(map[string]any{"status": statusDecodeError, "note": note})
 	}
-	return messageStruct.ExportSignals(), MustJSON(map[string]any{"status": "ok"})
+	NoteDecodeOutcome(bus, canID, statusOK, "")
+	return messageStruct.ExportSignals(), MustJSON(map[string]any{"status": statusOK})
 }
 
 func HandleMessage(vehicleID string, bus string, canID int, message []byte) {