(skywalking-banyandb) 07/11: Enhance error handling in batch publisher and circuit breaker logic

[ha...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hanahmily pushed a commit to branch pub/perf
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git

commit bab4865349c525f3a9c3f85a3688b1d47d05eb9a
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 17:39:38 2025 +0800

    Enhance error handling in batch publisher and circuit breaker logic
    
    - Added failure recording in the batch publisher for circuit breaker events, improving diagnostics during failover scenarios.
    - Updated the `isTransientError` and `isInternalError` functions to handle `common.Error` types, ensuring consistent error classification.
---
 banyand/queue/pub/batch.go               |  6 ++++
 banyand/queue/pub/circuitbreaker_test.go | 51 ++++++++++++++++++++++++++++++++
 banyand/queue/pub/client.go              | 21 +++++++++++++
 3 files changed, 78 insertions(+)

diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index f1cacc94..127bec11 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -172,6 +172,8 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
 			resp, errRecv := s.Recv()
 			if errRecv != nil {
 				if isFailoverError(errRecv) {
+					// Record circuit breaker failure before creating failover event
+					bp.pub.recordFailure(curNode, errRecv)
 					bc <- batchEvent{n: curNode, e: common.NewErrorWithStatus(modelv1.Status_STATUS_INTERNAL_ERROR, errRecv.Error())}
 				}
 				return
@@ -184,6 +186,8 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
 			}
 			if isFailoverStatus(resp.Status) {
 				ce := common.NewErrorWithStatus(resp.Status, resp.Error)
+				// Record circuit breaker failure before creating failover event
+				bp.pub.recordFailure(curNode, ce)
 				bc <- batchEvent{n: curNode, e: ce}
 			}
 		}(stream, deferFn, bp.f.events[len(bp.f.events)-1], nodeName)
@@ -206,6 +210,8 @@ func (bp *batchPublisher) Close() (cee map[string]*common.Error, err error) {
 		go func() {
 			defer bp.pub.closer.Done()
 			for n, e := range batchEvents {
+				// Record circuit breaker failure before failover
+				bp.pub.recordFailure(n, e.e)
 				if bp.topic == nil {
 					bp.pub.failover(n, e.e, data.TopicCommon)
 					continue
diff --git a/banyand/queue/pub/circuitbreaker_test.go b/banyand/queue/pub/circuitbreaker_test.go
index b22fc06e..8bc90223 100644
--- a/banyand/queue/pub/circuitbreaker_test.go
+++ b/banyand/queue/pub/circuitbreaker_test.go
@@ -27,6 +27,8 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/apache/skywalking-banyandb/api/common"
+	modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
 	"github.com/apache/skywalking-banyandb/pkg/logger"
 )
 
@@ -630,3 +632,52 @@ func TestCircuitBreakerErrorFiltering(t *testing.T) {
 	assert.Equal(t, StateOpen, cb.state)
 	assert.Equal(t, defaultCBThreshold, cb.consecutiveFailures)
 }
+
+func TestCircuitBreakerFailoverErrors(t *testing.T) {
+	p := &pub{
+		cbStates: make(map[string]*circuitState),
+		cbMu:     sync.RWMutex{},
+		log:      logger.GetLogger("test"),
+	}
+
+	node := testNodeName
+
+	// Test failover errors from Recv() - using codes.Unavailable which is a failover error
+	failoverErr := status.Error(codes.Unavailable, "service unavailable")
+
+	// Verify initial state is closed
+	assert.True(t, p.isRequestAllowed(node))
+
+	// Simulate failover errors that should increment circuit breaker
+	for i := 0; i < defaultCBThreshold; i++ {
+		p.recordFailure(node, failoverErr)
+	}
+
+	// Circuit should be open after failover errors
+	assert.False(t, p.isRequestAllowed(node))
+	p.cbMu.Lock()
+	cb, exists := p.cbStates[node]
+	p.cbMu.Unlock()
+	assert.True(t, exists)
+	assert.Equal(t, StateOpen, cb.state)
+	assert.Equal(t, defaultCBThreshold, cb.consecutiveFailures)
+
+	// Reset and test with common.Error (simulating Close() method errors)
+	p.recordSuccess(node)
+	assert.True(t, p.isRequestAllowed(node))
+
+	// Test with common.Error types that would come from Close() method
+	internalErr := common.NewErrorWithStatus(modelv1.Status_STATUS_INTERNAL_ERROR, "internal error")
+	for i := 0; i < defaultCBThreshold; i++ {
+		p.recordFailure(node, internalErr)
+	}
+
+	// Circuit should be open again
+	assert.False(t, p.isRequestAllowed(node))
+	p.cbMu.Lock()
+	cb, exists = p.cbStates[node]
+	p.cbMu.Unlock()
+	assert.True(t, exists)
+	assert.Equal(t, StateOpen, cb.state)
+	assert.Equal(t, defaultCBThreshold, cb.consecutiveFailures)
+}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 2b090f0f..482df7a0 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -20,6 +20,7 @@ package pub
 import (
 	"context"
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"math/big"
 	"time"
@@ -400,10 +401,23 @@ func isTransientError(err error) bool {
 		return false
 	}
 
+	// Handle gRPC status errors
 	if s, ok := status.FromError(err); ok {
 		return retryableCodes[s.Code()]
 	}
 
+	// Handle common.Error types
+	var ce *common.Error
+	if errors.As(err, &ce) {
+		// Map common status to gRPC codes for consistency
+		switch ce.Status() {
+		case modelv1.Status_STATUS_INTERNAL_ERROR:
+			return retryableCodes[codes.Internal]
+		default:
+			return false
+		}
+	}
+
 	return false
 }
 
@@ -413,10 +427,17 @@ func isInternalError(err error) bool {
 		return false
 	}
 
+	// Handle gRPC status errors
 	if s, ok := status.FromError(err); ok {
 		return s.Code() == codes.Internal
 	}
 
+	// Handle common.Error types
+	var ce *common.Error
+	if errors.As(err, &ce) {
+		return ce.Status() == modelv1.Status_STATUS_INTERNAL_ERROR
+	}
+
 	return false
 }