You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2025/09/04 12:17:45 UTC
(skywalking-banyandb) branch pub/perf created (now 1a4f4482)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch pub/perf
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 1a4f4482 Update CHANGES.md
This branch includes the following new commits:
new 3d162cbb Refactor batch publisher and client health check logic
new d22ec2fc Add circuit breaker implementation for client health management
new aa82e3fb Enhance circuit breaker logic in client health management
new d0483e2a Implement retry logic and circuit breaker enhancements for message publishing
new 971f583e Refactor circuit breaker tests and enhance probe logic
new 81fbc321 Enhance circuit breaker logic to record transient and internal errors
new bab48653 Enhance error handling in batch publisher and circuit breaker logic
new a23fdb91 Enhance circuit breaker logic in message publishing
new 110fd094 Refactor retry policy handling in pub client
new 0eb6b7bd Refactor pub client by removing unused circuit breaker and retry logic
new 1a4f4482 Update CHANGES.md
The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 10/11: Refactor pub client by removing unused circuit breaker and retry logic
Posted by 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 0eb6b7bd1815ecdfc602ba0c8ab03a63acaa0271
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 20:05:42 2025 +0800
Refactor pub client by removing unused circuit breaker and retry logic
---
banyand/queue/pub/circuitbreaker.go | 147 +++++++++++++++++++++
banyand/queue/pub/client.go | 254 +-----------------------------------
banyand/queue/pub/pub_test.go | 2 +-
banyand/queue/pub/retry.go | 161 +++++++++++++++++++++++
4 files changed, 310 insertions(+), 254 deletions(-)
diff --git a/banyand/queue/pub/circuitbreaker.go b/banyand/queue/pub/circuitbreaker.go
new file mode 100644
index 00000000..7bb64e29
--- /dev/null
+++ b/banyand/queue/pub/circuitbreaker.go
@@ -0,0 +1,147 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pub
+
+import (
+ "time"
+)
+
+const (
+ defaultCBThreshold = 5
+ defaultCBResetTimeout = 60 * time.Second
+)
+
+// CircuitState defines the circuit breaker states.
+type CircuitState int
+
+// CircuitState defines the circuit breaker states.
+const (
+ StateClosed CircuitState = iota // Normal operation
+ StateOpen // Reject requests until cooldown expires
+ StateHalfOpen // Allow a single probe
+)
+
+// circuitState holds circuit breaker metadata; it does NOT duplicate gRPC clients/conns.
+type circuitState struct {
+ lastFailureTime time.Time
+ openTime time.Time
+ state CircuitState
+ consecutiveFailures int
+ halfOpenProbeInFlight bool
+}
+
+// isRequestAllowed checks if a request to the given node is allowed based on circuit breaker state.
+// It also handles state transitions from Open to Half-Open when cooldown expires.
+func (p *pub) isRequestAllowed(node string) bool {
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ return true // No circuit breaker state, allow request
+ }
+
+ switch cb.state {
+ case StateClosed:
+ return true
+ case StateOpen:
+ // Check if cooldown period has expired
+ if time.Since(cb.openTime) >= defaultCBResetTimeout {
+ // Transition to Half-Open to allow a single probe request
+ cb.state = StateHalfOpen
+ cb.halfOpenProbeInFlight = true // Set token for the probe
+ p.log.Info().Str("node", node).Msg("circuit breaker transitioned to half-open")
+ return true
+ }
+ return false // Still in cooldown period
+ case StateHalfOpen:
+ // In half-open state, deny requests if probe is already in flight
+ if cb.halfOpenProbeInFlight {
+ return false // Probe already in progress, deny additional requests
+ }
+ // This case should not normally happen since we set the token on transition,
+ // but handle it defensively by allowing the request and setting the token
+ cb.halfOpenProbeInFlight = true
+ return true
+ default:
+ return true
+ }
+}
+
+// recordSuccess resets the circuit breaker state to Closed on successful operation.
+// This handles Half-Open -> Closed transitions.
+func (p *pub) recordSuccess(node string) {
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ // Initialize circuit breaker state
+ p.cbStates[node] = &circuitState{
+ state: StateClosed,
+ consecutiveFailures: 0,
+ }
+ return
+ }
+
+ // Reset to closed state
+ cb.state = StateClosed
+ cb.consecutiveFailures = 0
+ cb.lastFailureTime = time.Time{}
+ cb.openTime = time.Time{}
+ cb.halfOpenProbeInFlight = false // Clear probe token
+}
+
+// recordFailure updates the circuit breaker state on failed operation.
+// Only records failures for transient/internal errors that should count toward opening the circuit.
+func (p *pub) recordFailure(node string, err error) {
+ // Only record failure if the error is transient or internal
+ if !isTransientError(err) && !isInternalError(err) {
+ return
+ }
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ // Initialize circuit breaker state
+ cb = &circuitState{
+ state: StateClosed,
+ consecutiveFailures: 1,
+ lastFailureTime: time.Now(),
+ }
+ p.cbStates[node] = cb
+ } else {
+ cb.consecutiveFailures++
+ cb.lastFailureTime = time.Now()
+ }
+
+ // Check if we should open the circuit
+ threshold := defaultCBThreshold
+ if cb.consecutiveFailures >= threshold && cb.state == StateClosed {
+ cb.state = StateOpen
+ cb.openTime = time.Now()
+ p.log.Warn().Str("node", node).Int("failures", cb.consecutiveFailures).Msg("circuit breaker opened")
+ } else if cb.state == StateHalfOpen {
+ // Failed during half-open, go back to open
+ cb.state = StateOpen
+ cb.openTime = time.Now()
+ cb.halfOpenProbeInFlight = false // Clear probe token
+ p.log.Warn().Str("node", node).Msg("circuit breaker reopened after half-open failure")
+ }
+}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 74543ec4..bcca7868 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -19,16 +19,11 @@ package pub
import (
"context"
- "crypto/rand"
- "errors"
"fmt"
- "math/big"
"time"
"google.golang.org/grpc"
- "google.golang.org/grpc/codes"
"google.golang.org/grpc/health/grpc_health_v1"
- "google.golang.org/grpc/status"
"github.com/apache/skywalking-banyandb/api/common"
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
@@ -41,21 +36,10 @@ import (
)
const (
- rpcTimeout = 2 * time.Second
- defaultJitterFactor = 0.2
- defaultMaxRetries = 3
- defaultCBThreshold = 5
- defaultCBResetTimeout = 60 * time.Second
- defaultPerRequestTimeout = 2 * time.Second
- defaultBackoffBase = 500 * time.Millisecond
- defaultBackoffMax = 30 * time.Second
+ rpcTimeout = 2 * time.Second
)
var (
- // Retry policy for health check.
- initBackoff = time.Second
- maxBackoff = 20 * time.Second
-
// The timeout is set by each RPC.
retryPolicy = `{
"methodConfig": [
@@ -126,48 +110,8 @@ var (
}
}
]}`
-
- // Retryable gRPC status codes for streaming send retries.
- retryableCodes = map[codes.Code]bool{
- codes.OK: false,
- codes.Canceled: false,
- codes.Unknown: false,
- codes.InvalidArgument: false,
- codes.DeadlineExceeded: true, // Retryable - operation exceeded deadline
- codes.NotFound: false,
- codes.AlreadyExists: false,
- codes.PermissionDenied: false,
- codes.ResourceExhausted: true, // Retryable - server resource limits exceeded
- codes.FailedPrecondition: false,
- codes.Aborted: false,
- codes.OutOfRange: false,
- codes.Unimplemented: false,
- codes.Internal: true, // Retryable - internal server error should participate in circuit breaker
- codes.Unavailable: true, // Retryable - service temporarily unavailable
- codes.DataLoss: false,
- codes.Unauthenticated: false,
- }
-)
-
-// CircuitState defines the circuit breaker states.
-type CircuitState int
-
-// CircuitState defines the circuit breaker states.
-const (
- StateClosed CircuitState = iota // Normal operation
- StateOpen // Reject requests until cooldown expires
- StateHalfOpen // Allow a single probe
)
-// circuitState holds circuit breaker metadata; it does NOT duplicate gRPC clients/conns.
-type circuitState struct {
- lastFailureTime time.Time
- openTime time.Time
- state CircuitState
- consecutiveFailures int
- halfOpenProbeInFlight bool
-}
-
type client struct {
client clusterv1.ServiceClient
conn *grpc.ClientConn
@@ -346,101 +290,6 @@ func (p *pub) removeNodeIfUnhealthy(md schema.Metadata, node *databasev1.Node, c
return true
}
-// secureRandFloat64 generates a cryptographically secure random float64 in [0, 1).
-func secureRandFloat64() float64 {
- // Generate a random uint64
- maxVal := big.NewInt(1 << 53) // Use 53 bits for precision similar to math/rand
- n, err := rand.Int(rand.Reader, maxVal)
- if err != nil {
- // Fallback to a reasonable value if crypto/rand fails
- return 0.5
- }
- return float64(n.Uint64()) / float64(1<<53)
-}
-
-// jitteredBackoff calculates backoff duration with jitter to avoid thundering herds.
-// Uses bounded symmetric jitter: backoff * (1 + jitter * (rand() - 0.5) * 2).
-func jitteredBackoff(baseBackoff, maxBackoff time.Duration, attempt int, jitterFactor float64) time.Duration {
- if jitterFactor < 0 {
- jitterFactor = 0
- }
- if jitterFactor > 1 {
- jitterFactor = 1
- }
-
- // Exponential backoff: base * 2^attempt
- backoff := baseBackoff
- for i := 0; i < attempt; i++ {
- backoff *= 2
- if backoff > maxBackoff {
- backoff = maxBackoff
- break
- }
- }
-
- // Apply jitter: backoff * (1 + jitter * (rand() - 0.5) * 2)
- // This gives us a range of [backoff * (1-jitter), backoff * (1+jitter)]
- jitterRange := float64(backoff) * jitterFactor
- randomFloat := secureRandFloat64()
- randomOffset := (randomFloat - 0.5) * 2 * jitterRange
-
- jitteredDuration := time.Duration(float64(backoff) + randomOffset)
- if jitteredDuration < 0 {
- jitteredDuration = baseBackoff / 10 // Minimum backoff
- }
- if jitteredDuration > maxBackoff {
- jitteredDuration = maxBackoff
- }
-
- return jitteredDuration
-}
-
-// isTransientError checks if the error is considered transient and retryable.
-func isTransientError(err error) bool {
- if err == nil {
- 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
-}
-
-// isInternalError checks if the error is an internal server error.
-func isInternalError(err error) bool {
- if err == nil {
- 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
-}
-
func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Metadata) bool {
node, ok := md.Spec.(*databasev1.Node)
if !ok {
@@ -671,104 +520,3 @@ type evictNode struct {
n *databasev1.Node
c chan struct{}
}
-
-// isRequestAllowed checks if a request to the given node is allowed based on circuit breaker state.
-// It also handles state transitions from Open to Half-Open when cooldown expires.
-func (p *pub) isRequestAllowed(node string) bool {
- p.cbMu.Lock()
- defer p.cbMu.Unlock()
-
- cb, exists := p.cbStates[node]
- if !exists {
- return true // No circuit breaker state, allow request
- }
-
- switch cb.state {
- case StateClosed:
- return true
- case StateOpen:
- // Check if cooldown period has expired
- if time.Since(cb.openTime) >= defaultCBResetTimeout {
- // Transition to Half-Open to allow a single probe request
- cb.state = StateHalfOpen
- cb.halfOpenProbeInFlight = true // Set token for the probe
- p.log.Info().Str("node", node).Msg("circuit breaker transitioned to half-open")
- return true
- }
- return false // Still in cooldown period
- case StateHalfOpen:
- // In half-open state, deny requests if probe is already in flight
- if cb.halfOpenProbeInFlight {
- return false // Probe already in progress, deny additional requests
- }
- // This case should not normally happen since we set the token on transition,
- // but handle it defensively by allowing the request and setting the token
- cb.halfOpenProbeInFlight = true
- return true
- default:
- return true
- }
-}
-
-// recordSuccess resets the circuit breaker state to Closed on successful operation.
-// This handles Half-Open -> Closed transitions.
-func (p *pub) recordSuccess(node string) {
- p.cbMu.Lock()
- defer p.cbMu.Unlock()
-
- cb, exists := p.cbStates[node]
- if !exists {
- // Initialize circuit breaker state
- p.cbStates[node] = &circuitState{
- state: StateClosed,
- consecutiveFailures: 0,
- }
- return
- }
-
- // Reset to closed state
- cb.state = StateClosed
- cb.consecutiveFailures = 0
- cb.lastFailureTime = time.Time{}
- cb.openTime = time.Time{}
- cb.halfOpenProbeInFlight = false // Clear probe token
-}
-
-// recordFailure updates the circuit breaker state on failed operation.
-// Only records failures for transient/internal errors that should count toward opening the circuit.
-func (p *pub) recordFailure(node string, err error) {
- // Only record failure if the error is transient or internal
- if !isTransientError(err) && !isInternalError(err) {
- return
- }
- p.cbMu.Lock()
- defer p.cbMu.Unlock()
-
- cb, exists := p.cbStates[node]
- if !exists {
- // Initialize circuit breaker state
- cb = &circuitState{
- state: StateClosed,
- consecutiveFailures: 1,
- lastFailureTime: time.Now(),
- }
- p.cbStates[node] = cb
- } else {
- cb.consecutiveFailures++
- cb.lastFailureTime = time.Now()
- }
-
- // Check if we should open the circuit
- threshold := defaultCBThreshold
- if cb.consecutiveFailures >= threshold && cb.state == StateClosed {
- cb.state = StateOpen
- cb.openTime = time.Now()
- p.log.Warn().Str("node", node).Int("failures", cb.consecutiveFailures).Msg("circuit breaker opened")
- } else if cb.state == StateHalfOpen {
- // Failed during half-open, go back to open
- cb.state = StateOpen
- cb.openTime = time.Now()
- cb.halfOpenProbeInFlight = false // Clear probe token
- p.log.Warn().Str("node", node).Msg("circuit breaker reopened after half-open failure")
- }
-}
diff --git a/banyand/queue/pub/pub_test.go b/banyand/queue/pub/pub_test.go
index 5e614831..d6c8237d 100644
--- a/banyand/queue/pub/pub_test.go
+++ b/banyand/queue/pub/pub_test.go
@@ -273,7 +273,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
+ ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
gomega.Expect(ff).Should(gomega.HaveLen(2))
for i := range ff {
diff --git a/banyand/queue/pub/retry.go b/banyand/queue/pub/retry.go
new file mode 100644
index 00000000..1adfa0e2
--- /dev/null
+++ b/banyand/queue/pub/retry.go
@@ -0,0 +1,161 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pub
+
+import (
+ "crypto/rand"
+ "errors"
+ "math/big"
+ "time"
+
+ "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"
+)
+
+const (
+ defaultJitterFactor = 0.2
+ defaultMaxRetries = 3
+ defaultPerRequestTimeout = 2 * time.Second
+ defaultBackoffBase = 500 * time.Millisecond
+ defaultBackoffMax = 30 * time.Second
+)
+
+var (
+ // Retry policy for health check.
+ initBackoff = time.Second
+ maxBackoff = 20 * time.Second
+
+ // Retryable gRPC status codes for streaming send retries.
+ retryableCodes = map[codes.Code]bool{
+ codes.OK: false,
+ codes.Canceled: false,
+ codes.Unknown: false,
+ codes.InvalidArgument: false,
+ codes.DeadlineExceeded: true, // Retryable - operation exceeded deadline
+ codes.NotFound: false,
+ codes.AlreadyExists: false,
+ codes.PermissionDenied: false,
+ codes.ResourceExhausted: true, // Retryable - server resource limits exceeded
+ codes.FailedPrecondition: false,
+ codes.Aborted: false,
+ codes.OutOfRange: false,
+ codes.Unimplemented: false,
+ codes.Internal: true, // Retryable - internal server error should participate in circuit breaker
+ codes.Unavailable: true, // Retryable - service temporarily unavailable
+ codes.DataLoss: false,
+ codes.Unauthenticated: false,
+ }
+)
+
+// secureRandFloat64 generates a cryptographically secure random float64 in [0, 1).
+func secureRandFloat64() float64 {
+ // Generate a random uint64
+ maxVal := big.NewInt(1 << 53) // Use 53 bits for precision similar to math/rand
+ n, err := rand.Int(rand.Reader, maxVal)
+ if err != nil {
+ // Fallback to a reasonable value if crypto/rand fails
+ return 0.5
+ }
+ return float64(n.Uint64()) / float64(1<<53)
+}
+
+// jitteredBackoff calculates backoff duration with jitter to avoid thundering herds.
+// Uses bounded symmetric jitter: backoff * (1 + jitter * (rand() - 0.5) * 2).
+func jitteredBackoff(baseBackoff, maxBackoff time.Duration, attempt int, jitterFactor float64) time.Duration {
+ if jitterFactor < 0 {
+ jitterFactor = 0
+ }
+ if jitterFactor > 1 {
+ jitterFactor = 1
+ }
+
+ // Exponential backoff: base * 2^attempt
+ backoff := baseBackoff
+ for i := 0; i < attempt; i++ {
+ backoff *= 2
+ if backoff > maxBackoff {
+ backoff = maxBackoff
+ break
+ }
+ }
+
+ // Apply jitter: backoff * (1 + jitter * (rand() - 0.5) * 2)
+ // This gives us a range of [backoff * (1-jitter), backoff * (1+jitter)]
+ jitterRange := float64(backoff) * jitterFactor
+ randomFloat := secureRandFloat64()
+ randomOffset := (randomFloat - 0.5) * 2 * jitterRange
+
+ jitteredDuration := time.Duration(float64(backoff) + randomOffset)
+ if jitteredDuration < 0 {
+ jitteredDuration = baseBackoff / 10 // Minimum backoff
+ }
+ if jitteredDuration > maxBackoff {
+ jitteredDuration = maxBackoff
+ }
+
+ return jitteredDuration
+}
+
+// isTransientError checks if the error is considered transient and retryable.
+func isTransientError(err error) bool {
+ if err == nil {
+ 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
+}
+
+// isInternalError checks if the error is an internal server error.
+func isInternalError(err error) bool {
+ if err == nil {
+ 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
+}
(skywalking-banyandb) 09/11: Refactor retry policy handling in pub client
Posted by 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 110fd09440a0fc094863d67884e6847afe0e74bc
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 19:45:50 2025 +0800
Refactor retry policy handling in pub client
- Updated the pub client to use an instance-specific retry policy instead of a global one, enhancing flexibility in error handling.
- Introduced a new function to create a pub instance with a custom retry policy that does not retry on Unavailable errors, improving test coverage and behavior during failure scenarios.
- Adjusted tests to utilize the new pub instance creation method, ensuring consistent behavior in retry logic during message publishing.
---
banyand/queue/pub/client.go | 4 ++--
banyand/queue/pub/pub.go | 10 ++++++----
banyand/queue/pub/pub_suite_test.go | 28 ++++++++++++++++++++++++++++
banyand/queue/pub/pub_test.go | 20 ++++++++++----------
4 files changed, 46 insertions(+), 16 deletions(-)
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 482df7a0..74543ec4 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -225,7 +225,7 @@ func (p *pub) OnAddOrUpdate(md schema.Metadata) {
p.log.Error().Err(err).Msg("failed to load client TLS credentials")
return
}
- conn, err := grpc.NewClient(address, append(credOpts, grpc.WithDefaultServiceConfig(retryPolicy))...)
+ conn, err := grpc.NewClient(address, append(credOpts, grpc.WithDefaultServiceConfig(p.retryPolicy))...)
if err != nil {
p.log.Error().Err(err).Msg("failed to connect to grpc server")
return
@@ -469,7 +469,7 @@ func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Met
p.log.Error().Err(errEvict).Msg("failed to load client TLS credentials (evict)")
return
}
- connEvict, errEvict := grpc.NewClient(node.GrpcAddress, append(credOpts, grpc.WithDefaultServiceConfig(retryPolicy))...)
+ connEvict, errEvict := grpc.NewClient(node.GrpcAddress, append(credOpts, grpc.WithDefaultServiceConfig(p.retryPolicy))...)
if errEvict == nil && p.healthCheck(en.n.String(), connEvict) {
func() {
p.mu.Lock()
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index b7c17df0..7a40046e 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -64,16 +64,17 @@ var (
type pub struct {
schema.UnimplementedOnInitHandler
metadata metadata.Repo
- writableProbe map[string]map[string]struct{}
- cbStates map[string]*circuitState
+ handlers map[bus.Topic]schema.EventHandler
+ log *logger.Logger
registered map[string]*databasev1.Node
active map[string]*client
evictable map[string]evictNode
closer *run.Closer
- handlers map[bus.Topic]schema.EventHandler
- log *logger.Logger
+ writableProbe map[string]map[string]struct{}
+ cbStates map[string]*circuitState
caCertPath string
prefix string
+ retryPolicy string
allowedRoles []databasev1.Role
mu sync.RWMutex
cbMu sync.RWMutex
@@ -305,6 +306,7 @@ func New(metadata metadata.Repo, roles ...databasev1.Role) queue.Client {
prefix: strBuilder.String(),
writableProbe: make(map[string]map[string]struct{}),
cbStates: make(map[string]*circuitState),
+ retryPolicy: retryPolicy,
}
return p
}
diff --git a/banyand/queue/pub/pub_suite_test.go b/banyand/queue/pub/pub_suite_test.go
index bb5c0f7d..fab13eb4 100644
--- a/banyand/queue/pub/pub_suite_test.go
+++ b/banyand/queue/pub/pub_suite_test.go
@@ -222,6 +222,34 @@ func newPub(roles ...databasev1.Role) *pub {
return p.(*pub)
}
+// newPubWithNoRetry creates a pub with a retry policy that doesn't retry Unavailable errors.
+func newPubWithNoRetry(roles ...databasev1.Role) *pub {
+ p := New(nil, roles...)
+ p.(*pub).log = logger.GetLogger("queue-client")
+ p.Register(data.TopicStreamWrite, &mockHandler{})
+ p.Register(data.TopicMeasureWrite, &mockHandler{})
+
+ // Override the retry policy to not retry Unavailable errors
+ noRetryPolicy := `{
+ "methodConfig": [
+ {
+ "name": [{"service": "banyandb.cluster.v1.Service"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 2,
+ "InitialBackoff": ".5s",
+ "MaxBackoff": "10s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ }
+ ]}`
+
+ // Store the original retry policy and replace it
+ p.(*pub).retryPolicy = noRetryPolicy
+ return p.(*pub)
+}
+
func getDataNode(name string, address string) schema.Metadata {
return schema.Metadata{
TypeMeta: schema.TypeMeta{
diff --git a/banyand/queue/pub/pub_test.go b/banyand/queue/pub/pub_test.go
index a8f4251b..5e614831 100644
--- a/banyand/queue/pub/pub_test.go
+++ b/banyand/queue/pub/pub_test.go
@@ -64,7 +64,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- bp := p.NewBatchPublisher(3 * time.Second)
+ bp := p.NewBatchPublisher(15 * time.Second)
ctx := context.TODO()
for i := 0; i < 10; i++ {
_, err := bp.Publish(ctx, data.TopicStreamWrite,
@@ -94,7 +94,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- bp := p.NewBatchPublisher(3 * time.Second)
+ bp := p.NewBatchPublisher(15 * time.Second)
ctx := context.TODO()
for i := 0; i < 10; i++ {
_, err := bp.Publish(ctx, data.TopicStreamWrite,
@@ -141,7 +141,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- bp := p.NewBatchPublisher(3 * time.Second)
+ bp := p.NewBatchPublisher(15 * time.Second)
ctx := context.TODO()
for i := 0; i < 10; i++ {
_, err := bp.Publish(ctx, data.TopicStreamWrite,
@@ -178,7 +178,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- bp := p.NewBatchPublisher(3 * time.Second)
+ bp := p.NewBatchPublisher(15 * time.Second)
ctx := context.TODO()
for i := 0; i < 10; i++ {
_, err := bp.Publish(ctx, data.TopicStreamWrite,
@@ -217,7 +217,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
p.OnAddOrUpdate(node2)
req := &streamv1.QueryRequest{}
- ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), req))
+ ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), req))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
gomega.Expect(ff).Should(gomega.HaveLen(2))
messages, err := ff[0].GetAll()
@@ -230,7 +230,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
addr2 := getAddress()
closeFn1 := setup(addr1, codes.OK, 200*time.Millisecond)
closeFn2 := setup(addr2, codes.Unavailable, 0)
- p := newPub()
+ p := newPubWithNoRetry()
defer func() {
p.GracefulStop()
closeFn1()
@@ -241,7 +241,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
+ ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
gomega.Expect(ff).Should(gomega.HaveLen(2))
for i := range ff {
@@ -273,7 +273,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
node2 := getDataNode("node2", addr2)
p.OnAddOrUpdate(node2)
- ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
+ ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery, bus.NewMessage(bus.MessageID(1), &streamv1.QueryRequest{}))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
gomega.Expect(ff).Should(gomega.HaveLen(2))
for i := range ff {
@@ -326,7 +326,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
group1: {""},
}
- ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery,
+ ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery,
bus.NewMessageWithNodeSelectors(bus.MessageID(1), nodeSelectors, timeRange, &streamv1.QueryRequest{}))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
@@ -377,7 +377,7 @@ var _ = ginkgo.Describe("Publish and Broadcast", func() {
group1: {"role=ingest"},
}
- ff, err := p.Broadcast(3*time.Second, data.TopicStreamQuery,
+ ff, err := p.Broadcast(15*time.Second, data.TopicStreamQuery,
bus.NewMessageWithNodeSelectors(bus.MessageID(1), nodeSelectors, timeRange, &streamv1.QueryRequest{}))
gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
(skywalking-banyandb) 02/11: Add circuit breaker implementation for client health management
Posted by 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 d22ec2fc4081960f7c2f0f89ab39d6362d8bf43b
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 04:36:49 2025 +0800
Add circuit breaker implementation for client health management
- Introduced a circuit breaker mechanism to manage client health checks, allowing for better handling of failures and recovery.
- Added `CircuitState` type and associated logic to track the state of each client connection.
- Implemented methods to record successes and failures, adjusting the circuit state accordingly.
- Enhanced the `isRequestAllowed` function to determine request eligibility based on the circuit breaker state.
- Updated the `pub` struct to include a map for circuit breaker states, improving resilience in client operations.
---
banyand/queue/pub/client.go | 103 ++++++++++++++++++++++++++++++++++++++++++++
banyand/queue/pub/pub.go | 9 ++--
2 files changed, 109 insertions(+), 3 deletions(-)
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index a0a68a6e..de22b497 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -60,6 +60,23 @@ var (
}]}`, serviceName)
)
+// CircuitState defines the circuit breaker states.
+type CircuitState int
+
+const (
+ StateClosed CircuitState = iota // Normal operation
+ StateOpen // Reject requests until cooldown expires
+ StateHalfOpen // Allow a single probe
+)
+
+// circuitState holds circuit breaker metadata; it does NOT duplicate gRPC clients/conns.
+type circuitState struct {
+ lastFailureTime time.Time
+ openTime time.Time
+ state CircuitState
+ consecutiveFailures int
+}
+
type client struct {
client clusterv1.ServiceClient
conn *grpc.ClientConn
@@ -131,6 +148,8 @@ func (p *pub) OnAddOrUpdate(md schema.Metadata) {
c := clusterv1.NewServiceClient(conn)
p.active[name] = &client{conn: conn, client: c, md: md}
p.addClient(md)
+ // Initialize or reset circuit breaker state to closed
+ p.recordSuccess(name)
p.log.Info().Str("status", p.dump()).Stringer("node", node).Msg("new node is healthy, add it to active queue")
}
@@ -279,6 +298,8 @@ func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Met
p.active[name] = &client{conn: connEvict, client: c, md: md}
p.addClient(md)
delete(p.evictable, name)
+ // Reset circuit breaker state to closed
+ p.recordSuccess(name)
p.log.Info().Str("status", p.dump()).Stringer("node", en.n).Msg("node is healthy, move it back to active queue")
}()
return
@@ -417,6 +438,8 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
if !okCur {
return
}
+ // Record success for circuit breaker
+ p.recordSuccess(nodeName)
h.OnAddOrUpdate(nodeCur.md)
}()
return
@@ -471,3 +494,83 @@ type evictNode struct {
n *databasev1.Node
c chan struct{}
}
+
+// isRequestAllowed checks if a request to the given node is allowed based on circuit breaker state.
+func (p *pub) isRequestAllowed(node string) bool {
+ p.cbMu.RLock()
+ defer p.cbMu.RUnlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ return true // No circuit breaker state, allow request
+ }
+
+ switch cb.state {
+ case StateClosed:
+ return true
+ case StateOpen:
+ // Check if cooldown period has expired
+ return time.Since(cb.openTime) >= 60*time.Second // TODO: make configurable
+ case StateHalfOpen:
+ // Allow only one probe request in half-open state
+ // This is a simple implementation - in production, we'd need atomic operations
+ return true
+ default:
+ return true
+ }
+}
+
+// recordSuccess resets the circuit breaker state to Closed on successful operation.
+func (p *pub) recordSuccess(node string) {
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ // Initialize circuit breaker state
+ p.cbStates[node] = &circuitState{
+ state: StateClosed,
+ consecutiveFailures: 0,
+ }
+ return
+ }
+
+ // Reset to closed state
+ cb.state = StateClosed
+ cb.consecutiveFailures = 0
+ cb.lastFailureTime = time.Time{}
+ cb.openTime = time.Time{}
+}
+
+// recordFailure updates the circuit breaker state on failed operation.
+func (p *pub) recordFailure(node string) {
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
+
+ cb, exists := p.cbStates[node]
+ if !exists {
+ // Initialize circuit breaker state
+ cb = &circuitState{
+ state: StateClosed,
+ consecutiveFailures: 1,
+ lastFailureTime: time.Now(),
+ }
+ p.cbStates[node] = cb
+ } else {
+ cb.consecutiveFailures++
+ cb.lastFailureTime = time.Now()
+ }
+
+ // Check if we should open the circuit
+ threshold := 5 // TODO: make configurable
+ if cb.consecutiveFailures >= threshold && cb.state == StateClosed {
+ cb.state = StateOpen
+ cb.openTime = time.Now()
+ p.log.Warn().Str("node", node).Int("failures", cb.consecutiveFailures).Msg("circuit breaker opened")
+ } else if cb.state == StateHalfOpen {
+ // Failed during half-open, go back to open
+ cb.state = StateOpen
+ cb.openTime = time.Now()
+ p.log.Warn().Str("node", node).Msg("circuit breaker reopened after half-open failure")
+ }
+}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 987931db..d8eaa07b 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -64,17 +64,19 @@ var (
type pub struct {
schema.UnimplementedOnInitHandler
metadata metadata.Repo
- handlers map[bus.Topic]schema.EventHandler
- log *logger.Logger
+ writableProbe map[string]map[string]struct{}
+ cbStates map[string]*circuitState
registered map[string]*databasev1.Node
active map[string]*client
evictable map[string]evictNode
closer *run.Closer
- writableProbe map[string]map[string]struct{}
+ handlers map[bus.Topic]schema.EventHandler
+ log *logger.Logger
caCertPath string
prefix string
allowedRoles []databasev1.Role
mu sync.RWMutex
+ cbMu sync.RWMutex
writableProbeMu sync.Mutex
tlsEnabled bool
}
@@ -290,6 +292,7 @@ func New(metadata metadata.Repo, roles ...databasev1.Role) queue.Client {
allowedRoles: roles,
prefix: strBuilder.String(),
writableProbe: make(map[string]map[string]struct{}),
+ cbStates: make(map[string]*circuitState),
}
return p
}
(skywalking-banyandb) 03/11: Enhance circuit breaker logic in client health management
Posted by 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 aa82e3fb473aaffcd48bb0ef3a0f348eafe087a9
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 04:44:15 2025 +0800
Enhance circuit breaker logic in client health management
---
banyand/queue/pub/client.go | 18 ++++++++++++------
1 file changed, 12 insertions(+), 6 deletions(-)
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index de22b497..6b8a8ad6 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -496,9 +496,10 @@ type evictNode struct {
}
// isRequestAllowed checks if a request to the given node is allowed based on circuit breaker state.
+// It also handles state transitions from Open to Half-Open when cooldown expires.
func (p *pub) isRequestAllowed(node string) bool {
- p.cbMu.RLock()
- defer p.cbMu.RUnlock()
+ p.cbMu.Lock()
+ defer p.cbMu.Unlock()
cb, exists := p.cbStates[node]
if !exists {
@@ -510,17 +511,22 @@ func (p *pub) isRequestAllowed(node string) bool {
return true
case StateOpen:
// Check if cooldown period has expired
- return time.Since(cb.openTime) >= 60*time.Second // TODO: make configurable
+ if time.Since(cb.openTime) >= 60*time.Second { // TODO: make configurable
+ // Transition to Half-Open to allow probe requests
+ cb.state = StateHalfOpen
+ p.log.Info().Str("node", node).Msg("circuit breaker transitioned to half-open")
+ return true
+ }
+ return false // Still in cooldown period
case StateHalfOpen:
- // Allow only one probe request in half-open state
- // This is a simple implementation - in production, we'd need atomic operations
+ // Allow requests in half-open state (probe requests)
return true
default:
return true
}
}
-
// recordSuccess resets the circuit breaker state to Closed on successful operation.
+// This handles Half-Open -> Closed transitions.
func (p *pub) recordSuccess(node string) {
p.cbMu.Lock()
defer p.cbMu.Unlock()
(skywalking-banyandb) 11/11: Update CHANGES.md
Posted by 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 1a4f44823c99ead37755ac52ac52996be65719c6
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 20:15:39 2025 +0800
Update CHANGES.md
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
CHANGES.md | 3 ++-
banyand/queue/pub/client.go | 6 ++----
2 files changed, 4 insertions(+), 5 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 6650f795..e45678c6 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -37,7 +37,8 @@ Release Notes.
- Data ingestion and retrieval.
- Flush memory data to disk.
- Merge memory data and disk data.
-- Enhance access log functionality with sampling option
+- Enhance access log functionality with sampling option.
+- Implement a resilient publisher with circuit breaker and retry logic with exponential backoff.
### Bug Fixes
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index bcca7868..ddd5e02d 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -39,9 +39,8 @@ const (
rpcTimeout = 2 * time.Second
)
-var (
- // The timeout is set by each RPC.
- retryPolicy = `{
+// The timeout is set by each RPC.
+var retryPolicy = `{
"methodConfig": [
{
"name": [{"service": "banyandb.cluster.v1.Service"}],
@@ -110,7 +109,6 @@ var (
}
}
]}`
-)
type client struct {
client clusterv1.ServiceClient
(skywalking-banyandb) 01/11: Refactor batch publisher and client health check logic
Posted by 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 3d162cbbe4c578d1442b472592cf43a2d27bc385
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 03:44:32 2025 +0800
Refactor batch publisher and client health check logic
- Updated the batch publisher to pass the current node name to goroutines for better error handling during message publishing.
- Modified the client health check to deduplicate concurrent probes for the same node and topic, improving efficiency and reducing unnecessary checks.
---
banyand/queue/pub/batch.go | 9 +++++----
banyand/queue/pub/client.go | 39 ++++++++++++++++++++++++++++--------
banyand/queue/pub/client_test.go | 2 +-
banyand/queue/pub/pub.go | 43 +++++++++++++++++++++-------------------
4 files changed, 60 insertions(+), 33 deletions(-)
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index 00a442ef..4d6fdc8c 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -147,7 +147,8 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
}
bp.f.events = append(bp.f.events, make(chan batchEvent))
_ = sendData()
- go func(s clusterv1.Service_SendClient, deferFn func(), bc chan batchEvent) {
+ nodeName := node
+ go func(s clusterv1.Service_SendClient, deferFn func(), bc chan batchEvent, curNode string) {
defer func() {
close(bc)
deferFn()
@@ -160,7 +161,7 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
resp, errRecv := s.Recv()
if errRecv != nil {
if isFailoverError(errRecv) {
- bc <- batchEvent{n: node, e: common.NewErrorWithStatus(modelv1.Status_STATUS_INTERNAL_ERROR, errRecv.Error())}
+ bc <- batchEvent{n: curNode, e: common.NewErrorWithStatus(modelv1.Status_STATUS_INTERNAL_ERROR, errRecv.Error())}
}
return
}
@@ -172,9 +173,9 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
}
if isFailoverStatus(resp.Status) {
ce := common.NewErrorWithStatus(resp.Status, resp.Error)
- bc <- batchEvent{n: node, e: ce}
+ bc <- batchEvent{n: curNode, e: ce}
}
- }(stream, deferFn, bp.f.events[len(bp.f.events)-1])
+ }(stream, deferFn, bp.f.events[len(bp.f.events)-1], nodeName)
}
return nil, err
}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 39daf2af..a0a68a6e 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -372,7 +372,8 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
if !ok {
return false, nil
}
- err := p.checkServiceHealth(topic.String(), node.conn)
+ topicStr := topic.String()
+ err := p.checkServiceHealth(topicStr, node.conn)
if err == nil {
return true, nil
}
@@ -380,25 +381,47 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
if !p.closer.AddRunning() {
return false, err
}
- go func() {
+ // Deduplicate concurrent probes for the same node/topic.
+ p.writableProbeMu.Lock()
+ if _, ok := p.writableProbe[n]; !ok {
+ p.writableProbe[n] = make(map[string]struct{})
+ }
+ if _, exists := p.writableProbe[n][topicStr]; exists {
+ p.writableProbeMu.Unlock()
+ return false, err
+ }
+ p.writableProbe[n][topicStr] = struct{}{}
+ p.writableProbeMu.Unlock()
+
+ go func(nodeName, t string) {
defer p.closer.Done()
+ defer func() {
+ p.writableProbeMu.Lock()
+ if topics, ok := p.writableProbe[nodeName]; ok {
+ delete(topics, t)
+ if len(topics) == 0 {
+ delete(p.writableProbe, nodeName)
+ }
+ }
+ p.writableProbeMu.Unlock()
+ }()
backoff := initBackoff
for {
select {
case <-time.After(backoff):
- if errInternal := p.checkServiceHealth(topic.String(), node.conn); errInternal != nil {
+ if errInternal := p.checkServiceHealth(t, node.conn); errInternal == nil {
func() {
p.mu.Lock()
defer p.mu.Unlock()
- node, ok := p.active[n]
- if !ok {
+ nodeCur, okCur := p.active[nodeName]
+ if !okCur {
return
}
- h.OnAddOrUpdate(node.md)
+ h.OnAddOrUpdate(nodeCur.md)
}()
return
}
- p.log.Warn().Str("topic", topic.String()).Err(err).Str("node", n).Dur("backoff", backoff).Msg("data node can not ingest data")
+ p.log.Warn().Str("topic", t).Err(err).Str("node", nodeName).Dur("backoff", backoff).Msg("data node can not ingest data")
case <-p.closer.CloseNotify():
return
}
@@ -408,7 +431,7 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
backoff = maxBackoff
}
}
- }()
+ }(n, topicStr)
return false, err
}
diff --git a/banyand/queue/pub/client_test.go b/banyand/queue/pub/client_test.go
index 59ad104d..0b797cb1 100644
--- a/banyand/queue/pub/client_test.go
+++ b/banyand/queue/pub/client_test.go
@@ -144,7 +144,7 @@ var _ = ginkgo.Describe("publish clients register/unregister", func() {
verifyClientsWithGomega(gomega.Default, p, data.TopicMeasureWrite, 1, 0, 1, 0)
hs.SetServingStatus(data.TopicStreamWrite.String(), grpc_health_v1.HealthCheckResponse_SERVING)
gomega.Eventually(func(g gomega.Gomega) {
- verifyClientsWithGomega(g, p, data.TopicStreamWrite, 1, 0, 2, 1)
+ verifyClientsWithGomega(g, p, data.TopicStreamWrite, 1, 0, 1, 1)
}, flags.EventuallyTimeout).Should(gomega.Succeed())
})
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index f5feaf6a..987931db 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -63,18 +63,20 @@ var (
type pub struct {
schema.UnimplementedOnInitHandler
- metadata metadata.Repo
- evictable map[string]evictNode
- log *logger.Logger
- registered map[string]*databasev1.Node
- active map[string]*client
- handlers map[bus.Topic]schema.EventHandler
- closer *run.Closer
- caCertPath string
- prefix string
- allowedRoles []databasev1.Role
- mu sync.RWMutex
- tlsEnabled bool
+ metadata metadata.Repo
+ handlers map[bus.Topic]schema.EventHandler
+ log *logger.Logger
+ registered map[string]*databasev1.Node
+ active map[string]*client
+ evictable map[string]evictNode
+ closer *run.Closer
+ writableProbe map[string]map[string]struct{}
+ caCertPath string
+ prefix string
+ allowedRoles []databasev1.Role
+ mu sync.RWMutex
+ writableProbeMu sync.Mutex
+ tlsEnabled bool
}
func (p *pub) FlagSet() *run.FlagSet {
@@ -279,14 +281,15 @@ func New(metadata metadata.Repo, roles ...databasev1.Role) queue.Client {
}
}
p := &pub{
- metadata: metadata,
- active: make(map[string]*client),
- evictable: make(map[string]evictNode),
- registered: make(map[string]*databasev1.Node),
- handlers: make(map[bus.Topic]schema.EventHandler),
- closer: run.NewCloser(1),
- allowedRoles: roles,
- prefix: strBuilder.String(),
+ metadata: metadata,
+ active: make(map[string]*client),
+ evictable: make(map[string]evictNode),
+ registered: make(map[string]*databasev1.Node),
+ handlers: make(map[bus.Topic]schema.EventHandler),
+ closer: run.NewCloser(1),
+ allowedRoles: roles,
+ prefix: strBuilder.String(),
+ writableProbe: make(map[string]map[string]struct{}),
}
return p
}
(skywalking-banyandb) 04/11: Implement retry logic and circuit breaker enhancements for message publishing
Posted by 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 d0483e2a538214f28c47973c0e49fab312a32f43
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 16:48:46 2025 +0800
Implement retry logic and circuit breaker enhancements for message publishing
- Added a retry mechanism with exponential backoff and jitter for sending messages in the batch publisher.
- Enhanced circuit breaker logic to manage request eligibility based on success and failure rates.
- Introduced tests for the retry functionality and circuit breaker state transitions, ensuring robust error handling and recovery.
- Updated existing methods to integrate the new retry and circuit breaker features, improving resilience in message publishing operations.
---
banyand/queue/pub/batch.go | 81 ++++-
banyand/queue/pub/circuitbreaker_test.go | 356 +++++++++++++++++++++
banyand/queue/pub/client.go | 217 ++++++++++---
banyand/queue/pub/retry_test.go | 528 +++++++++++++++++++++++++++++++
4 files changed, 1143 insertions(+), 39 deletions(-)
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index 4d6fdc8c..f0b9198e 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -70,6 +70,13 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
continue
}
node := m.Node()
+
+ // Check circuit breaker before attempting send
+ if !bp.pub.isRequestAllowed(node) {
+ err = multierr.Append(err, fmt.Errorf("circuit breaker open for node %s", node))
+ continue
+ }
+
sendData := func() (success bool) {
if stream, ok := bp.streams[node]; ok {
defer func() {
@@ -84,12 +91,16 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
return false
default:
}
- errSend := stream.client.Send(r)
+ errSend := bp.retrySend(ctx, stream.client, r, node)
if errSend != nil {
err = multierr.Append(err, fmt.Errorf("failed to send message to node %s: %w", node, errSend))
+ // Record failure for circuit breaker
+ bp.pub.recordFailure(node)
return false
}
- return errSend == nil
+ // Record success for circuit breaker
+ bp.pub.recordSuccess(node)
+ return true
}
return false
}
@@ -255,3 +266,69 @@ func (b *batchFuture) get() map[string]batchEvent {
func isFailoverStatus(s modelv1.Status) bool {
return s == modelv1.Status_STATUS_DISK_FULL
}
+
+// retrySend implements bounded retries for client streaming sends with exponential backoff and jitter.
+func (bp *batchPublisher) retrySend(ctx context.Context, stream clusterv1.Service_SendClient, r *clusterv1.SendRequest, node string) error {
+ var lastErr error
+
+ for attempt := 0; attempt <= defaultMaxRetries; attempt++ {
+ // Create per-attempt timeout context
+ attemptCtx, cancel := context.WithTimeout(ctx, defaultPerRequestTimeout)
+
+ // Check if parent context is canceled or stream context is done
+ select {
+ case <-ctx.Done():
+ cancel()
+ return ctx.Err()
+ case <-stream.Context().Done():
+ cancel()
+ return stream.Context().Err()
+ case <-attemptCtx.Done():
+ cancel()
+ if attempt == 0 {
+ // First attempt timed out before we could even try
+ lastErr = attemptCtx.Err()
+ continue
+ }
+ default:
+ }
+
+ // Attempt to send
+ sendErr := stream.Send(r)
+ cancel()
+
+ if sendErr == nil {
+ // Success
+ return nil
+ }
+
+ lastErr = sendErr
+
+ // Check if error is retryable
+ if !isTransientError(sendErr) {
+ // Non-transient error, don't retry
+ return sendErr
+ }
+
+ // If this was the last attempt, don't sleep
+ if attempt >= defaultMaxRetries {
+ break
+ }
+
+ // Calculate backoff with jitter
+ backoff := jitteredBackoff(defaultBackoffBase, defaultBackoffMax, attempt, defaultJitterFactor)
+
+ // Sleep with backoff, but respect context cancellation
+ select {
+ case <-time.After(backoff):
+ // Continue to next attempt
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-stream.Context().Done():
+ return stream.Context().Err()
+ }
+ }
+
+ // All retries exhausted
+ return fmt.Errorf("retry exhausted for node %s after %d attempts, last error: %w", node, defaultMaxRetries+1, lastErr)
+}
diff --git a/banyand/queue/pub/circuitbreaker_test.go b/banyand/queue/pub/circuitbreaker_test.go
new file mode 100644
index 00000000..9f1976b1
--- /dev/null
+++ b/banyand/queue/pub/circuitbreaker_test.go
@@ -0,0 +1,356 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pub
+
+import (
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+func TestCircuitBreakerStateTransitions(t *testing.T) {
+ tests := []struct {
+ name string
+ setup func(*pub)
+ actions []func(*pub, string)
+ expectedState CircuitState
+ allowsRequests bool
+ }{
+ {
+ name: "closed_to_open_after_failures",
+ setup: func(p *pub) {
+ p.recordSuccess("test-node")
+ },
+ actions: []func(*pub, string){
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ },
+ expectedState: StateOpen,
+ allowsRequests: false,
+ },
+ {
+ name: "closed_remains_closed_below_threshold",
+ setup: func(p *pub) {
+ p.recordSuccess("test-node")
+ },
+ actions: []func(*pub, string){
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node) },
+ },
+ expectedState: StateClosed,
+ allowsRequests: true,
+ },
+ {
+ name: "open_to_half_open_after_cooldown",
+ setup: func(p *pub) {
+ p.recordSuccess("test-node")
+ // Trip the circuit breaker
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure("test-node")
+ }
+ // Simulate cooldown period has passed
+ p.cbMu.Lock()
+ cb := p.cbStates["test-node"]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+ },
+ actions: []func(*pub, string){},
+ expectedState: StateOpen, // Will transition to half-open in isRequestAllowed
+ allowsRequests: true, // Should allow requests after cooldown
+ },
+ {
+ name: "half_open_to_closed_on_success",
+ setup: func(p *pub) {
+ p.recordSuccess("test-node")
+ // Trip the circuit breaker
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure("test-node")
+ }
+ // Set to half-open state
+ p.cbMu.Lock()
+ cb := p.cbStates["test-node"]
+ cb.state = StateHalfOpen
+ p.cbMu.Unlock()
+ },
+ actions: []func(*pub, string){
+ func(p *pub, node string) { p.recordSuccess(node) },
+ },
+ expectedState: StateClosed,
+ allowsRequests: true,
+ },
+ {
+ name: "half_open_to_open_on_failure",
+ setup: func(p *pub) {
+ p.recordSuccess("test-node")
+ // Trip the circuit breaker
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure("test-node")
+ }
+ // Set to half-open state
+ p.cbMu.Lock()
+ cb := p.cbStates["test-node"]
+ cb.state = StateHalfOpen
+ p.cbMu.Unlock()
+ },
+ actions: []func(*pub, string){
+ func(p *pub, node string) { p.recordFailure(node) },
+ },
+ expectedState: StateOpen,
+ allowsRequests: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ // Setup
+ if tt.setup != nil {
+ tt.setup(p)
+ }
+
+ // Execute actions
+ for _, action := range tt.actions {
+ action(p, "test-node")
+ }
+
+ // Check final state
+ p.cbMu.RLock()
+ cb, exists := p.cbStates["test-node"]
+ p.cbMu.RUnlock()
+
+ require.True(t, exists, "circuit breaker state should exist")
+ assert.Equal(t, tt.expectedState, cb.state, "circuit breaker state mismatch")
+
+ // Check if requests are allowed
+ allowed := p.isRequestAllowed("test-node")
+ assert.Equal(t, tt.allowsRequests, allowed, "request allowance mismatch")
+ })
+ }
+}
+
+func TestCircuitBreakerConcurrency(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ const numGoroutines = 100
+ const numOperations = 50
+ node := "test-node"
+
+ var wg sync.WaitGroup
+ wg.Add(numGoroutines * 2) // Half for success, half for failure
+
+ // Simulate concurrent successes
+ for i := 0; i < numGoroutines; i++ {
+ go func() {
+ defer wg.Done()
+ for j := 0; j < numOperations; j++ {
+ p.recordSuccess(node)
+ p.isRequestAllowed(node)
+ }
+ }()
+ }
+
+ // Simulate concurrent failures
+ for i := 0; i < numGoroutines; i++ {
+ go func() {
+ defer wg.Done()
+ for j := 0; j < numOperations; j++ {
+ p.recordFailure(node)
+ p.isRequestAllowed(node)
+ }
+ }()
+ }
+
+ wg.Wait()
+
+ // Verify circuit breaker state exists and is in a valid state
+ p.cbMu.RLock()
+ cb, exists := p.cbStates[node]
+ p.cbMu.RUnlock()
+
+ require.True(t, exists, "circuit breaker state should exist")
+ assert.Contains(t, []CircuitState{StateClosed, StateOpen, StateHalfOpen}, cb.state, "circuit breaker should be in a valid state")
+}
+
+func TestCircuitBreakerMultipleNodes(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ nodes := []string{"node1", "node2", "node3"}
+
+ // Initialize all nodes
+ for _, node := range nodes {
+ p.recordSuccess(node)
+ }
+
+ // Trip circuit breaker for node1 only
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure("node1")
+ }
+
+ // Add some failures to node2 but below threshold
+ for i := 0; i < defaultCBThreshold-1; i++ {
+ p.recordFailure("node2")
+ }
+
+ // Keep node3 healthy
+ p.recordSuccess("node3")
+
+ // Verify states
+ assert.False(t, p.isRequestAllowed("node1"), "node1 should have circuit breaker open")
+ assert.True(t, p.isRequestAllowed("node2"), "node2 should still allow requests")
+ assert.True(t, p.isRequestAllowed("node3"), "node3 should allow requests")
+
+ // Check circuit breaker states
+ p.cbMu.RLock()
+ defer p.cbMu.RUnlock()
+
+ cb1, exists1 := p.cbStates["node1"]
+ require.True(t, exists1)
+ assert.Equal(t, StateOpen, cb1.state)
+
+ cb2, exists2 := p.cbStates["node2"]
+ require.True(t, exists2)
+ assert.Equal(t, StateClosed, cb2.state)
+
+ cb3, exists3 := p.cbStates["node3"]
+ require.True(t, exists3)
+ assert.Equal(t, StateClosed, cb3.state)
+}
+
+func TestCircuitBreakerRecoveryAfterCooldown(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := "test-node"
+
+ // Initialize node
+ p.recordSuccess(node)
+
+ // Trip circuit breaker
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node)
+ }
+
+ // Verify circuit is open
+ assert.False(t, p.isRequestAllowed(node), "circuit should be open")
+
+ // Simulate cooldown period passage
+ p.cbMu.Lock()
+ cb := p.cbStates[node]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+
+ // Check that circuit allows requests (transitions to half-open)
+ allowed := p.isRequestAllowed(node)
+ assert.True(t, allowed, "circuit should allow requests after cooldown")
+
+ // Verify state transitioned to half-open
+ p.cbMu.RLock()
+ assert.Equal(t, StateHalfOpen, cb.state, "circuit should be in half-open state")
+ p.cbMu.RUnlock()
+
+ // Successful request should close the circuit
+ p.recordSuccess(node)
+
+ p.cbMu.RLock()
+ assert.Equal(t, StateClosed, cb.state, "circuit should be closed after success")
+ assert.Equal(t, 0, cb.consecutiveFailures, "failure count should be reset")
+ p.cbMu.RUnlock()
+}
+
+func TestCircuitBreakerInitialization(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := "new-node"
+
+ // First request to non-existent circuit breaker should be allowed
+ allowed := p.isRequestAllowed(node)
+ assert.True(t, allowed, "requests should be allowed for non-existent circuit breaker")
+
+ // Record success should initialize the circuit breaker
+ p.recordSuccess(node)
+
+ p.cbMu.RLock()
+ cb, exists := p.cbStates[node]
+ p.cbMu.RUnlock()
+
+ require.True(t, exists, "circuit breaker should be initialized")
+ assert.Equal(t, StateClosed, cb.state, "new circuit breaker should be closed")
+ assert.Equal(t, 0, cb.consecutiveFailures, "new circuit breaker should have zero failures")
+}
+
+func TestCircuitBreakerFailureThresholdEdgeCase(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := "test-node"
+
+ // Initialize node
+ p.recordSuccess(node)
+
+ // Add failures just below threshold
+ for i := 0; i < defaultCBThreshold-1; i++ {
+ p.recordFailure(node)
+ }
+
+ // Circuit should still be closed
+ p.cbMu.RLock()
+ cb := p.cbStates[node]
+ assert.Equal(t, StateClosed, cb.state, "circuit should still be closed")
+ assert.Equal(t, defaultCBThreshold-1, cb.consecutiveFailures, "failure count should be at threshold-1")
+ p.cbMu.RUnlock()
+
+ // One more failure should open the circuit
+ p.recordFailure(node)
+
+ p.cbMu.RLock()
+ assert.Equal(t, StateOpen, cb.state, "circuit should be open after reaching threshold")
+ assert.Equal(t, defaultCBThreshold, cb.consecutiveFailures, "failure count should be at threshold")
+ p.cbMu.RUnlock()
+}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 6b8a8ad6..73f44491 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -19,11 +19,15 @@ package pub
import (
"context"
+ "crypto/rand"
"fmt"
+ "math/big"
"time"
"google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
"google.golang.org/grpc/health/grpc_health_v1"
+ "google.golang.org/grpc/status"
"github.com/apache/skywalking-banyandb/api/common"
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
@@ -35,34 +39,119 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-const rpcTimeout = 2 * time.Second
+const (
+ rpcTimeout = 2 * time.Second
+ defaultJitterFactor = 0.2
+ defaultMaxRetries = 3
+ defaultCBThreshold = 5
+ defaultCBResetTimeout = 60 * time.Second
+ defaultPerRequestTimeout = 2 * time.Second
+ defaultBackoffBase = 500 * time.Millisecond
+ defaultBackoffMax = 30 * time.Second
+)
var (
// Retry policy for health check.
- initBackoff = time.Second
- maxBackoff = 20 * time.Second
- backoffMultiplier = 2.0
-
- serviceName = clusterv1.Service_ServiceDesc.ServiceName
+ initBackoff = time.Second
+ maxBackoff = 20 * time.Second
// The timeout is set by each RPC.
- retryPolicy = fmt.Sprintf(`{
- "methodConfig": [{
- "name": [{"service": "%s"}],
- "waitForReady": true,
- "retryPolicy": {
- "MaxAttempts": 4,
- "InitialBackoff": ".5s",
- "MaxBackoff": "10s",
- "BackoffMultiplier": 1.0,
- "RetryableStatusCodes": [ "UNAVAILABLE" ]
+ retryPolicy = `{
+ "methodConfig": [
+ {
+ "name": [{"service": "banyandb.cluster.v1.Service"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".5s",
+ "MaxBackoff": "10s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ },
+ {
+ "name": [{"service": "banyandb.cluster.v1.ChunkedSyncService"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 3,
+ "InitialBackoff": ".5s",
+ "MaxBackoff": "5s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ },
+ {
+ "name": [{"service": "banyandb.trace.v1.TraceService"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".2s",
+ "MaxBackoff": "5s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ },
+ {
+ "name": [{"service": "banyandb.stream.v1.StreamService"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".2s",
+ "MaxBackoff": "5s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ },
+ {
+ "name": [{"service": "banyandb.measure.v1.MeasureService"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".2s",
+ "MaxBackoff": "5s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
+ },
+ {
+ "name": [{"service": "banyandb.property.v1.PropertyService"}],
+ "waitForReady": true,
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".2s",
+ "MaxBackoff": "5s",
+ "BackoffMultiplier": 2.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED" ]
+ }
}
- }]}`, serviceName)
+ ]}`
+
+ // Retryable gRPC status codes for streaming send retries.
+ retryableCodes = map[codes.Code]bool{
+ codes.OK: false,
+ codes.Canceled: false,
+ codes.Unknown: false,
+ codes.InvalidArgument: false,
+ codes.DeadlineExceeded: true, // Retryable - operation exceeded deadline
+ codes.NotFound: false,
+ codes.AlreadyExists: false,
+ codes.PermissionDenied: false,
+ codes.ResourceExhausted: true, // Retryable - server resource limits exceeded
+ codes.FailedPrecondition: false,
+ codes.Aborted: false,
+ codes.OutOfRange: false,
+ codes.Unimplemented: false,
+ codes.Internal: false,
+ codes.Unavailable: true, // Retryable - service temporarily unavailable
+ codes.DataLoss: false,
+ codes.Unauthenticated: false,
+ }
)
// CircuitState defines the circuit breaker states.
type CircuitState int
+// CircuitState defines the circuit breaker states.
const (
StateClosed CircuitState = iota // Normal operation
StateOpen // Reject requests until cooldown expires
@@ -213,9 +302,10 @@ func (p *pub) OnDelete(md schema.Metadata) {
}
go func() {
defer p.closer.Done()
- backoff := initBackoff
var elapsed time.Duration
+ attempt := 0
for {
+ backoff := jitteredBackoff(initBackoff, maxBackoff, attempt, defaultJitterFactor)
select {
case <-time.After(backoff):
if func() bool {
@@ -237,11 +327,7 @@ func (p *pub) OnDelete(md schema.Metadata) {
case <-p.closer.CloseNotify():
return
}
- if backoff < maxBackoff {
- backoff *= time.Duration(backoffMultiplier)
- } else {
- backoff = maxBackoff
- }
+ attempt++
}
}()
}
@@ -258,6 +344,68 @@ func (p *pub) removeNodeIfUnhealthy(md schema.Metadata, node *databasev1.Node, c
return true
}
+// secureRandFloat64 generates a cryptographically secure random float64 in [0, 1).
+func secureRandFloat64() float64 {
+ // Generate a random uint64
+ maxVal := big.NewInt(1 << 53) // Use 53 bits for precision similar to math/rand
+ n, err := rand.Int(rand.Reader, maxVal)
+ if err != nil {
+ // Fallback to a reasonable value if crypto/rand fails
+ return 0.5
+ }
+ return float64(n.Uint64()) / float64(1<<53)
+}
+
+// jitteredBackoff calculates backoff duration with jitter to avoid thundering herds.
+// Uses bounded symmetric jitter: backoff * (1 + jitter * (rand() - 0.5) * 2).
+func jitteredBackoff(baseBackoff, maxBackoff time.Duration, attempt int, jitterFactor float64) time.Duration {
+ if jitterFactor < 0 {
+ jitterFactor = 0
+ }
+ if jitterFactor > 1 {
+ jitterFactor = 1
+ }
+
+ // Exponential backoff: base * 2^attempt
+ backoff := baseBackoff
+ for i := 0; i < attempt; i++ {
+ backoff *= 2
+ if backoff > maxBackoff {
+ backoff = maxBackoff
+ break
+ }
+ }
+
+ // Apply jitter: backoff * (1 + jitter * (rand() - 0.5) * 2)
+ // This gives us a range of [backoff * (1-jitter), backoff * (1+jitter)]
+ jitterRange := float64(backoff) * jitterFactor
+ randomFloat := secureRandFloat64()
+ randomOffset := (randomFloat - 0.5) * 2 * jitterRange
+
+ jitteredDuration := time.Duration(float64(backoff) + randomOffset)
+ if jitteredDuration < 0 {
+ jitteredDuration = baseBackoff / 10 // Minimum backoff
+ }
+ if jitteredDuration > maxBackoff {
+ jitteredDuration = maxBackoff
+ }
+
+ return jitteredDuration
+}
+
+// isTransientError checks if the error is considered transient and retryable.
+func isTransientError(err error) bool {
+ if err == nil {
+ return false
+ }
+
+ if s, ok := status.FromError(err); ok {
+ return retryableCodes[s.Code()]
+ }
+
+ return false
+}
+
func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Metadata) bool {
node, ok := md.Spec.(*databasev1.Node)
if !ok {
@@ -276,8 +424,9 @@ func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Met
p.deleteClient(md)
go func(p *pub, name string, en evictNode, md schema.Metadata) {
defer p.closer.Done()
- backoff := initBackoff
+ attempt := 0
for {
+ backoff := jitteredBackoff(initBackoff, maxBackoff, attempt, defaultJitterFactor)
select {
case <-time.After(backoff):
credOpts, errEvict := p.getClientTransportCredentials()
@@ -314,11 +463,7 @@ func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Met
case <-p.closer.CloseNotify():
return
}
- if backoff < maxBackoff {
- backoff *= time.Duration(backoffMultiplier)
- } else {
- backoff = maxBackoff
- }
+ attempt++
}
}(p, name, p.evictable[name], md)
return false
@@ -426,8 +571,9 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
}
p.writableProbeMu.Unlock()
}()
- backoff := initBackoff
+ attempt := 0
for {
+ backoff := jitteredBackoff(initBackoff, maxBackoff, attempt, defaultJitterFactor)
select {
case <-time.After(backoff):
if errInternal := p.checkServiceHealth(t, node.conn); errInternal == nil {
@@ -448,11 +594,7 @@ func (p *pub) checkWritable(n string, topic bus.Topic) (bool, *common.Error) {
case <-p.closer.CloseNotify():
return
}
- if backoff < maxBackoff {
- backoff *= time.Duration(backoffMultiplier)
- } else {
- backoff = maxBackoff
- }
+ attempt++
}
}(n, topicStr)
return false, err
@@ -511,7 +653,7 @@ func (p *pub) isRequestAllowed(node string) bool {
return true
case StateOpen:
// Check if cooldown period has expired
- if time.Since(cb.openTime) >= 60*time.Second { // TODO: make configurable
+ if time.Since(cb.openTime) >= defaultCBResetTimeout {
// Transition to Half-Open to allow probe requests
cb.state = StateHalfOpen
p.log.Info().Str("node", node).Msg("circuit breaker transitioned to half-open")
@@ -525,6 +667,7 @@ func (p *pub) isRequestAllowed(node string) bool {
return true
}
}
+
// recordSuccess resets the circuit breaker state to Closed on successful operation.
// This handles Half-Open -> Closed transitions.
func (p *pub) recordSuccess(node string) {
@@ -568,7 +711,7 @@ func (p *pub) recordFailure(node string) {
}
// Check if we should open the circuit
- threshold := 5 // TODO: make configurable
+ threshold := defaultCBThreshold
if cb.consecutiveFailures >= threshold && cb.state == StateClosed {
cb.state = StateOpen
cb.openTime = time.Now()
diff --git a/banyand/queue/pub/retry_test.go b/banyand/queue/pub/retry_test.go
new file mode 100644
index 00000000..979eb167
--- /dev/null
+++ b/banyand/queue/pub/retry_test.go
@@ -0,0 +1,528 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pub
+
+import (
+ "context"
+ "fmt"
+ "math"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+
+ clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+)
+
+//go:generate mockgen -source=../../../api/proto/banyandb/cluster/v1/rpc_grpc.pb.go -destination=mock_send_client.go -package=pub
+
+// MockSendClient implements the clusterv1.Service_SendClient interface for testing.
+type MockSendClient struct {
+ ctx context.Context
+ sendFunc func(*clusterv1.SendRequest) error
+ recvFunc func() (*clusterv1.SendResponse, error)
+ closeSendFunc func() error
+}
+
+func NewMockSendClient(ctx context.Context) *MockSendClient {
+ return &MockSendClient{ctx: ctx}
+}
+
+func (m *MockSendClient) Send(req *clusterv1.SendRequest) error {
+ if m.sendFunc != nil {
+ return m.sendFunc(req)
+ }
+ return nil
+}
+
+func (m *MockSendClient) Recv() (*clusterv1.SendResponse, error) {
+ if m.recvFunc != nil {
+ return m.recvFunc()
+ }
+ return nil, nil
+}
+
+func (m *MockSendClient) CloseSend() error {
+ if m.closeSendFunc != nil {
+ return m.closeSendFunc()
+ }
+ return nil
+}
+
+func (m *MockSendClient) Context() context.Context {
+ return m.ctx
+}
+
+func (m *MockSendClient) Header() (metadata.MD, error) {
+ return metadata.MD{}, nil
+}
+
+func (m *MockSendClient) Trailer() metadata.MD {
+ return metadata.MD{}
+}
+
+func (m *MockSendClient) SendMsg(_ interface{}) error {
+ return nil
+}
+
+func (m *MockSendClient) RecvMsg(_ interface{}) error {
+ return nil
+}
+
+func (m *MockSendClient) SetSendFunc(f func(*clusterv1.SendRequest) error) {
+ m.sendFunc = f
+}
+
+func (m *MockSendClient) SetRecvFunc(f func() (*clusterv1.SendResponse, error)) {
+ m.recvFunc = f
+}
+
+func (m *MockSendClient) SetCloseSendFunc(f func() error) {
+ m.closeSendFunc = f
+}
+
+func TestJitteredBackoff(t *testing.T) {
+ tests := []struct {
+ name string
+ baseBackoff time.Duration
+ maxBackoff time.Duration
+ attempt int
+ jitterFactor float64
+ expectedMin time.Duration
+ expectedMax time.Duration
+ }{
+ {
+ name: "zero_attempt",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 0,
+ jitterFactor: 0.2,
+ expectedMin: 80 * time.Millisecond,
+ expectedMax: 120 * time.Millisecond,
+ },
+ {
+ name: "first_attempt",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 1,
+ jitterFactor: 0.2,
+ expectedMin: 160 * time.Millisecond,
+ expectedMax: 240 * time.Millisecond,
+ },
+ {
+ name: "max_backoff_reached",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 300 * time.Millisecond,
+ attempt: 10,
+ jitterFactor: 0.2,
+ expectedMin: 240 * time.Millisecond,
+ expectedMax: 360 * time.Millisecond,
+ },
+ {
+ name: "no_jitter",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 0,
+ jitterFactor: 0.0,
+ expectedMin: 100 * time.Millisecond,
+ expectedMax: 100 * time.Millisecond,
+ },
+ {
+ name: "max_jitter",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 0,
+ jitterFactor: 1.0,
+ expectedMin: 0,
+ expectedMax: 200 * time.Millisecond,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Test multiple times to account for randomness
+ for i := 0; i < 100; i++ {
+ result := jitteredBackoff(tt.baseBackoff, tt.maxBackoff, tt.attempt, tt.jitterFactor)
+
+ assert.GreaterOrEqual(t, result, tt.expectedMin,
+ "result %v should be >= expected min %v", result, tt.expectedMin)
+ assert.LessOrEqual(t, result, tt.expectedMax,
+ "result %v should be <= expected max %v", result, tt.expectedMax)
+ }
+ })
+ }
+}
+
+func TestJitteredBackoffEdgeCases(t *testing.T) {
+ tests := []struct {
+ name string
+ baseBackoff time.Duration
+ maxBackoff time.Duration
+ attempt int
+ jitterFactor float64
+ }{
+ {
+ name: "negative_jitter_factor",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 0,
+ jitterFactor: -0.5,
+ },
+ {
+ name: "jitter_factor_greater_than_one",
+ baseBackoff: 100 * time.Millisecond,
+ maxBackoff: 10 * time.Second,
+ attempt: 0,
+ jitterFactor: 1.5,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ // Should not panic and should return a reasonable value
+ result := jitteredBackoff(tt.baseBackoff, tt.maxBackoff, tt.attempt, tt.jitterFactor)
+ assert.Greater(t, result, time.Duration(0), "result should be positive")
+ assert.LessOrEqual(t, result, tt.maxBackoff, "result should not exceed max backoff")
+ })
+ }
+}
+
+func TestIsTransientError(t *testing.T) {
+ tests := []struct {
+ err error
+ name string
+ expectRetry bool
+ }{
+ {
+ name: "nil_error",
+ err: nil,
+ expectRetry: false,
+ },
+ {
+ name: "unavailable_error",
+ err: status.Error(codes.Unavailable, "service unavailable"),
+ expectRetry: true,
+ },
+ {
+ name: "deadline_exceeded_error",
+ err: status.Error(codes.DeadlineExceeded, "deadline exceeded"),
+ expectRetry: true,
+ },
+ {
+ name: "resource_exhausted_error",
+ err: status.Error(codes.ResourceExhausted, "rate limited"),
+ expectRetry: true,
+ },
+ {
+ name: "not_found_error",
+ err: status.Error(codes.NotFound, "not found"),
+ expectRetry: false,
+ },
+ {
+ name: "permission_denied_error",
+ err: status.Error(codes.PermissionDenied, "permission denied"),
+ expectRetry: false,
+ },
+ {
+ name: "invalid_argument_error",
+ err: status.Error(codes.InvalidArgument, "invalid argument"),
+ expectRetry: false,
+ },
+ {
+ name: "generic_error",
+ err: fmt.Errorf("generic error"),
+ expectRetry: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ result := isTransientError(tt.err)
+ assert.Equal(t, tt.expectRetry, result, "transient error classification mismatch")
+ })
+ }
+}
+
+func TestRetrySendSuccess(t *testing.T) {
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock successful send on first attempt
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ return nil
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.NoError(t, err, "successful send should not return error")
+}
+
+func TestRetrySendTransientErrorWithRecovery(t *testing.T) {
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock transient error on first attempt, success on second
+ callCount := 0
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ callCount++
+ if callCount == 1 {
+ return status.Error(codes.Unavailable, "unavailable")
+ }
+ return nil
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.NoError(t, err, "should succeed after retry")
+}
+
+func TestRetrySendNonTransientError(t *testing.T) {
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock non-transient error
+ nonTransientErr := status.Error(codes.InvalidArgument, "invalid argument")
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ return nonTransientErr
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.Error(t, err, "non-transient error should be returned immediately")
+ assert.Equal(t, nonTransientErr, err, "should return the original error")
+}
+
+func TestRetrySendExhaustedRetries(t *testing.T) {
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock transient error on all attempts
+ transientErr := status.Error(codes.Unavailable, "unavailable")
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ return transientErr
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.Error(t, err, "should return error after exhausting retries")
+ assert.Contains(t, err.Error(), "retry exhausted", "error should indicate retry exhaustion")
+ assert.Contains(t, err.Error(), "test-node", "error should contain node name")
+}
+
+func TestRetrySendContextCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ mockStream := NewMockSendClient(context.Background())
+
+ // Cancel context before retry
+ cancel()
+
+ // Mock transient error (may or may not be called)
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ return status.Error(codes.Unavailable, "unavailable")
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.Error(t, err, "should return error when context is canceled")
+ assert.Equal(t, context.Canceled, err, "should return context cancellation error")
+}
+
+func TestRetrySendStreamContextDone(t *testing.T) {
+ ctx := context.Background()
+ streamCtx, cancel := context.WithCancel(context.Background())
+
+ // Cancel stream context
+ cancel()
+
+ mockStream := NewMockSendClient(streamCtx)
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+
+ assert.Error(t, err, "should return error when stream context is done")
+ assert.Equal(t, context.Canceled, err, "should return stream context cancellation error")
+}
+
+func TestRetrySendPerAttemptTimeout(t *testing.T) {
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock slow send that would trigger per-attempt timeout
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ time.Sleep(defaultPerRequestTimeout + 100*time.Millisecond)
+ return nil
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ start := time.Now()
+ _ = bp.retrySend(ctx, mockStream, req, "test-node")
+ duration := time.Since(start)
+
+ // Should timeout quickly due to per-attempt timeout, not wait for the full operation
+ assert.Less(t, duration, defaultPerRequestTimeout*time.Duration(defaultMaxRetries+1)*2,
+ "should timeout relatively quickly due to per-attempt timeouts")
+}
+
+func TestRetrySendBackoffTiming(t *testing.T) {
+ if testing.Short() {
+ t.Skip("skipping timing test in short mode")
+ }
+
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock transient errors for first few attempts, then success
+ transientErr := status.Error(codes.Unavailable, "unavailable")
+ callCount := 0
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ callCount++
+ if callCount <= 2 {
+ return transientErr
+ }
+ return nil
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ start := time.Now()
+ err := bp.retrySend(ctx, mockStream, req, "test-node")
+ duration := time.Since(start)
+
+ assert.NoError(t, err, "should eventually succeed")
+
+ // Should have some backoff delay but not too much (considering jitter)
+ expectedMinDelay := defaultBackoffBase // First backoff
+ expectedMaxDelay := (defaultBackoffBase + defaultBackoffBase*2) * 2 // With max jitter
+
+ assert.GreaterOrEqual(t, duration, expectedMinDelay/2,
+ "should have some backoff delay")
+ assert.LessOrEqual(t, duration, expectedMaxDelay,
+ "should not take too long with reasonable backoff")
+}
+
+func TestRetrySendConcurrency(t *testing.T) {
+ const numGoroutines = 50
+
+ var wg sync.WaitGroup
+ errors := make(chan error, numGoroutines)
+
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func(id int) {
+ defer wg.Done()
+
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+
+ // Mock some transient errors followed by success
+ transientErr := status.Error(codes.Unavailable, "unavailable")
+ callCount := 0 // Each goroutine has its own callCount
+ mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ callCount++
+ if callCount <= 2 { // Fail for first 2 attempts, succeed on 3rd (within defaultMaxRetries=3)
+ return transientErr
+ }
+ return nil // Succeed on final attempt
+ })
+
+ bp := &batchPublisher{}
+ req := &clusterv1.SendRequest{}
+
+ err := bp.retrySend(ctx, mockStream, req, fmt.Sprintf("test-node-%d", id))
+ errors <- err
+ }(i)
+ }
+
+ wg.Wait()
+ close(errors)
+
+ successCount := 0
+ failureCount := 0
+ for err := range errors {
+ if err == nil {
+ successCount++
+ } else {
+ failureCount++
+ }
+ }
+
+ assert.Equal(t, numGoroutines, successCount, "all goroutines should succeed eventually (got %d successes, %d failures)", successCount, failureCount)
+}
+
+func TestBackoffDistribution(t *testing.T) {
+ // Test that jitter produces a reasonable distribution
+ const numSamples = 1000
+ baseBackoff := 100 * time.Millisecond
+ maxBackoff := 10 * time.Second
+ attempt := 0
+ jitterFactor := 0.2
+
+ durations := make([]time.Duration, numSamples)
+ for i := 0; i < numSamples; i++ {
+ durations[i] = jitteredBackoff(baseBackoff, maxBackoff, attempt, jitterFactor)
+ }
+
+ // Calculate mean and standard deviation
+ var sum time.Duration
+ for _, d := range durations {
+ sum += d
+ }
+ mean := sum / time.Duration(numSamples)
+
+ var sumSquaredDiffs float64
+ for _, d := range durations {
+ diff := float64(d - mean)
+ sumSquaredDiffs += diff * diff
+ }
+ variance := sumSquaredDiffs / float64(numSamples)
+ stdDev := time.Duration(math.Sqrt(variance))
+
+ // Expected mean should be around baseBackoff (100ms)
+ expectedMean := baseBackoff
+ assert.InDelta(t, float64(expectedMean), float64(mean), float64(baseBackoff)/10,
+ "mean should be close to base backoff")
+
+ // Standard deviation should be reasonable (not too low, indicating proper jitter)
+ minStdDev := time.Duration(float64(baseBackoff) * jitterFactor / 4)
+ assert.Greater(t, stdDev, minStdDev,
+ "standard deviation should indicate proper jitter is applied")
+}
(skywalking-banyandb) 05/11: Refactor circuit breaker tests and enhance probe logic
Posted by 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 971f583ee0dfb1ae3e3d00aacf0f260f6c443e30
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 17:04:56 2025 +0800
Refactor circuit breaker tests and enhance probe logic
- Introduced a constant for the test node name to improve code readability and maintainability.
- Updated circuit breaker tests to use the new constant, ensuring consistency across test cases.
---
banyand/queue/pub/circuitbreaker_test.go | 236 ++++++++++++++++++++++++++++---
banyand/queue/pub/client.go | 22 ++-
2 files changed, 235 insertions(+), 23 deletions(-)
diff --git a/banyand/queue/pub/circuitbreaker_test.go b/banyand/queue/pub/circuitbreaker_test.go
index 9f1976b1..73181b96 100644
--- a/banyand/queue/pub/circuitbreaker_test.go
+++ b/banyand/queue/pub/circuitbreaker_test.go
@@ -28,6 +28,8 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
+const testNodeName = "test-node"
+
func TestCircuitBreakerStateTransitions(t *testing.T) {
tests := []struct {
name string
@@ -39,7 +41,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
{
name: "closed_to_open_after_failures",
setup: func(p *pub) {
- p.recordSuccess("test-node")
+ p.recordSuccess(testNodeName)
},
actions: []func(*pub, string){
func(p *pub, node string) { p.recordFailure(node) },
@@ -54,7 +56,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
{
name: "closed_remains_closed_below_threshold",
setup: func(p *pub) {
- p.recordSuccess("test-node")
+ p.recordSuccess(testNodeName)
},
actions: []func(*pub, string){
func(p *pub, node string) { p.recordFailure(node) },
@@ -67,14 +69,14 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
{
name: "open_to_half_open_after_cooldown",
setup: func(p *pub) {
- p.recordSuccess("test-node")
+ p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure("test-node")
+ p.recordFailure(testNodeName)
}
// Simulate cooldown period has passed
p.cbMu.Lock()
- cb := p.cbStates["test-node"]
+ cb := p.cbStates[testNodeName]
cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
p.cbMu.Unlock()
},
@@ -85,14 +87,14 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
{
name: "half_open_to_closed_on_success",
setup: func(p *pub) {
- p.recordSuccess("test-node")
+ p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure("test-node")
+ p.recordFailure(testNodeName)
}
// Set to half-open state
p.cbMu.Lock()
- cb := p.cbStates["test-node"]
+ cb := p.cbStates[testNodeName]
cb.state = StateHalfOpen
p.cbMu.Unlock()
},
@@ -105,14 +107,14 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
{
name: "half_open_to_open_on_failure",
setup: func(p *pub) {
- p.recordSuccess("test-node")
+ p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure("test-node")
+ p.recordFailure(testNodeName)
}
// Set to half-open state
p.cbMu.Lock()
- cb := p.cbStates["test-node"]
+ cb := p.cbStates[testNodeName]
cb.state = StateHalfOpen
p.cbMu.Unlock()
},
@@ -139,19 +141,19 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
// Execute actions
for _, action := range tt.actions {
- action(p, "test-node")
+ action(p, testNodeName)
}
// Check final state
p.cbMu.RLock()
- cb, exists := p.cbStates["test-node"]
+ cb, exists := p.cbStates[testNodeName]
p.cbMu.RUnlock()
require.True(t, exists, "circuit breaker state should exist")
assert.Equal(t, tt.expectedState, cb.state, "circuit breaker state mismatch")
// Check if requests are allowed
- allowed := p.isRequestAllowed("test-node")
+ allowed := p.isRequestAllowed(testNodeName)
assert.Equal(t, tt.allowsRequests, allowed, "request allowance mismatch")
})
}
@@ -166,7 +168,7 @@ func TestCircuitBreakerConcurrency(t *testing.T) {
const numGoroutines = 100
const numOperations = 50
- node := "test-node"
+ node := testNodeName
var wg sync.WaitGroup
wg.Add(numGoroutines * 2) // Half for success, half for failure
@@ -260,7 +262,7 @@ func TestCircuitBreakerRecoveryAfterCooldown(t *testing.T) {
log: logger.GetLogger("test"),
}
- node := "test-node"
+ node := testNodeName
// Initialize node
p.recordSuccess(node)
@@ -329,7 +331,7 @@ func TestCircuitBreakerFailureThresholdEdgeCase(t *testing.T) {
log: logger.GetLogger("test"),
}
- node := "test-node"
+ node := testNodeName
// Initialize node
p.recordSuccess(node)
@@ -354,3 +356,203 @@ func TestCircuitBreakerFailureThresholdEdgeCase(t *testing.T) {
assert.Equal(t, defaultCBThreshold, cb.consecutiveFailures, "failure count should be at threshold")
p.cbMu.RUnlock()
}
+
+func TestCircuitBreakerSingleProbeEnforcement(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Initialize and trip circuit breaker
+ p.recordSuccess(node)
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node)
+ }
+
+ // Verify circuit is open
+ assert.False(t, p.isRequestAllowed(node), "circuit should be open")
+
+ // Simulate cooldown period passage
+ p.cbMu.Lock()
+ cb := p.cbStates[node]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+
+ // First request should transition to half-open and be allowed
+ allowed1 := p.isRequestAllowed(node)
+ assert.True(t, allowed1, "first request after cooldown should be allowed and transition to half-open")
+
+ // Verify state is half-open with probe in flight
+ p.cbMu.RLock()
+ assert.Equal(t, StateHalfOpen, cb.state, "circuit should be in half-open state")
+ assert.True(t, cb.halfOpenProbeInFlight, "probe should be marked as in flight")
+ p.cbMu.RUnlock()
+
+ // Second request should be denied while probe is in flight
+ allowed2 := p.isRequestAllowed(node)
+ assert.False(t, allowed2, "second request should be denied while probe is in flight")
+
+ // Third request should also be denied
+ allowed3 := p.isRequestAllowed(node)
+ assert.False(t, allowed3, "third request should also be denied while probe is in flight")
+}
+
+func TestCircuitBreakerSingleProbeSuccess(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Setup half-open state with probe in flight
+ p.recordSuccess(node)
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node)
+ }
+
+ p.cbMu.Lock()
+ cb := p.cbStates[node]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+
+ // Transition to half-open
+ allowed := p.isRequestAllowed(node)
+ assert.True(t, allowed, "first request should be allowed")
+
+ // Verify probe is in flight
+ p.cbMu.RLock()
+ assert.True(t, cb.halfOpenProbeInFlight, "probe should be in flight")
+ p.cbMu.RUnlock()
+
+ // Record success (probe succeeds)
+ p.recordSuccess(node)
+
+ // Verify circuit is closed and probe token is cleared
+ p.cbMu.RLock()
+ assert.Equal(t, StateClosed, cb.state, "circuit should be closed after successful probe")
+ assert.False(t, cb.halfOpenProbeInFlight, "probe token should be cleared")
+ p.cbMu.RUnlock()
+
+ // Subsequent requests should be allowed in closed state
+ assert.True(t, p.isRequestAllowed(node), "requests should be allowed in closed state")
+}
+
+func TestCircuitBreakerSingleProbeFailure(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Setup half-open state with probe in flight
+ p.recordSuccess(node)
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node)
+ }
+
+ p.cbMu.Lock()
+ cb := p.cbStates[node]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+
+ // Transition to half-open
+ allowed := p.isRequestAllowed(node)
+ assert.True(t, allowed, "first request should be allowed")
+
+ // Verify probe is in flight
+ p.cbMu.RLock()
+ assert.True(t, cb.halfOpenProbeInFlight, "probe should be in flight")
+ p.cbMu.RUnlock()
+
+ // Record failure (probe fails)
+ p.recordFailure(node)
+
+ // Verify circuit is back to open and probe token is cleared
+ p.cbMu.RLock()
+ assert.Equal(t, StateOpen, cb.state, "circuit should be back to open after failed probe")
+ assert.False(t, cb.halfOpenProbeInFlight, "probe token should be cleared")
+ p.cbMu.RUnlock()
+
+ // Subsequent requests should be denied in open state
+ assert.False(t, p.isRequestAllowed(node), "requests should be denied in open state")
+}
+
+func TestCircuitBreakerConcurrentProbeAttempts(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Setup open state ready for half-open transition
+ p.recordSuccess(node)
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node)
+ }
+
+ p.cbMu.Lock()
+ cb := p.cbStates[node]
+ cb.openTime = time.Now().Add(-defaultCBResetTimeout - time.Second)
+ p.cbMu.Unlock()
+
+ const numGoroutines = 10
+ var wg sync.WaitGroup
+ results := make(chan bool, numGoroutines)
+
+ // Simulate concurrent requests attempting to probe
+ for i := 0; i < numGoroutines; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ allowed := p.isRequestAllowed(node)
+ results <- allowed
+ }()
+ }
+
+ wg.Wait()
+ close(results)
+
+ // Count allowed requests
+ allowedCount := 0
+ for result := range results {
+ if result {
+ allowedCount++
+ }
+ }
+
+ // Only one request should be allowed (the one that set the probe token)
+ assert.Equal(t, 1, allowedCount, "exactly one request should be allowed in half-open state")
+
+ // Verify circuit is in half-open with probe in flight
+ p.cbMu.RLock()
+ assert.Equal(t, StateHalfOpen, cb.state, "circuit should be in half-open state")
+ assert.True(t, cb.halfOpenProbeInFlight, "probe should be marked as in flight")
+ p.cbMu.RUnlock()
+}
+
+func TestCircuitBreakerProbeTokenInitialization(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Test that new circuit breaker states have probe token cleared
+ p.recordSuccess(node)
+
+ p.cbMu.RLock()
+ cb := p.cbStates[node]
+ assert.False(t, cb.halfOpenProbeInFlight, "new circuit breaker should have probe token cleared")
+ p.cbMu.RUnlock()
+}
diff --git a/banyand/queue/pub/client.go b/banyand/queue/pub/client.go
index 73f44491..cf665dc3 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -160,10 +160,11 @@ const (
// circuitState holds circuit breaker metadata; it does NOT duplicate gRPC clients/conns.
type circuitState struct {
- lastFailureTime time.Time
- openTime time.Time
- state CircuitState
- consecutiveFailures int
+ lastFailureTime time.Time
+ openTime time.Time
+ state CircuitState
+ consecutiveFailures int
+ halfOpenProbeInFlight bool
}
type client struct {
@@ -654,14 +655,21 @@ func (p *pub) isRequestAllowed(node string) bool {
case StateOpen:
// Check if cooldown period has expired
if time.Since(cb.openTime) >= defaultCBResetTimeout {
- // Transition to Half-Open to allow probe requests
+ // Transition to Half-Open to allow a single probe request
cb.state = StateHalfOpen
+ cb.halfOpenProbeInFlight = true // Set token for the probe
p.log.Info().Str("node", node).Msg("circuit breaker transitioned to half-open")
return true
}
return false // Still in cooldown period
case StateHalfOpen:
- // Allow requests in half-open state (probe requests)
+ // In half-open state, deny requests if probe is already in flight
+ if cb.halfOpenProbeInFlight {
+ return false // Probe already in progress, deny additional requests
+ }
+ // This case should not normally happen since we set the token on transition,
+ // but handle it defensively by allowing the request and setting the token
+ cb.halfOpenProbeInFlight = true
return true
default:
return true
@@ -689,6 +697,7 @@ func (p *pub) recordSuccess(node string) {
cb.consecutiveFailures = 0
cb.lastFailureTime = time.Time{}
cb.openTime = time.Time{}
+ cb.halfOpenProbeInFlight = false // Clear probe token
}
// recordFailure updates the circuit breaker state on failed operation.
@@ -720,6 +729,7 @@ func (p *pub) recordFailure(node string) {
// Failed during half-open, go back to open
cb.state = StateOpen
cb.openTime = time.Now()
+ cb.halfOpenProbeInFlight = false // Clear probe token
p.log.Warn().Str("node", node).Msg("circuit breaker reopened after half-open failure")
}
}
(skywalking-banyandb) 08/11: Enhance circuit breaker logic in message publishing
Posted by 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 a23fdb91dde99032a9885c88a4bc47969195266d
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 18:31:04 2025 +0800
Enhance circuit breaker logic in message publishing
- Added checks for circuit breaker status before sending messages, preventing attempts when the circuit is open.
- Implemented failure and success recording for circuit breaker state management during message send operations, improving resilience and diagnostics.
---
banyand/queue/pub/pub.go | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index d8eaa07b..b7c17df0 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -233,6 +233,12 @@ func (p *pub) publish(timeout time.Duration, topic bus.Topic, messages ...bus.Me
return multierr.Append(err, fmt.Errorf("failed to marshal message[%d]: %w", m.ID(), errSend))
}
node := m.Node()
+
+ // Check circuit breaker before attempting send
+ if !p.isRequestAllowed(node) {
+ return multierr.Append(err, fmt.Errorf("circuit breaker open for node %s", node))
+ }
+
p.mu.RLock()
client, ok := p.active[node]
p.mu.RUnlock()
@@ -243,12 +249,18 @@ func (p *pub) publish(timeout time.Duration, topic bus.Topic, messages ...bus.Me
f.cancelFn = append(f.cancelFn, cancel)
stream, errCreateStream := client.client.Send(ctx)
if errCreateStream != nil {
+ // Record failure for circuit breaker (only for transient/internal errors)
+ p.recordFailure(node, errCreateStream)
return multierr.Append(err, fmt.Errorf("failed to get stream for node %s: %w", node, errCreateStream))
}
errSend = stream.Send(r)
if errSend != nil {
+ // Record failure for circuit breaker (only for transient/internal errors)
+ p.recordFailure(node, errSend)
return multierr.Append(err, fmt.Errorf("failed to send message to node %s: %w", node, errSend))
}
+ // Record success for circuit breaker
+ p.recordSuccess(node)
f.clients = append(f.clients, stream)
f.topics = append(f.topics, topic)
f.nodes = append(f.nodes, node)
(skywalking-banyandb) 06/11: Enhance circuit breaker logic to record transient and internal errors
Posted by 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 81fbc3219b3b6003693f50a3b44d19e8278a20c6
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Thu Sep 4 17:30:48 2025 +0800
Enhance circuit breaker logic to record transient and internal errors
- Updated the `recordFailure` method to only log failures for transient and internal errors, improving the accuracy of circuit breaker state management.
- Introduced new error handling in the batch publisher to ensure failures are recorded with the associated error for better diagnostics.
- Enhanced circuit breaker tests to validate the behavior with transient and non-retryable errors, ensuring robust error filtering and state transitions.
---
banyand/queue/pub/batch.go | 4 +-
banyand/queue/pub/circuitbreaker_test.go | 120 +++++++++++++++++++++++++------
banyand/queue/pub/client.go | 22 +++++-
3 files changed, 119 insertions(+), 27 deletions(-)
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index f0b9198e..f1cacc94 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -94,8 +94,8 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
errSend := bp.retrySend(ctx, stream.client, r, node)
if errSend != nil {
err = multierr.Append(err, fmt.Errorf("failed to send message to node %s: %w", node, errSend))
- // Record failure for circuit breaker
- bp.pub.recordFailure(node)
+ // Record failure for circuit breaker (only for transient/internal errors)
+ bp.pub.recordFailure(node, errSend)
return false
}
// Record success for circuit breaker
diff --git a/banyand/queue/pub/circuitbreaker_test.go b/banyand/queue/pub/circuitbreaker_test.go
index 73181b96..b22fc06e 100644
--- a/banyand/queue/pub/circuitbreaker_test.go
+++ b/banyand/queue/pub/circuitbreaker_test.go
@@ -24,12 +24,21 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
"github.com/apache/skywalking-banyandb/pkg/logger"
)
const testNodeName = "test-node"
+var (
+ // Test errors for circuit breaker testing.
+ errTransient = status.Error(codes.Unavailable, "service unavailable")
+ errInternal = status.Error(codes.Internal, "internal server error")
+ errNonRetry = status.Error(codes.InvalidArgument, "invalid argument")
+)
+
func TestCircuitBreakerStateTransitions(t *testing.T) {
tests := []struct {
name string
@@ -44,11 +53,11 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.recordSuccess(testNodeName)
},
actions: []func(*pub, string){
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
},
expectedState: StateOpen,
allowsRequests: false,
@@ -59,9 +68,9 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.recordSuccess(testNodeName)
},
actions: []func(*pub, string){
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
- func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
},
expectedState: StateClosed,
allowsRequests: true,
@@ -72,7 +81,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(testNodeName)
+ p.recordFailure(testNodeName, errTransient)
}
// Simulate cooldown period has passed
p.cbMu.Lock()
@@ -90,7 +99,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(testNodeName)
+ p.recordFailure(testNodeName, errTransient)
}
// Set to half-open state
p.cbMu.Lock()
@@ -110,7 +119,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.recordSuccess(testNodeName)
// Trip the circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(testNodeName)
+ p.recordFailure(testNodeName, errTransient)
}
// Set to half-open state
p.cbMu.Lock()
@@ -119,7 +128,7 @@ func TestCircuitBreakerStateTransitions(t *testing.T) {
p.cbMu.Unlock()
},
actions: []func(*pub, string){
- func(p *pub, node string) { p.recordFailure(node) },
+ func(p *pub, node string) { p.recordFailure(node, errTransient) },
},
expectedState: StateOpen,
allowsRequests: false,
@@ -189,7 +198,7 @@ func TestCircuitBreakerConcurrency(t *testing.T) {
go func() {
defer wg.Done()
for j := 0; j < numOperations; j++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
p.isRequestAllowed(node)
}
}()
@@ -222,12 +231,12 @@ func TestCircuitBreakerMultipleNodes(t *testing.T) {
// Trip circuit breaker for node1 only
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure("node1")
+ p.recordFailure("node1", errTransient)
}
// Add some failures to node2 but below threshold
for i := 0; i < defaultCBThreshold-1; i++ {
- p.recordFailure("node2")
+ p.recordFailure("node2", errTransient)
}
// Keep node3 healthy
@@ -269,7 +278,7 @@ func TestCircuitBreakerRecoveryAfterCooldown(t *testing.T) {
// Trip circuit breaker
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
// Verify circuit is open
@@ -338,7 +347,7 @@ func TestCircuitBreakerFailureThresholdEdgeCase(t *testing.T) {
// Add failures just below threshold
for i := 0; i < defaultCBThreshold-1; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
// Circuit should still be closed
@@ -349,7 +358,7 @@ func TestCircuitBreakerFailureThresholdEdgeCase(t *testing.T) {
p.cbMu.RUnlock()
// One more failure should open the circuit
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
p.cbMu.RLock()
assert.Equal(t, StateOpen, cb.state, "circuit should be open after reaching threshold")
@@ -369,7 +378,7 @@ func TestCircuitBreakerSingleProbeEnforcement(t *testing.T) {
// Initialize and trip circuit breaker
p.recordSuccess(node)
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
// Verify circuit is open
@@ -412,7 +421,7 @@ func TestCircuitBreakerSingleProbeSuccess(t *testing.T) {
// Setup half-open state with probe in flight
p.recordSuccess(node)
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
p.cbMu.Lock()
@@ -454,7 +463,7 @@ func TestCircuitBreakerSingleProbeFailure(t *testing.T) {
// Setup half-open state with probe in flight
p.recordSuccess(node)
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
p.cbMu.Lock()
@@ -472,7 +481,7 @@ func TestCircuitBreakerSingleProbeFailure(t *testing.T) {
p.cbMu.RUnlock()
// Record failure (probe fails)
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
// Verify circuit is back to open and probe token is cleared
p.cbMu.RLock()
@@ -496,7 +505,7 @@ func TestCircuitBreakerConcurrentProbeAttempts(t *testing.T) {
// Setup open state ready for half-open transition
p.recordSuccess(node)
for i := 0; i < defaultCBThreshold; i++ {
- p.recordFailure(node)
+ p.recordFailure(node, errTransient)
}
p.cbMu.Lock()
@@ -556,3 +565,68 @@ func TestCircuitBreakerProbeTokenInitialization(t *testing.T) {
assert.False(t, cb.halfOpenProbeInFlight, "new circuit breaker should have probe token cleared")
p.cbMu.RUnlock()
}
+
+func TestCircuitBreakerErrorFiltering(t *testing.T) {
+ p := &pub{
+ cbStates: make(map[string]*circuitState),
+ cbMu: sync.RWMutex{},
+ log: logger.GetLogger("test"),
+ }
+
+ node := testNodeName
+
+ // Verify initial state is closed
+ assert.True(t, p.isRequestAllowed(node))
+ p.cbMu.Lock()
+ _, exists := p.cbStates[node]
+ p.cbMu.Unlock()
+ assert.False(t, exists)
+
+ // Try to trip circuit with non-retryable error
+ for i := 0; i < defaultCBThreshold*2; i++ {
+ p.recordFailure(node, errNonRetry)
+ }
+
+ // Circuit should still be closed (non-retryable errors don't count)
+ assert.True(t, p.isRequestAllowed(node))
+ p.cbMu.Lock()
+ cb, exists := p.cbStates[node]
+ p.cbMu.Unlock()
+ // State should not exist or be closed with 0 failures
+ if exists {
+ assert.Equal(t, StateClosed, cb.state)
+ assert.Equal(t, 0, cb.consecutiveFailures)
+ }
+
+ // Now try with transient errors - these should count
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node, errTransient)
+ }
+
+ // Circuit should now be open
+ 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 for internal error test
+ p.recordSuccess(node)
+ assert.True(t, p.isRequestAllowed(node))
+
+ // Try with internal errors - these should also count
+ for i := 0; i < defaultCBThreshold; i++ {
+ p.recordFailure(node, errInternal)
+ }
+
+ // 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 cf665dc3..2b090f0f 100644
--- a/banyand/queue/pub/client.go
+++ b/banyand/queue/pub/client.go
@@ -141,7 +141,7 @@ var (
codes.Aborted: false,
codes.OutOfRange: false,
codes.Unimplemented: false,
- codes.Internal: false,
+ codes.Internal: true, // Retryable - internal server error should participate in circuit breaker
codes.Unavailable: true, // Retryable - service temporarily unavailable
codes.DataLoss: false,
codes.Unauthenticated: false,
@@ -407,6 +407,19 @@ func isTransientError(err error) bool {
return false
}
+// isInternalError checks if the error is an internal server error.
+func isInternalError(err error) bool {
+ if err == nil {
+ return false
+ }
+
+ if s, ok := status.FromError(err); ok {
+ return s.Code() == codes.Internal
+ }
+
+ return false
+}
+
func (p *pub) checkClientHealthAndReconnect(conn *grpc.ClientConn, md schema.Metadata) bool {
node, ok := md.Spec.(*databasev1.Node)
if !ok {
@@ -701,7 +714,12 @@ func (p *pub) recordSuccess(node string) {
}
// recordFailure updates the circuit breaker state on failed operation.
-func (p *pub) recordFailure(node string) {
+// Only records failures for transient/internal errors that should count toward opening the circuit.
+func (p *pub) recordFailure(node string, err error) {
+ // Only record failure if the error is transient or internal
+ if !isTransientError(err) && !isInternalError(err) {
+ return
+ }
p.cbMu.Lock()
defer p.cbMu.Unlock()
(skywalking-banyandb) 07/11: Enhance error handling in batch publisher and circuit breaker logic
Posted by 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
}