fix: Use topic instead of session_id as the Prometheus label

CHANGES.md

@@ -34,6 +34,7 @@ Release Notes.
 
 ### Bug Fixes
 
+- Use `topic` instead of `session_id` as the Prometheus label on liaison `queue_sub` chunk-ordering counters to avoid unbounded metric cardinality.
 - Fix flaky trace query filtering caused by non-deterministic sidx tag ordering and add consistency checks for integration query cases.
 - Fix index-mode measure queries returning documents outside the requested time range when a widened segment overlaps the query window.
 - MCP: Add validation for properties and harden the mcp server.

banyand/queue/sub/chunked_sync.go

@@ -159,7 +159,7 @@ func (s *server) SyncPart(stream clusterv1.ChunkedSyncService_SyncPartServer) er
 				if currentSession.partCtx != nil {
 					if currentSession.partCtx.Handler != nil {
 						if finishErr := currentSession.partCtx.Handler.FinishSync(); finishErr != nil {
-							s.updateChunkOrderMetrics("finish_sync_err", currentSession.sessionID)
+							s.updateChunkOrderMetrics("finish_sync_err", currentSession.metadata.Topic)
 							s.log.Error().Err(finishErr).Str("session_id", currentSession.sessionID).Msg("failed to finish sync for previous session")
 						}
 						if closeErr := currentSession.partCtx.Close(); closeErr != nil {
@@ -253,6 +253,10 @@ func (s *server) processChunkSequential(stream clusterv1.ChunkedSyncService_Sync
 
 func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
 	buffer := session.chunkBuffer
+	// must check buffer timeout before refreshing lastActivity, otherwise it will never timeout.
+	if err := s.checkBufferTimeout(session); err != nil {
+		return err
+	}
 	buffer.lastActivity = time.Now()
 
 	if req.ChunkIndex == buffer.expectedIndex {
@@ -266,7 +270,7 @@ func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_
 
 	if req.ChunkIndex > buffer.expectedIndex {
 		gap := req.ChunkIndex - buffer.expectedIndex
-		s.updateChunkOrderMetrics("out_of_order_received", req.SessionId)
+		s.updateChunkOrderMetrics("out_of_order_received", session.metadata.Topic)
 
 		if gap > s.maxChunkGapSize {
 			errMsg := fmt.Sprintf("chunk gap too large: expected %d, got %d (gap: %d > max: %d)",
@@ -277,7 +281,7 @@ func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_
 				Uint32("gap", gap).
 				Uint32("max_gap", s.maxChunkGapSize).
 				Msg("chunk gap too large, rejecting")
-			s.updateChunkOrderMetrics("gap_too_large", req.SessionId)
+			s.updateChunkOrderMetrics("gap_too_large", session.metadata.Topic)
 			return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
 		}
 
@@ -288,7 +292,7 @@ func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_
 				Uint32("buffer_size", uint32(len(buffer.chunks))).
 				Uint32("max_buffer_size", buffer.maxBufferSize).
 				Msg("chunk buffer full, rejecting chunk")
-			s.updateChunkOrderMetrics("buffer_full", req.SessionId)
+			s.updateChunkOrderMetrics("buffer_full", session.metadata.Topic)
 			return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
 		}
 
@@ -301,7 +305,7 @@ func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_
 				Uint32("buffered_chunks", uint32(len(buffer.chunks))).
 				Msg("buffered out-of-order chunk")
 		}
-		s.updateChunkOrderMetrics("chunk_buffered", req.SessionId)
+		s.updateChunkOrderMetrics("chunk_buffered", session.metadata.Topic)
 
 		return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED,
 			fmt.Sprintf("chunk %d buffered (waiting for %d)", req.ChunkIndex, buffer.expectedIndex), nil)
@@ -444,7 +448,7 @@ func (s *server) checkBufferTimeout(session *syncSession) error {
 					missing = append(missing, i)
 				}
 			}
-
+			s.updateChunkOrderMetrics("buffer_timeout", session.metadata.Topic)
 			return fmt.Errorf("buffer timeout: missing chunks %v after %v",
 				missing, session.chunkBuffer.bufferTimeout)
 		}

banyand/queue/sub/chunked_sync_test.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"hash/crc32"
+	"sync"
 	"testing"
 	"time"
 
@@ -120,6 +121,9 @@ func TestChunkedSyncOutOfOrderHandling(t *testing.T) {
 				enableChunkReordering: tt.enableChunkReordering,
 				maxChunkBufferSize:    tt.maxChunkBufferSize,
 				maxChunkGapSize:       tt.maxChunkGapSize,
+				// Avoid accidental immediate buffer-timeout when bufferTimeout is zero.
+				// Some tests buffer out-of-order chunks and expect no error.
+				chunkBufferTimeout: time.Hour,
 			}
 
 			// Register a mock handler
@@ -189,6 +193,127 @@ func TestChunkedSyncOutOfOrderHandling(t *testing.T) {
 	}
 }
 
+type capturingCounter struct {
+	labelValues [][]string
+	mu          sync.Mutex
+}
+
+func (c *capturingCounter) Inc(_ float64, labelValues ...string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	cp := make([]string, len(labelValues))
+	copy(cp, labelValues)
+	c.labelValues = append(c.labelValues, cp)
+}
+
+func (c *capturingCounter) Delete(_ ...string) bool {
+	return true
+}
+
+func (c *capturingCounter) uniqueFirstLabelValues() map[string]struct{} {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	m := make(map[string]struct{})
+	for _, lv := range c.labelValues {
+		if len(lv) > 0 {
+			m[lv[0]] = struct{}{}
+		}
+	}
+	return m
+}
+
+func TestChunkOrderingMetricsAreLabeledByTopic_NotSessionID(t *testing.T) {
+	// enable reordering, otherwise the chunk-ordering metrics will not be triggered.
+	s := &server{
+		log:                   logger.GetLogger("test-server-metrics-label"),
+		chunkedSyncHandlers:   make(map[bus.Topic]queue.ChunkedSyncHandler),
+		enableChunkReordering: true,
+		maxChunkBufferSize:    10,
+		maxChunkGapSize:       5,
+	}
+
+	// handler: avoid "no handler registered" in processExpectedChunk.
+	mockHandler := &MockChunkedSyncHandler{}
+	s.chunkedSyncHandlers[data.TopicStreamPartSync] = mockHandler
+
+	// metrics: this test will trigger at least two events:
+	// - out_of_order_received
+	// - chunk_buffered
+	// so must put both counters, otherwise nil.Inc will panic.
+	outOfOrder := &capturingCounter{}
+	buffered := &capturingCounter{}
+	s.metrics = &metrics{
+		outOfOrderChunksReceived: outOfOrder,
+		chunksBuffered:           buffered,
+		// other counters will not be triggered in this test, leave them nil
+	}
+
+	topic := data.TopicStreamPartSync.String()
+
+	drive := func(sessionID string) {
+		mockStream := &MockSyncPartStream{}
+		session := &syncSession{
+			sessionID:      sessionID,
+			startTime:      time.Now(),
+			chunksReceived: 0,
+			partsProgress:  make(map[int]*partProgress),
+			metadata: &clusterv1.SyncMetadata{
+				Group: "test-group",
+				Topic: topic,
+			},
+		}
+
+		// send chunk 0 (establish buffer.expectedIndex=1)
+		req0 := &clusterv1.SyncPartRequest{
+			SessionId:     sessionID,
+			ChunkIndex:    0,
+			ChunkData:     []byte("chunk-0"),
+			ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE([]byte("chunk-0"))),
+			PartsInfo: []*clusterv1.PartInfo{
+				{Id: 1, Files: []*clusterv1.FileInfo{{Name: "f", Offset: 0, Size: 7}}},
+			},
+		}
+		require.NoError(t, s.processChunk(mockStream, session, req0))
+
+		// send chunk 2 (out-of-order: expected 1 got 2),
+		// will trigger out_of_order_received + chunk_buffered.
+		req2 := &clusterv1.SyncPartRequest{
+			SessionId:     sessionID,
+			ChunkIndex:    2,
+			ChunkData:     []byte("chunk-2"),
+			ChunkChecksum: fmt.Sprintf("%x", crc32.ChecksumIEEE([]byte("chunk-2"))),
+			PartsInfo: []*clusterv1.PartInfo{
+				{Id: 2, Files: []*clusterv1.FileInfo{{Name: "f", Offset: 0, Size: 7}}},
+			},
+		}
+		require.NoError(t, s.processChunk(mockStream, session, req2))
+	}
+
+	drive("test-session-A")
+	drive("test-session-B")
+
+	// assert: labelValues[0] must be topic; and unique label must be only one (topic)
+	uniqOut := outOfOrder.uniqueFirstLabelValues()
+	uniqBuf := buffered.uniqueFirstLabelValues()
+
+	assert.Equal(t, 1, len(uniqOut))
+	assert.Equal(t, 1, len(uniqBuf))
+
+	_, okOut := uniqOut[topic]
+	_, okBuf := uniqBuf[topic]
+	assert.True(t, okOut, "out_of_order_received label must be topic")
+	assert.True(t, okBuf, "chunk_buffered label must be topic")
+
+	// assert: never should have session_id as label
+	_, bad1 := uniqOut["test-session-A"]
+	_, bad2 := uniqOut["test-session-B"]
+	_, bad3 := uniqBuf["test-session-A"]
+	_, bad4 := uniqBuf["test-session-B"]
+	assert.False(t, bad1 || bad2 || bad3 || bad4, "metrics must not be labeled by session_id")
+}
+
 // MockChunkedSyncHandler implements queue.ChunkedSyncHandler for testing.
 type MockChunkedSyncHandler struct{}
 
@@ -264,6 +389,10 @@ func TestChunkedSyncBufferTimeout(t *testing.T) {
 	session := &syncSession{
 		sessionID: "test-session-timeout",
 		startTime: time.Now(),
+		metadata: &clusterv1.SyncMetadata{
+			Topic: data.TopicStreamPartSync.String(),
+			Group: "test-group",
+		},
 		chunkBuffer: &chunkBuffer{
 			chunks:        make(map[uint32]*clusterv1.SyncPartRequest),
 			expectedIndex: 1, // Waiting for chunk 1

banyand/queue/sub/server.go

@@ -435,32 +435,32 @@ func newMetrics(factory observability.Factory) *metrics {
 		totalMsgSentErr:     factory.NewCounter("total_msg_sent_err", "topic"),
 
 		// Chunk ordering metrics
-		outOfOrderChunksReceived: factory.NewCounter("out_of_order_chunks_received", "session_id"),
-		chunksBuffered:           factory.NewCounter("chunks_buffered", "session_id"),
-		bufferTimeouts:           factory.NewCounter("buffer_timeouts", "session_id"),
-		largeGapsRejected:        factory.NewCounter("large_gaps_rejected", "session_id"),
-		bufferCapacityExceeded:   factory.NewCounter("buffer_capacity_exceeded", "session_id"),
-		finishSyncErr:            factory.NewCounter("finish_sync_err", "session_id"),
+		outOfOrderChunksReceived: factory.NewCounter("out_of_order_chunks_received", "topic"),
+		chunksBuffered:           factory.NewCounter("chunks_buffered", "topic"),
+		bufferTimeouts:           factory.NewCounter("buffer_timeouts", "topic"),
+		largeGapsRejected:        factory.NewCounter("large_gaps_rejected", "topic"),
+		bufferCapacityExceeded:   factory.NewCounter("buffer_capacity_exceeded", "topic"),
+		finishSyncErr:            factory.NewCounter("finish_sync_err", "topic"),
 	}
 }
 
 // updateChunkOrderMetrics updates chunk ordering metrics.
-func (s *server) updateChunkOrderMetrics(event, sessionID string) {
+func (s *server) updateChunkOrderMetrics(event, topic string) {
 	if s.metrics == nil {
 		return // Skip metrics if not initialized (e.g., during tests)
 	}
 	switch event {
 	case "out_of_order_received":
-		s.metrics.outOfOrderChunksReceived.Inc(1, sessionID)
+		s.metrics.outOfOrderChunksReceived.Inc(1, topic)
 	case "chunk_buffered":
-		s.metrics.chunksBuffered.Inc(1, sessionID)
+		s.metrics.chunksBuffered.Inc(1, topic)
 	case "buffer_timeout":
-		s.metrics.bufferTimeouts.Inc(1, sessionID)
+		s.metrics.bufferTimeouts.Inc(1, topic)
 	case "gap_too_large":
-		s.metrics.largeGapsRejected.Inc(1, sessionID)
+		s.metrics.largeGapsRejected.Inc(1, topic)
 	case "buffer_full":
-		s.metrics.bufferCapacityExceeded.Inc(1, sessionID)
+		s.metrics.bufferCapacityExceeded.Inc(1, topic)
 	case "finish_sync_err":
-		s.metrics.finishSyncErr.Inc(1, sessionID)
+		s.metrics.finishSyncErr.Inc(1, topic)
 	}
 }