feat(go): expose history batch pending metrics

go/vehicle-gateway/cmd/history-writer/main.go

@@ -174,6 +174,8 @@ func processHistoryBatch(ctx context.Context, logger interface {
 		envelopes = append(envelopes, env)
 	}
 	if len(envelopes) > 0 {
+		setBatchPending(registry, len(messages), len(envelopes))
+		defer setBatchPending(registry, 0, 0)
 		started := time.Now()
 		err := appender.AppendAllBatch(messageCtx, envelopes)
 		elapsed := time.Since(started)
@@ -230,6 +232,14 @@ func setBatchDuration(registry *metrics.Registry, status string, elapsed time.Du
 	registry.SetGauge("vehicle_history_batch_flush_duration_ms", metrics.Labels{"status": status}, float64(elapsed.Milliseconds()))
 }
 
+func setBatchPending(registry *metrics.Registry, messages int, rows int) {
+	if registry == nil {
+		return
+	}
+	registry.SetGauge("vehicle_history_batch_pending_messages", nil, float64(messages))
+	registry.SetGauge("vehicle_history_batch_pending_rows", nil, float64(rows))
+}
+
 func addWriterLagMetric(registry *metrics.Registry, message kafka.Message) {
 	if registry == nil {
 		return

go/vehicle-gateway/cmd/history-writer/main_test.go

@@ -125,6 +125,55 @@ func TestProcessHistoryBatchAppendsAllBeforeCommit(t *testing.T) {
 	}
 }
 
+func TestProcessHistoryBatchExposesPendingBatchDepthDuringAppend(t *testing.T) {
+	first := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
+	second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
+	firstPayload, err := json.Marshal(first)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondPayload, err := json.Marshal(second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	registry := metrics.NewRegistry()
+	appender := &contextCheckingHistoryAppender{
+		onAppendBatch: func() {
+			text := registry.Render()
+			for _, want := range []string{
+				`vehicle_history_batch_pending_messages 2`,
+				`vehicle_history_batch_pending_rows 2`,
+			} {
+				if !strings.Contains(text, want) {
+					t.Fatalf("pending batch metric missing %s while appending:\n%s", want, text)
+				}
+			}
+		},
+	}
+
+	processHistoryBatch(
+		context.Background(),
+		discardHistoryLogger{},
+		registry,
+		appender,
+		&contextCheckingHistoryCommitter{},
+		[]kafka.Message{
+			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 13, Value: firstPayload},
+			{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 20, HighWaterMark: 21, Value: secondPayload},
+		},
+	)
+
+	text := registry.Render()
+	for _, want := range []string{
+		`vehicle_history_batch_pending_messages 0`,
+		`vehicle_history_batch_pending_rows 0`,
+	} {
+		if !strings.Contains(text, want) {
+			t.Fatalf("pending batch metric should reset after append, missing %s:\n%s", want, text)
+		}
+	}
+}
+
 func TestProcessHistoryBatchDoesNotCommitWhenAppendFails(t *testing.T) {
 	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
 	payload, err := json.Marshal(env)
@@ -189,11 +238,12 @@ func TestLoadConfigReadsBatchSettings(t *testing.T) {
 }
 
 type contextCheckingHistoryAppender struct {
-	ctxErr     error
-	count      int
-	batchCount int
-	batch      []envelope.FrameEnvelope
-	err        error
+	ctxErr        error
+	count         int
+	batchCount    int
+	batch         []envelope.FrameEnvelope
+	err           error
+	onAppendBatch func()
 }
 
 func (a *contextCheckingHistoryAppender) AppendAll(ctx context.Context, _ envelope.FrameEnvelope) error {
@@ -209,6 +259,9 @@ func (a *contextCheckingHistoryAppender) AppendAllBatch(ctx context.Context, env
 	a.ctxErr = ctx.Err()
 	a.batchCount++
 	a.batch = append([]envelope.FrameEnvelope(nil), envs...)
+	if a.onAppendBatch != nil {
+		a.onAppendBatch()
+	}
 	if a.ctxErr != nil {
 		return a.ctxErr
 	}