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 2026/03/31 14:02:51 UTC
(skywalking-banyandb) branch main updated: Fix property schema client connection not stable after data node restarted (#1038)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 75bf904bd Fix property schema client connection not stable after data node restarted (#1038)
75bf904bd is described below
commit 75bf904bd1528b911a9c63f6f82edbc453d14926
Author: mrproliu <74...@qq.com>
AuthorDate: Tue Mar 31 22:02:45 2026 +0800
Fix property schema client connection not stable after data node restarted (#1038)
* Fix property schema client connection not stable after data node restarted
---
CHANGES.md | 2 +
banyand/backup/lifecycle/steps.go | 3 +-
banyand/metadata/client.go | 96 +++++------
banyand/metadata/schema/property/client.go | 189 +++++++++++++++------
banyand/metadata/schema/property/client_test.go | 118 +++++++++++++-
banyand/queue/pub/pub.go | 2 +-
docs/operation/configuration.md | 11 ++
pkg/grpchelper/connmanager.go | 74 ++++++---
pkg/grpchelper/connmanager_test.go | 208 ++++++++++++++++++++++++
pkg/grpchelper/helpers_test.go | 21 +++
10 files changed, 594 insertions(+), 130 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index c751121e9..7a3400353 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -10,12 +10,14 @@ Release Notes.
- Collect BanyanDB data on e2e test failure for CI debugging.
- Add log query e2e test.
- Sync lifecycle e2e test from SkyWalking stages test.
+- Add periodic health check for property schema connection.
### Bug Fixes
- Fix flaky trace query filtering caused by non-deterministic sidx tag ordering and add consistency checks for integration query cases.
- Fix index-mode measure queries returning documents outside the requested time range when a widened segment overlaps the query window.
- MCP: Add validation for properties and harden the mcp server.
+- Fix property schema client connection not stable after data node restarted.
## 0.10.0
diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go
index deba6d767..700897840 100644
--- a/banyand/backup/lifecycle/steps.go
+++ b/banyand/backup/lifecycle/steps.go
@@ -101,6 +101,7 @@ func (gc *GroupConfig) Close() {
}
}
+//nolint:contextcheck // health check goroutine uses context.Background()
func parseGroup(
g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node,
l *logger.Logger, metadata metadata.Repo, clusterStateMgr *clusterStateManager,
@@ -148,7 +149,7 @@ func parseGroup(
if ok, _ := nodeSel.OnInit([]schema.Kind{schema.KindGroup}); !ok {
return nil, fmt.Errorf("failed to initialize node selector for group %s", g.Metadata.Name)
}
- client := pub.NewWithoutMetadata()
+ client := pub.NewWithoutMetadata() //nolint:contextcheck // health check goroutine uses context.Background()
if g.Catalog == commonv1.Catalog_CATALOG_STREAM {
_ = grpc.NewClusterNodeRegistry(data.TopicStreamWrite, client, nodeSel)
} else {
diff --git a/banyand/metadata/client.go b/banyand/metadata/client.go
index 0608e19ce..9ae0dd359 100644
--- a/banyand/metadata/client.go
+++ b/banyand/metadata/client.go
@@ -98,44 +98,45 @@ func NewClient(toRegisterNode, forceRegisterNode bool) (Service, error) {
}
type clientService struct {
- schemaRegistry schema.Registry
- nodeDiscoveryRegistry schema.NodeDiscovery
- omr observability.MetricsRegistry
- dataBroadcaster bus.Broadcaster
- liaisonBroadcaster bus.Broadcaster
- infoCollectorRegistry *schema.InfoCollectorRegistry
- closer *run.Closer
- nodeInfo *databasev1.Node
- etcdTLSCertFile string
- etcdPassword string
- etcdTLSCAFile string
- etcdUsername string
- etcdTLSKeyFile string
- namespace string
- nodeDiscoveryMode string
- schemaRegistryMode string
- filePath string
- propertySchemaClientCACert string
- dnsCACertPaths []string
- dnsSRVAddresses []string
- endpoints []string
- registryTimeout time.Duration
- dnsFetchInitInterval time.Duration
- dnsFetchInitDuration time.Duration
- dnsFetchInterval time.Duration
- grpcTimeout time.Duration
- etcdFullSyncInterval time.Duration
- propertySchemaSyncInterval time.Duration
- fileFetchInterval time.Duration
- fileRetryInitialInterval time.Duration
- fileRetryMaxInterval time.Duration
- propertySchemaMaxRecvSize run.Bytes
- fileRetryMultiplier float64
- nodeInfoMux sync.Mutex
- forceRegisterNode bool
- toRegisterNode bool
- dnsTLSEnabled bool
- propertySchemaClientTLS bool
+ schemaRegistry schema.Registry
+ nodeDiscoveryRegistry schema.NodeDiscovery
+ omr observability.MetricsRegistry
+ dataBroadcaster bus.Broadcaster
+ liaisonBroadcaster bus.Broadcaster
+ infoCollectorRegistry *schema.InfoCollectorRegistry
+ closer *run.Closer
+ nodeInfo *databasev1.Node
+ etcdTLSCertFile string
+ etcdPassword string
+ etcdTLSCAFile string
+ etcdUsername string
+ etcdTLSKeyFile string
+ namespace string
+ nodeDiscoveryMode string
+ schemaRegistryMode string
+ filePath string
+ propertySchemaClientCACert string
+ dnsCACertPaths []string
+ dnsSRVAddresses []string
+ endpoints []string
+ registryTimeout time.Duration
+ dnsFetchInitInterval time.Duration
+ dnsFetchInitDuration time.Duration
+ dnsFetchInterval time.Duration
+ grpcTimeout time.Duration
+ etcdFullSyncInterval time.Duration
+ propertySchemaSyncInterval time.Duration
+ propertySchemaHealthCheckInterval time.Duration
+ fileFetchInterval time.Duration
+ fileRetryInitialInterval time.Duration
+ fileRetryMaxInterval time.Duration
+ propertySchemaMaxRecvSize run.Bytes
+ fileRetryMultiplier float64
+ nodeInfoMux sync.Mutex
+ forceRegisterNode bool
+ toRegisterNode bool
+ dnsTLSEnabled bool
+ propertySchemaClientTLS bool
}
func (s *clientService) SchemaRegistry() schema.Registry {
@@ -159,6 +160,8 @@ func (s *clientService) FlagSet() *run.FlagSet {
"Schema registry mode: 'etcd' for etcd-based storage, 'property' for property-based storage")
fs.DurationVar(&s.propertySchemaSyncInterval, "schema-property-client-sync-interval", property.DefaultSyncInterval,
"Polling interval for property-based schema sync")
+ fs.DurationVar(&s.propertySchemaHealthCheckInterval, "schema-property-client-health-check-interval", property.DefaultHealthCheckInterval,
+ "Interval for periodic connection health checks to schema servers. 0 uses default, negative disables")
s.propertySchemaMaxRecvSize = defaultRecvSize
fs.VarP(&s.propertySchemaMaxRecvSize, "schema-property-client-max-recv-msg-size", "",
"Max gRPC receive message size for property schema client")
@@ -426,14 +429,15 @@ func (s *clientService) initPropertySchemaRegistry(ctx context.Context, l *logge
}
}
cfg := &property.ClientConfig{
- GRPCTimeout: s.grpcTimeout,
- SyncInterval: s.propertySchemaSyncInterval,
- OMR: s.omr,
- CurNode: currentNode,
- NodeRegistry: s.NodeRegistry(),
- MaxRecvMsgSize: int(s.propertySchemaMaxRecvSize),
- TLSEnabled: s.propertySchemaClientTLS,
- CACertPath: s.propertySchemaClientCACert,
+ GRPCTimeout: s.grpcTimeout,
+ SyncInterval: s.propertySchemaSyncInterval,
+ HealthCheckInterval: s.propertySchemaHealthCheckInterval,
+ OMR: s.omr,
+ CurNode: currentNode,
+ NodeRegistry: s.NodeRegistry(),
+ MaxRecvMsgSize: int(s.propertySchemaMaxRecvSize),
+ TLSEnabled: s.propertySchemaClientTLS,
+ CACertPath: s.propertySchemaClientCACert,
}
for attempt := 1; attempt <= propertyRegistryInitRetryCount; attempt++ {
registry, createErr := property.NewSchemaRegistryClient(cfg) //nolint:contextcheck // healthCheck uses its own 2s timeout via context.Background()
diff --git a/banyand/metadata/schema/property/client.go b/banyand/metadata/schema/property/client.go
index 19b6d42fe..0d7f8d381 100644
--- a/banyand/metadata/schema/property/client.go
+++ b/banyand/metadata/schema/property/client.go
@@ -54,6 +54,10 @@ const (
DefaultSyncInterval = 30 * time.Second
// DefaultInitWaitTime is the default maximum time to wait for at least one schema server to become active during initialization.
DefaultInitWaitTime = time.Minute
+ // DefaultHealthCheckInterval is the default interval for periodic connection health checks.
+ DefaultHealthCheckInterval = 10 * time.Second
+ // defaultSyncTimeout is the default timeout for waiting for a sync response from a single node.
+ defaultSyncTimeout = 30 * time.Second
)
var _ schema.Registry = (*SchemaRegistry)(nil)
@@ -62,16 +66,19 @@ var errNoActiveServers = fmt.Errorf("no active schema servers available")
// ClientConfig holds configuration for the property-based schema registry client.
type ClientConfig struct {
- OMR observability.MetricsRegistry
- NodeRegistry schema.Node
- CurNode *databasev1.Node
- CACertPath string
- GRPCTimeout time.Duration
- SyncInterval time.Duration
- InitWaitTime time.Duration
- FullReconcileEvery uint64
- MaxRecvMsgSize int
- TLSEnabled bool
+ OMR observability.MetricsRegistry
+ NodeRegistry schema.Node
+ CurNode *databasev1.Node
+ CACertPath string
+ GRPCTimeout time.Duration
+ SyncInterval time.Duration
+ SyncTimeout time.Duration
+ WatchMaxBackoff time.Duration
+ InitWaitTime time.Duration
+ HealthCheckInterval time.Duration
+ FullReconcileEvery uint64
+ MaxRecvMsgSize int
+ TLSEnabled bool
}
type syncRequest struct {
@@ -79,10 +86,14 @@ type syncRequest struct {
tagProjection []string
}
+type syncMessage struct {
+ responseCh chan []*digestEntry
+ syncRequest
+}
+
type watchSession struct {
cancelFn context.CancelFunc
- syncReqCh chan *syncRequest
- syncResCh chan []*digestEntry
+ syncReqCh chan *syncMessage
}
type digestEntry struct {
@@ -107,6 +118,8 @@ type SchemaRegistry struct {
handlers map[schema.Kind][]schema.EventHandler
watchSessions map[string]*watchSession
syncInterval time.Duration
+ syncTimeout time.Duration
+ watchMaxBackoff time.Duration
fullReconcileEvery uint64
syncRound uint64
mux sync.RWMutex
@@ -129,10 +142,17 @@ func NewSchemaRegistryClient(cfg *ClientConfig) (*SchemaRegistry, error) {
caCertReloader: caCertReloader,
tlsEnabled: cfg.TLSEnabled,
}
+ healthCheckInterval := cfg.HealthCheckInterval
+ if healthCheckInterval == 0 {
+ healthCheckInterval = DefaultHealthCheckInterval
+ } else if healthCheckInterval < 0 {
+ healthCheckInterval = 0 // negative means disabled
+ }
connMgr := grpchelper.NewConnManager[*schemaClient](grpchelper.ConnManagerConfig[*schemaClient]{
- Handler: handler,
- Logger: l,
- MaxRecvMsgSize: cfg.MaxRecvMsgSize,
+ Handler: handler,
+ Logger: l,
+ MaxRecvMsgSize: cfg.MaxRecvMsgSize,
+ HealthCheckInterval: healthCheckInterval,
})
syncInterval := cfg.SyncInterval
if syncInterval == 0 {
@@ -146,6 +166,14 @@ func NewSchemaRegistryClient(cfg *ClientConfig) (*SchemaRegistry, error) {
if fullReconcileEvery <= 0 {
fullReconcileEvery = DefaultFullReconcileEvery
}
+ syncTimeout := cfg.SyncTimeout
+ if syncTimeout == 0 {
+ syncTimeout = defaultSyncTimeout
+ }
+ watchMaxBackoff := cfg.WatchMaxBackoff
+ if watchMaxBackoff == 0 {
+ watchMaxBackoff = defaultWatchMaxBackoff
+ }
reg := &SchemaRegistry{
connMgr: connMgr,
closer: run.NewCloser(1),
@@ -155,6 +183,8 @@ func NewSchemaRegistryClient(cfg *ClientConfig) (*SchemaRegistry, error) {
handlers: make(map[schema.Kind][]schema.EventHandler),
watchSessions: make(map[string]*watchSession),
syncInterval: syncInterval,
+ syncTimeout: syncTimeout,
+ watchMaxBackoff: watchMaxBackoff,
fullReconcileEvery: fullReconcileEvery,
}
handler.registry = reg
@@ -180,8 +210,6 @@ func NewSchemaRegistryClient(cfg *ClientConfig) (*SchemaRegistry, error) {
}
// Wait for at least one schema server to become active.
- // OnAddOrUpdate health-checks synchronously; if it fails, a background
- // retry goroutine retries backoff. Poll until active or timeout.
if connMgr.ActiveCount() == 0 {
waitDeadline := time.Now().Add(initWaitTime)
for connMgr.ActiveCount() == 0 && time.Now().Before(waitDeadline) {
@@ -939,7 +967,7 @@ func (r *SchemaRegistry) RegisterHandler(name string, kind schema.Kind, handler
}
}
r.l.Info().Str("name", name).Interface("kinds", kinds).Msg("registering property schema handler")
- handler.OnInit(kinds)
+ r.initHandlerWithRetry(name, handler, kinds)
r.mux.Lock()
defer r.mux.Unlock()
for _, ki := range kinds {
@@ -947,6 +975,34 @@ func (r *SchemaRegistry) RegisterHandler(name string, kind schema.Kind, handler
}
}
+func (r *SchemaRegistry) initHandlerWithRetry(name string, handler schema.EventHandler, kinds []schema.Kind) {
+ deadline := time.Now().Add(time.Minute)
+ var lastErr interface{}
+ for {
+ if tryCallOnInit(handler, kinds, &lastErr) {
+ return
+ }
+ if time.Now().After(deadline) {
+ r.l.Panic().Str("name", name).Interface("error", lastErr).
+ Msg("handler OnInit failed after 1m, giving up")
+ }
+ r.l.Warn().Str("name", name).Interface("error", lastErr).
+ Msg("handler OnInit panicked due to transient error, retrying in 1s")
+ time.Sleep(time.Second)
+ }
+}
+
+func tryCallOnInit(handler schema.EventHandler, kinds []schema.Kind, lastErr *interface{}) (ok bool) {
+ defer func() {
+ if rec := recover(); rec != nil {
+ *lastErr = rec
+ ok = false
+ }
+ }()
+ handler.OnInit(kinds)
+ return true
+}
+
// Start starts a single goroutine that periodically syncs schemas.
func (r *SchemaRegistry) Start(ctx context.Context) error {
if r.caCertReloader != nil {
@@ -1017,8 +1073,7 @@ func (r *SchemaRegistry) launchWatch(nodeName string, client *schemaClient) {
ctx, cancel := context.WithCancel(r.closer.Ctx())
session := &watchSession{
cancelFn: cancel,
- syncReqCh: make(chan *syncRequest, 1),
- syncResCh: make(chan []*digestEntry, 1),
+ syncReqCh: make(chan *syncMessage, 1),
}
r.watchSessions[nodeName] = session
r.l.Debug().Str("node", nodeName).Msg("launchWatch: starting new watch session")
@@ -1045,7 +1100,8 @@ func (r *SchemaRegistry) stopWatch(nodeName string) {
const (
watchInitialBackoff = time.Second
- watchMaxBackoff = 30 * time.Second
+ // defaultWatchMaxBackoff is the default maximum backoff for watch stream reconnection.
+ defaultWatchMaxBackoff = 30 * time.Second
)
func (r *SchemaRegistry) watchLoop(ctx context.Context, nodeName string,
@@ -1073,8 +1129,8 @@ func (r *SchemaRegistry) watchLoop(ctx context.Context, nodeName string,
case <-time.After(backoff):
}
backoff *= 2
- if backoff > watchMaxBackoff {
- backoff = watchMaxBackoff
+ if backoff > r.watchMaxBackoff {
+ backoff = r.watchMaxBackoff
}
}
}
@@ -1113,6 +1169,7 @@ func (r *SchemaRegistry) processWatchSession(ctx context.Context,
var inSync bool
var metadataOnly bool
var digests []*digestEntry
+ var currentResponseCh chan []*digestEntry
for {
select {
@@ -1132,18 +1189,20 @@ func (r *SchemaRegistry) processWatchSession(ctx context.Context,
inSync = true
metadataOnly = len(req.tagProjection) > 0
digests = nil
+ currentResponseCh = req.responseCh
case resp, ok := <-recvCh:
if !ok {
return <-recvErrCh
}
if resp.EventType == schemav1.SchemaEventType_SCHEMA_EVENT_TYPE_REPLAY_DONE {
r.l.Debug().Bool("inSync", inSync).Msg("processWatchSession: received REPLAY_DONE")
- if inSync {
+ if inSync && currentResponseCh != nil {
r.l.Debug().Int("digestCount", len(digests)).Msg("processWatchSession: sync replay completed, sending digests")
- session.syncResCh <- digests
+ currentResponseCh <- digests
digests = nil
inSync = false
metadataOnly = false
+ currentResponseCh = nil
}
continue
}
@@ -1219,7 +1278,9 @@ func (r *SchemaRegistry) performIncrementalSync(ctx context.Context) {
criteria: buildRevisionCriteria(maxRev),
}
r.l.Debug().Int64("maxRevision", maxRev).Msg("incremental sync: requesting changes")
- r.sendSyncRequest(ctx, sessions, req)
+ if _, err := r.sendSyncRequest(ctx, sessions, req); err != nil {
+ r.l.Warn().Err(err).Msg("incremental sync: sync request partially failed")
+ }
}
func (r *SchemaRegistry) getActiveSessions() map[string]*watchSession {
@@ -1234,29 +1295,43 @@ func (r *SchemaRegistry) getActiveSessions() map[string]*watchSession {
func (r *SchemaRegistry) sendSyncRequest(ctx context.Context,
sessions map[string]*watchSession, req *syncRequest,
-) map[string][]*digestEntry {
+) (map[string][]*digestEntry, error) {
r.l.Debug().Int("sessionCount", len(sessions)).Msg("sendSyncRequest: dispatching to watch sessions")
- sent := make(map[string]*watchSession, len(sessions))
+ sentChs := make(map[string]chan []*digestEntry, len(sessions))
+ var skipped int
for name, s := range sessions {
+ msg := &syncMessage{
+ syncRequest: *req,
+ responseCh: make(chan []*digestEntry, 1),
+ }
select {
- case s.syncReqCh <- req:
- sent[name] = s
+ case s.syncReqCh <- msg:
+ sentChs[name] = msg.responseCh
default:
+ skipped++
r.l.Warn().Str("node", name).Msg("sendSyncRequest: channel full, skipping session")
}
}
- allDigests := make(map[string][]*digestEntry, len(sent))
- for name, s := range sent {
+ var timedOut int
+ allDigests := make(map[string][]*digestEntry, len(sentChs))
+ for name, ch := range sentChs {
select {
- case digests := <-s.syncResCh:
+ case digests := <-ch:
allDigests[name] = digests
- case <-time.After(30 * time.Second):
+ r.l.Debug().Str("node", name).Int("digestCount", len(digests)).Msg("sendSyncRequest: received digests")
+ case <-time.After(r.syncTimeout):
+ timedOut++
r.l.Warn().Str("node", name).Msg("sync timeout")
case <-ctx.Done():
- return nil
+ return nil, ctx.Err()
}
}
- return allDigests
+ r.l.Debug().Int("totalSessions", len(sessions)).Int("sent", len(sentChs)).Int("responded", len(allDigests)).Msg("sendSyncRequest: completed")
+ if skipped > 0 || timedOut > 0 {
+ return allDigests, fmt.Errorf("sync partially failed: %d skipped, %d timed out out of %d sessions",
+ skipped, timedOut, len(sessions))
+ }
+ return allDigests, nil
}
func (r *SchemaRegistry) performFullSync(ctx context.Context) {
@@ -1268,14 +1343,21 @@ func (r *SchemaRegistry) performFullSync(ctx context.Context) {
req := &syncRequest{
tagProjection: basicTagKeys,
}
- allDigests := r.sendSyncRequest(ctx, sessions, req)
+ allDigests, err := r.sendSyncRequest(ctx, sessions, req)
if allDigests == nil {
return
}
+ if err != nil {
+ // Partial failure: some sessions timed out or were skipped.
+ // Use whatever digests were received for the add/update path,
+ // but skip the deletion reconciliation to avoid removing entries
+ // that may exist on unreachable nodes.
+ r.l.Warn().Err(err).Msg("performFullSync: sync request partially failed, skipping deletion reconciliation")
+ }
merged := r.mergeDigests(allDigests)
cachedEntries := r.cache.GetAllEntries()
- r.l.Debug().Int("mergedCount", len(merged)).Int("cachedCount", len(cachedEntries)).
+ r.l.Debug().Int("respondedNodes", len(allDigests)).Int("mergedCount", len(merged)).Int("cachedCount", len(cachedEntries)).
Msg("performFullSync: comparing server digests with local cache")
for propID, d := range merged {
@@ -1297,30 +1379,35 @@ func (r *SchemaRegistry) performFullSync(ctx context.Context) {
r.l.Warn().Str("kind", d.kind).Msg("full sync: unknown kind")
continue
}
- r.l.Debug().Str("propID", propID).Stringer("kind", kind).
- Int64("serverRevision", d.revision).Msg("performFullSync: fetching updated schema from server")
- prop, err := r.getSchema(ctx, kind, d.group, d.name)
- if err != nil {
- r.l.Warn().Err(err).Str("propID", propID).Msg("full sync: failed to get schema")
+ r.l.Info().Str("propID", propID).Stringer("kind", kind).
+ Int64("serverRevision", d.revision).
+ Bool("localMissing", localEntry == nil).
+ Msg("performFullSync: fetching updated schema from server")
+ prop, getErr := r.getSchema(ctx, kind, d.group, d.name)
+ if getErr != nil {
+ r.l.Warn().Err(getErr).Str("propID", propID).Msg("full sync: failed to get schema")
continue
}
if prop == nil {
continue
}
- md, err := ToSchema(kind, prop)
- if err != nil {
- r.l.Warn().Err(err).Str("propID", propID).Msg("full sync: failed to convert property")
+ md, convertErr := ToSchema(kind, prop)
+ if convertErr != nil {
+ r.l.Warn().Err(convertErr).Str("propID", propID).Msg("full sync: failed to convert property")
continue
}
r.processInitialResourceFromProperty(kind, prop, md.Spec.(proto.Message))
}
}
- for propID, entry := range cachedEntries {
- if _, exists := merged[propID]; !exists {
- r.l.Debug().Str("propID", propID).Stringer("kind", entry.kind).
- Msg("performFullSync: cached entry not found on server, deleting")
- r.handleDeletion(entry.kind, propID, entry, entry.latestUpdateAt)
+ if err == nil {
+ for propID, entry := range cachedEntries {
+ if _, exists := merged[propID]; !exists {
+ r.l.Warn().Str("propID", propID).Stringer("kind", entry.kind).
+ Str("group", entry.group).Str("name", entry.name).
+ Msg("performFullSync: cached entry not found on server, deleting")
+ r.handleDeletion(entry.kind, propID, entry, entry.latestUpdateAt)
+ }
}
}
}
diff --git a/banyand/metadata/schema/property/client_test.go b/banyand/metadata/schema/property/client_test.go
index 4986e9192..cbb997649 100644
--- a/banyand/metadata/schema/property/client_test.go
+++ b/banyand/metadata/schema/property/client_test.go
@@ -1337,11 +1337,7 @@ func (s *recordingSchemaServer) DeleteSchema(_ context.Context, req *schemav1.De
return &schemav1.DeleteSchemaResponse{Found: false}, nil
}
-func startRecordingServer(t *testing.T) (*recordingSchemaServer, string) {
- t.Helper()
- mock := newRecordingSchemaServer()
- lis, lisErr := net.Listen("tcp", "127.0.0.1:0")
- require.NoError(t, lisErr)
+func serveRecordingServer(mock *recordingSchemaServer, lis net.Listener) *grpc.Server {
srv := grpc.NewServer()
schemav1.RegisterSchemaManagementServiceServer(srv, mock)
schemav1.RegisterSchemaUpdateServiceServer(srv, mock)
@@ -1349,6 +1345,15 @@ func startRecordingServer(t *testing.T) (*recordingSchemaServer, string) {
healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
grpc_health_v1.RegisterHealthServer(srv, healthSrv)
go func() { _ = srv.Serve(lis) }()
+ return srv
+}
+
+func startRecordingServer(t *testing.T) (*recordingSchemaServer, string) {
+ t.Helper()
+ mock := newRecordingSchemaServer()
+ lis, lisErr := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, lisErr)
+ srv := serveRecordingServer(mock, lis)
t.Cleanup(func() { srv.GracefulStop() })
return mock, lis.Addr().String()
}
@@ -1567,6 +1572,109 @@ func deleteSchemaProperty(t *testing.T, mgmt schemav1.SchemaManagementServiceCli
require.NoError(t, err)
}
+func TestFullSyncServerDown(t *testing.T) {
+ t.Run("groups_preserved_when_server_stops", func(t *testing.T) {
+ // Verifies that when the schema server stops (simulating hot node restart),
+ // performFullSync does NOT delete locally cached groups because
+ // sendSyncRequest returns an error on timeout.
+ _, addr, stopFn := startTestSchemaServerFull(t)
+ mgmt := rawMgmtClient(t, addr)
+
+ insertSchemaProperty(t, mgmt, buildMockGroupProperty(t))
+ insertSchemaProperty(t, mgmt, buildMockStreamProperty(t, "sv-stream", time.Now()))
+
+ reg := newTestRegistryWithConfig(t, &property.ClientConfig{
+ SyncInterval: 200 * time.Millisecond,
+ SyncTimeout: 2 * time.Second,
+ FullReconcileEvery: 1,
+ }, addr)
+ handler := &testEventHandler{}
+ reg.RegisterHandler("test-handler", schema.KindGroup|schema.KindStream, handler)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ require.NoError(t, reg.Start(ctx))
+
+ waitForAddOrUpdateEvent(t, handler, schema.KindGroup, "test-group")
+ waitForAddOrUpdateEvent(t, handler, schema.KindStream, "sv-stream")
+ initialDeleteCount := len(handler.getDeleteEvents())
+
+ // Stop the server. The watch session stays "active" but broken.
+ stopFn()
+
+ // Wait for the sync timeout (2s) to fire plus some buffer.
+ time.Sleep(5 * time.Second)
+
+ // Verify NO group delete events were fired.
+ deleteEvents := handler.getDeleteEvents()
+ groupDeleteCount := 0
+ for _, evt := range deleteEvents[initialDeleteCount:] {
+ if evt.Kind == schema.KindGroup && evt.Name == "test-group" {
+ groupDeleteCount++
+ }
+ }
+ assert.Equal(t, 0, groupDeleteCount, "Groups should NOT be deleted when server is unreachable")
+ })
+
+ t.Run("groups_recovered_after_server_restart", func(t *testing.T) {
+ // Verifies that after server stops and restarts on the same address,
+ // the watch session reconnects and groups remain consistent.
+ lis, lisErr := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, lisErr)
+ addr := lis.Addr().String()
+
+ mock := newRecordingSchemaServer()
+ mock.addSchema(buildMockGroupProperty(t))
+ mock.addSchema(buildMockStreamProperty(t, "sv-stream", time.Now()))
+ srv := serveRecordingServer(mock, lis)
+
+ reg := newTestRegistryWithConfig(t, &property.ClientConfig{
+ SyncInterval: 200 * time.Millisecond,
+ SyncTimeout: 2 * time.Second,
+ WatchMaxBackoff: 2 * time.Second,
+ FullReconcileEvery: 1,
+ }, addr)
+ handler := &testEventHandler{}
+ reg.RegisterHandler("test-handler", schema.KindGroup|schema.KindStream, handler)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ require.NoError(t, reg.Start(ctx))
+
+ waitForAddOrUpdateEvent(t, handler, schema.KindGroup, "test-group")
+ waitForAddOrUpdateEvent(t, handler, schema.KindStream, "sv-stream")
+ initialDeleteCount := len(handler.getDeleteEvents())
+
+ // Stop the server (hard stop to avoid blocking on bidi streams).
+ srv.Stop()
+
+ // Wait for sync timeout (2s) to fire. Groups should NOT be deleted.
+ time.Sleep(5 * time.Second)
+ for _, evt := range handler.getDeleteEvents()[initialDeleteCount:] {
+ require.NotEqual(t, schema.KindGroup, evt.Kind,
+ "Group should not be deleted during server outage")
+ }
+
+ // Restart server on same address with same data.
+ lis2, lisErr2 := net.Listen("tcp", addr)
+ require.NoError(t, lisErr2)
+ mock2 := newRecordingSchemaServer()
+ mock2.addSchema(buildMockGroupProperty(t))
+ mock2.addSchema(buildMockStreamProperty(t, "sv-stream", time.Now()))
+ srv2 := serveRecordingServer(mock2, lis2)
+ t.Cleanup(func() { srv2.Stop() })
+
+ // Wait for watch reconnection (backoff up to 2s) and a successful full sync.
+ time.Sleep(5 * time.Second)
+
+ // Verify no group was deleted during the entire process.
+ for _, evt := range handler.getDeleteEvents()[initialDeleteCount:] {
+ assert.NotEqual(t, schema.KindGroup, evt.Kind,
+ "Group should not be deleted after server restart")
+ }
+ })
+}
+
func TestWatchPush(t *testing.T) {
t.Run("watch_push_insert", func(t *testing.T) {
srv, addr, _ := startTestSchemaServerFull(t)
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 05043ac0b..e7b975176 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -388,7 +388,7 @@ func (p *pub) PreRun(context.Context) error {
p.log = logger.GetLogger("server-queue-pub-" + p.prefix)
// Initialize connection manager with the pub as the handler
- p.connMgr = grpchelper.NewConnManager(grpchelper.ConnManagerConfig[*client]{
+ p.connMgr = grpchelper.NewConnManager(grpchelper.ConnManagerConfig[*client]{ //nolint:contextcheck // health check runs in background goroutine
Handler: p,
Logger: p.log,
RetryPolicy: p.retryPolicy,
diff --git a/docs/operation/configuration.md b/docs/operation/configuration.md
index 7c7b38a50..0782a000b 100644
--- a/docs/operation/configuration.md
+++ b/docs/operation/configuration.md
@@ -133,6 +133,17 @@ The following flags are used to configure the memory protector:
- `--dst-slow-query duration`: distributed slow query threshold, 0 means no slow query log. This is only used for the liaison server (default: 0).
- `--slow-query duration`: slow query threshold, 0 means no slow query log. This is only used for the data and standalone server (default: 0).
+### Schema Registry
+
+The following flags configure the property-based schema registry used for metadata synchronization between nodes:
+
+- `--schema-registry-mode string`: Schema registry mode: `etcd` for etcd-based storage, `property` for property-based storage (default: "property").
+- `--schema-property-client-sync-interval duration`: Polling interval for property-based schema sync (default: 30s).
+- `--schema-property-client-health-check-interval duration`: Interval for periodic connection health checks to schema servers. Set to 0 to use the default interval (10s), or to a negative value to disable.
+- `--schema-property-client-max-recv-msg-size bytes`: Max gRPC receive message size for property schema client.
+- `--schema-property-client-tls`: Enable TLS for property schema client connections.
+- `--schema-property-client-ca-cert string`: CA certificate file to verify the property schema server.
+
### Other
- `-n, --name string`: Name of this service.
diff --git a/pkg/grpchelper/connmanager.go b/pkg/grpchelper/connmanager.go
index 4e9793679..eb1406bac 100644
--- a/pkg/grpchelper/connmanager.go
+++ b/pkg/grpchelper/connmanager.go
@@ -60,27 +60,29 @@ type ConnectionHandler[C Client] interface {
// ConnManagerConfig holds configuration for ConnManager.
type ConnManagerConfig[C Client] struct {
- Handler ConnectionHandler[C]
- Logger *logger.Logger
- RetryPolicy string
- ExtraDialOpts []grpc.DialOption
- MaxRecvMsgSize int
- HealthCheckTimeout time.Duration
+ Handler ConnectionHandler[C]
+ Logger *logger.Logger
+ RetryPolicy string
+ ExtraDialOpts []grpc.DialOption
+ HealthCheckTimeout time.Duration
+ HealthCheckInterval time.Duration
+ MaxRecvMsgSize int
}
// ConnManager manages gRPC connections with health checking, circuit breaking, and eviction.
type ConnManager[C Client] struct {
- handler ConnectionHandler[C]
- log *logger.Logger
- registered map[string]*databasev1.Node
- active map[string]*managedNode[C]
- evictable map[string]evictNode
- cbStates map[string]*circuitState
- closer *run.Closer
- dialOpts []grpc.DialOption
- healthCheckTimeout time.Duration
- mu sync.RWMutex
- cbMu sync.RWMutex
+ handler ConnectionHandler[C]
+ log *logger.Logger
+ registered map[string]*databasev1.Node
+ active map[string]*managedNode[C]
+ evictable map[string]evictNode
+ cbStates map[string]*circuitState
+ closer *run.Closer
+ dialOpts []grpc.DialOption
+ healthCheckTimeout time.Duration
+ healthCheckInterval time.Duration
+ mu sync.RWMutex
+ cbMu sync.RWMutex
}
type managedNode[C Client] struct {
@@ -109,15 +111,19 @@ func NewConnManager[C Client](cfg ConnManagerConfig[C]) *ConnManager[C] {
healthCheckTimeout = 2 * time.Second
}
m := &ConnManager[C]{
- handler: cfg.Handler,
- log: cfg.Logger,
- registered: make(map[string]*databasev1.Node),
- active: make(map[string]*managedNode[C]),
- evictable: make(map[string]evictNode),
- cbStates: make(map[string]*circuitState),
- closer: run.NewCloser(1),
- dialOpts: dialOpts,
- healthCheckTimeout: healthCheckTimeout,
+ handler: cfg.Handler,
+ log: cfg.Logger,
+ registered: make(map[string]*databasev1.Node),
+ active: make(map[string]*managedNode[C]),
+ evictable: make(map[string]evictNode),
+ cbStates: make(map[string]*circuitState),
+ closer: run.NewCloser(1),
+ dialOpts: dialOpts,
+ healthCheckTimeout: healthCheckTimeout,
+ healthCheckInterval: cfg.HealthCheckInterval,
+ }
+ if m.healthCheckInterval > 0 && m.closer.AddRunning() {
+ go m.periodicHealthCheck()
}
return m
}
@@ -499,6 +505,22 @@ func (m *ConnManager[C]) checkHealthAndReconnect(conn *grpc.ClientConn, node *da
return false
}
+func (m *ConnManager[C]) periodicHealthCheck() {
+ defer m.closer.Done()
+ ticker := time.NewTicker(m.healthCheckInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ for _, name := range m.ActiveNames() {
+ m.FailoverNode(name)
+ }
+ case <-m.closer.CloseNotify():
+ return
+ }
+ }
+}
+
func (m *ConnManager[C]) healthCheck(node string, conn *grpc.ClientConn) bool {
var resp *grpc_health_v1.HealthCheckResponse
if requestErr := Request(context.Background(), m.healthCheckTimeout, func(rpcCtx context.Context) (err error) {
diff --git a/pkg/grpchelper/connmanager_test.go b/pkg/grpchelper/connmanager_test.go
index f85303fbf..7c99e6e6a 100644
--- a/pkg/grpchelper/connmanager_test.go
+++ b/pkg/grpchelper/connmanager_test.go
@@ -18,12 +18,23 @@
package grpchelper
import (
+ "net"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "google.golang.org/grpc"
"google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/grpc/health"
+ "google.golang.org/grpc/health/grpc_health_v1"
"google.golang.org/grpc/status"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ testflags "github.com/apache/skywalking-banyandb/pkg/test/flags"
)
func addTestClient(m *ConnManager[*mockClient], node string) {
@@ -113,3 +124,200 @@ func TestExecute_ClientNotFound(t *testing.T) {
assert.ErrorIs(t, execErr, ErrClientNotFound)
assert.Contains(t, execErr.Error(), node)
}
+
+func startHealthServerOnAddr(t *testing.T, addr string) (string, func()) {
+ t.Helper()
+ lis, lisErr := net.Listen("tcp", addr)
+ require.NoError(t, lisErr)
+ srv := grpc.NewServer()
+ healthSrv := health.NewServer()
+ healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_SERVING)
+ grpc_health_v1.RegisterHealthServer(srv, healthSrv)
+ go func() { _ = srv.Serve(lis) }()
+ return lis.Addr().String(), func() { srv.Stop() }
+}
+
+func startHealthServer(t *testing.T) (string, func()) {
+ t.Helper()
+ return startHealthServerOnAddr(t, "127.0.0.1:0")
+}
+
+func buildNode(name, addr string) *databasev1.Node {
+ return &databasev1.Node{
+ Metadata: &commonv1.Metadata{Name: name},
+ GrpcAddress: addr,
+ }
+}
+
+// mockHandlerWithInsecure extends mockHandler with insecure dial options for real gRPC connections.
+type mockHandlerWithInsecure struct {
+ mockHandler
+}
+
+func (h *mockHandlerWithInsecure) GetDialOptions() ([]grpc.DialOption, error) {
+ return []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
+}
+
+func newHealthCheckConnManager(t *testing.T, interval time.Duration) (*ConnManager[*mockClient], *mockHandlerWithInsecure) {
+ t.Helper()
+ handler := &mockHandlerWithInsecure{}
+ mgr := NewConnManager(ConnManagerConfig[*mockClient]{
+ Handler: handler,
+ Logger: logger.GetLogger("test-hc"),
+ HealthCheckInterval: interval,
+ HealthCheckTimeout: 1 * time.Second,
+ })
+ t.Cleanup(func() { mgr.GracefulStop() })
+ return mgr, handler
+}
+
+func TestPeriodicHealthCheck(t *testing.T) {
+ t.Run("unhealthy_node_moves_to_evictable", func(t *testing.T) {
+ // Start a real gRPC health server, register a node, then stop the server.
+ // The periodic health check should detect the node is unhealthy
+ // and call OnInactive.
+ addr, stopServer := startHealthServer(t)
+
+ mgr, handler := newHealthCheckConnManager(t, 200*time.Millisecond)
+
+ // Register the node — health check passes, node becomes active.
+ mgr.OnAddOrUpdate(buildNode("hc-node", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+ require.True(t, handler.isActive("hc-node"))
+
+ // Stop the server to make the node unhealthy.
+ stopServer()
+
+ // Wait for periodic health check to detect the failure.
+ require.Eventually(t, func() bool {
+ return mgr.ActiveCount() == 0
+ }, testflags.EventuallyTimeout, 100*time.Millisecond,
+ "node should be moved from active to evictable after health check failure")
+
+ // Verify OnInactive was called.
+ assert.True(t, handler.isInactive("hc-node"))
+ assert.Equal(t, 1, mgr.EvictableCount())
+ })
+
+ t.Run("healthy_node_stays_active", func(t *testing.T) {
+ // Start a real gRPC health server that remains running.
+ // The periodic health check should keep the node active.
+ addr, stopServer := startHealthServer(t)
+ defer stopServer()
+
+ mgr, handler := newHealthCheckConnManager(t, 200*time.Millisecond)
+
+ mgr.OnAddOrUpdate(buildNode("hc-stable", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+
+ // Wait for several health check cycles.
+ time.Sleep(1 * time.Second)
+
+ // Node should still be active.
+ assert.Equal(t, 1, mgr.ActiveCount())
+ assert.Equal(t, 0, mgr.EvictableCount())
+ assert.True(t, handler.isActive("hc-stable"))
+ })
+
+ t.Run("node_recovers_after_server_restart", func(t *testing.T) {
+ // Start server, register node, stop server (node becomes evictable),
+ // restart server on same address (node should recover to active).
+ addr, stopServer := startHealthServer(t)
+
+ mgr, handler := newHealthCheckConnManager(t, 200*time.Millisecond)
+
+ mgr.OnAddOrUpdate(buildNode("hc-restart", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+
+ // Stop server — node should become evictable.
+ stopServer()
+ require.Eventually(t, func() bool {
+ return mgr.ActiveCount() == 0
+ }, testflags.EventuallyTimeout, 100*time.Millisecond,
+ "node should become evictable after server stop")
+ assert.True(t, handler.isInactive("hc-restart"))
+
+ // Restart server on the same address.
+ _, stopRestarted := startHealthServerOnAddr(t, addr)
+ defer stopRestarted()
+
+ // The eviction retry goroutine should reconnect the node.
+ require.Eventually(t, func() bool {
+ return mgr.ActiveCount() == 1
+ }, testflags.EventuallyTimeout, 100*time.Millisecond,
+ "node should recover to active after server restart")
+ assert.True(t, handler.isActive("hc-restart"))
+ })
+
+ t.Run("disabled_when_interval_is_zero", func(t *testing.T) {
+ // With HealthCheckInterval=0, no periodic health check goroutine should run.
+ // Start a real server but the health check should never fire.
+ addr, stopServer := startHealthServer(t)
+
+ mgr, _ := newHealthCheckConnManager(t, 0)
+ mgr.OnAddOrUpdate(buildNode("no-hc-node", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+
+ // Stop the server. Without periodic health check, the node stays active.
+ stopServer()
+ time.Sleep(500 * time.Millisecond)
+
+ // Node should still be active — no health check removed it.
+ assert.Equal(t, 1, mgr.ActiveCount())
+ })
+
+ t.Run("graceful_stop_terminates_health_check", func(t *testing.T) {
+ // Verify that GracefulStop cleanly shuts down the periodic health check goroutine.
+ addr, stopServer := startHealthServer(t)
+ defer stopServer()
+
+ handler := &mockHandlerWithInsecure{}
+ mgr := NewConnManager(ConnManagerConfig[*mockClient]{
+ Handler: handler,
+ Logger: logger.GetLogger("test-hc-stop"),
+ HealthCheckInterval: 100 * time.Millisecond,
+ HealthCheckTimeout: 1 * time.Second,
+ })
+
+ mgr.OnAddOrUpdate(buildNode("hc-stop", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+
+ // GracefulStop should return without hanging (health check goroutine exits).
+ done := make(chan struct{})
+ go func() {
+ mgr.GracefulStop()
+ close(done)
+ }()
+ select {
+ case <-done:
+ // Success — GracefulStop returned.
+ case <-time.After(5 * time.Second):
+ t.Fatal("GracefulStop did not return in time — periodic health check may be leaked")
+ }
+ })
+}
+
+func TestPeriodicHealthCheck_ConnectsViaGRPC(t *testing.T) {
+ // Integration test: verify that OnAddOrUpdate creates a real gRPC connection
+ // and the periodic health check keeps the node active end-to-end.
+ addr, stopServer := startHealthServer(t)
+
+ mgr, handler := newHealthCheckConnManager(t, 200*time.Millisecond)
+
+ // Register the node via ConnManager — should establish a real gRPC connection.
+ mgr.OnAddOrUpdate(buildNode("grpc-node", addr))
+ require.Equal(t, 1, mgr.ActiveCount())
+ require.True(t, handler.isActive("grpc-node"))
+
+ // Wait for several health check cycles — node should stay active.
+ time.Sleep(1 * time.Second)
+ assert.Equal(t, 1, mgr.ActiveCount())
+
+ // Stop server — periodic health check should detect failure and evict.
+ stopServer()
+ require.Eventually(t, func() bool {
+ return mgr.ActiveCount() == 0
+ }, testflags.EventuallyTimeout, 100*time.Millisecond,
+ "node should be evicted after server stops")
+ assert.True(t, handler.isInactive("grpc-node"))
+}
diff --git a/pkg/grpchelper/helpers_test.go b/pkg/grpchelper/helpers_test.go
index 1c6c9815c..5f62a8fc1 100644
--- a/pkg/grpchelper/helpers_test.go
+++ b/pkg/grpchelper/helpers_test.go
@@ -18,6 +18,8 @@
package grpchelper
import (
+ "sync"
+
"google.golang.org/grpc"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
@@ -37,6 +39,7 @@ func (m *mockClient) Close() error {
type mockHandler struct {
activeNodes map[string]*mockClient
inactiveNodes map[string]*mockClient
+ mu sync.Mutex
}
func (h *mockHandler) AddressOf(node *databasev1.Node) string {
@@ -52,6 +55,8 @@ func (h *mockHandler) NewClient(_ *grpc.ClientConn, _ *databasev1.Node) (*mockCl
}
func (h *mockHandler) OnActive(name string, client *mockClient) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
if h.activeNodes == nil {
h.activeNodes = make(map[string]*mockClient)
}
@@ -59,9 +64,25 @@ func (h *mockHandler) OnActive(name string, client *mockClient) {
}
func (h *mockHandler) OnInactive(name string, client *mockClient) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
if h.inactiveNodes == nil {
h.inactiveNodes = make(map[string]*mockClient)
}
h.inactiveNodes[name] = client
delete(h.activeNodes, name)
}
+
+func (h *mockHandler) isActive(name string) bool {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ _, ok := h.activeNodes[name]
+ return ok
+}
+
+func (h *mockHandler) isInactive(name string) bool {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ _, ok := h.inactiveNodes[name]
+ return ok
+}