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/04/30 06:05:09 UTC
(skywalking-banyandb) branch phase-2-step-2-1-node-schema-status updated (bc9944bee -> c1e1616b3)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a change to branch phase-2-step-2-1-node-schema-status
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
from bc9944bee feat(schema): add NodeSchemaStatusService for cluster-wide barrier (Phase 2 Step 2.1)
new fb457a744 fix(schema): resolve lint warnings in NodeSchemaStatusService
new c1e1616b3 fix(schema): correct NodeSchemaStatusService nil-cache and add request cap
The 2 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.
Summary of changes:
banyand/metadata/schema/property/cache.go | 19 -------
banyand/metadata/schema/property/converter.go | 2 +-
banyand/metadata/schema/property/node_status.go | 58 +++++++++++--------
.../metadata/schema/property/node_status_test.go | 65 +++++++++++++++++-----
4 files changed, 86 insertions(+), 58 deletions(-)
(skywalking-banyandb) 02/02: fix(schema): correct NodeSchemaStatusService nil-cache and add request cap
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch phase-2-step-2-1-node-schema-status
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit c1e1616b3dd42f4cb0b3e2f6833fa7fe96719dac
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Thu Apr 30 03:30:18 2026 +0000
fix(schema): correct NodeSchemaStatusService nil-cache and add request cap
Three review fixes against Phase 2 Step 2.1's NodeSchemaStatusService:
- GetAbsentKeys now reports every requested key as still-present when
the local cache is nil (was: reported as absent). The previous
semantic was the inverse of Phase 1's collectPresentKeys nil-cache
contract in banyand/liaison/grpc/barrier.go and would let a remote
AwaitSchemaDeleted falsely conclude deletion-applied during cluster
bootstrap, before any data node had observed a delete event.
Refactored the partition to call schemaCache.GetKeyRevisions so the
pass runs under one read-lock instead of one per key.
- Added nodeStatusMaxKeys = 10000 and rejected over-cap requests with
codes.InvalidArgument on both GetKeyRevisions and GetAbsentKeys,
matching the barrierMaxKeys cap that SchemaBarrierService already
enforces. Without this the RPC would allocate an unbounded response
slice for a misbehaving peer.
- Removed the unused schemaCache.IsAbsent helper that landed with the
Phase 2.1 commit but had no caller (GetAbsentKeys was looping
GetKeyModRevision instead). The new partition path consolidates on
GetKeyRevisions, so IsAbsent's surface area is dead.
Tests: flipped TestGetAbsentKeys_NilCache_AllAbsent to
TestGetAbsentKeys_NilCache_AllStillPresent so the corrected contract is
asserted, and added TestGet{KeyRevisions,AbsentKeys}_OverLimit_RejectsWithInvalidArgument.
via [HAPI](https://hapi.run)
---
banyand/metadata/schema/property/cache.go | 19 --------
banyand/metadata/schema/property/node_status.go | 56 +++++++++++++---------
.../metadata/schema/property/node_status_test.go | 41 ++++++++++++++--
3 files changed, 71 insertions(+), 45 deletions(-)
diff --git a/banyand/metadata/schema/property/cache.go b/banyand/metadata/schema/property/cache.go
index d575a3a73..4ea64b411 100644
--- a/banyand/metadata/schema/property/cache.go
+++ b/banyand/metadata/schema/property/cache.go
@@ -228,25 +228,6 @@ func (c *schemaCache) GetKeyRevisions(propIDs []string) []KeyRevision {
return out
}
-// IsAbsent partitions propIDs into "absent from the cache" and "still present"
-// in a single read-lock pass. The split mirrors GetKeyModRevision's gate:
-// entries whose mod_revision exceeds notifiedModRevision count as absent
-// because downstream handlers have not yet observed them. The returned slices
-// preserve input order within each subset.
-func (c *schemaCache) IsAbsent(propIDs []string) (absent, present []string) {
- c.mu.RLock()
- defer c.mu.RUnlock()
- for _, pid := range propIDs {
- entry, ok := c.entries[pid]
- if !ok || entry.modRevision > c.notifiedModRevision {
- absent = append(absent, pid)
- continue
- }
- present = append(present, pid)
- }
- return absent, present
-}
-
// AdvanceNotified pushes the downstream-handler watermark forward monotonically.
// Callers must invoke this only after notifyHandlers returns for the given revision,
// so downstream caches (groupRepo, entityRepo, pkg/schema.schemaRepo, …) are coherent
diff --git a/banyand/metadata/schema/property/node_status.go b/banyand/metadata/schema/property/node_status.go
index 8ec61e31d..6ae3ac2d1 100644
--- a/banyand/metadata/schema/property/node_status.go
+++ b/banyand/metadata/schema/property/node_status.go
@@ -20,10 +20,20 @@ package property
import (
"context"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
schemav1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/schema/v1"
)
+// nodeStatusMaxKeys caps the per-request key count to match SchemaBarrierService
+// (barrierMaxKeys in banyand/liaison/grpc/barrier.go). The proto leaves this
+// uncapped so chunking is the caller's responsibility, but the server still
+// enforces the same bound the SchemaBarrierService does so a misbehaving peer
+// cannot allocate unbounded memory.
+const nodeStatusMaxKeys = 10000
+
// NodeSchemaStatusServer implements clusterv1.NodeSchemaStatusServiceServer
// against a local SchemaRegistry cache. The same implementation runs on every
// cluster member that holds a schema cache: liaisons (Role_ROLE_LIAISON) and
@@ -89,6 +99,9 @@ func (s *NodeSchemaStatusServer) GetMaxRevision(_ context.Context, _ *clusterv1.
// An empty request is valid and produces an empty response.
func (s *NodeSchemaStatusServer) GetKeyRevisions(_ context.Context, req *clusterv1.GetKeyRevisionsRequest) (*clusterv1.GetKeyRevisionsResponse, error) {
keys := req.GetKeys()
+ if len(keys) > nodeStatusMaxKeys {
+ return nil, status.Errorf(codes.InvalidArgument, "too many keys: max=%d", nodeStatusMaxKeys)
+ }
resp := &clusterv1.GetKeyRevisionsResponse{
Revisions: make([]*clusterv1.KeyRevision, len(keys)),
}
@@ -101,9 +114,9 @@ func (s *NodeSchemaStatusServer) GetKeyRevisions(_ context.Context, req *cluster
}
propIDs := schemaKeysToPropIDs(keys)
statuses := c.GetKeyRevisions(propIDs)
- for i, status := range statuses {
- resp.Revisions[i].ModRevision = status.ModRevision
- resp.Revisions[i].Present = status.Present
+ for i, keyStatus := range statuses {
+ resp.Revisions[i].ModRevision = keyStatus.ModRevision
+ resp.Revisions[i].Present = keyStatus.Present
}
return resp, nil
}
@@ -114,36 +127,35 @@ func (s *NodeSchemaStatusServer) GetKeyRevisions(_ context.Context, req *cluster
// erroring so the caller can rely on the partition adding up to the input.
func (s *NodeSchemaStatusServer) GetAbsentKeys(_ context.Context, req *clusterv1.GetAbsentKeysRequest) (*clusterv1.GetAbsentKeysResponse, error) {
keys := req.GetKeys()
+ if len(keys) > nodeStatusMaxKeys {
+ return nil, status.Errorf(codes.InvalidArgument, "too many keys: max=%d", nodeStatusMaxKeys)
+ }
resp := &clusterv1.GetAbsentKeysResponse{}
if len(keys) == 0 {
return resp, nil
}
c := s.cacheProvider()
if c == nil {
- // Cache not yet initialized — every requested key is treated as absent
- // from the local perspective. The liaison's barrier will retry on the
- // next backoff iteration once the cache is online.
- resp.AbsentKeys = append([]*schemav1.SchemaKey(nil), keys...)
+ // Cache not yet initialized — the node has not observed any schema
+ // state, so it cannot claim deletion has been applied. Report every
+ // key as still present so the liaison's AwaitSchemaDeleted barrier
+ // keeps polling until the cache is online. Mirrors the Phase 1
+ // collectPresentKeys nil-cache contract in
+ // banyand/liaison/grpc/barrier.go.
+ resp.StillPresentKeys = append([]*schemav1.SchemaKey(nil), keys...)
return resp, nil
}
propIDs := schemaKeysToPropIDs(keys)
- for i, key := range keys {
- propID := propIDs[i]
- if propID == "" {
- // Unknown kind — treat as absent so the caller's "every key absent"
- // invariant holds without surfacing a parse error.
- resp.AbsentKeys = append(resp.AbsentKeys, key)
- continue
- }
- // GetKeyModRevision handles both the cache-map lookup and the
- // notifiedModRevision gate atomically under the cache's read lock,
- // matching the semantics that AwaitSchemaDeleted's collectPresentKeys
- // already relies on.
- if _, ok := c.GetKeyModRevision(propID); !ok {
- resp.AbsentKeys = append(resp.AbsentKeys, key)
+ // One read-lock pass over the cache; an empty propID (unknown kind) maps
+ // to Present=false in c.GetKeyRevisions because c.entries[""] never
+ // matches, which is the same "absent" partition the proto requires.
+ statuses := c.GetKeyRevisions(propIDs)
+ for i, keyStatus := range statuses {
+ if keyStatus.Present {
+ resp.StillPresentKeys = append(resp.StillPresentKeys, keys[i])
continue
}
- resp.StillPresentKeys = append(resp.StillPresentKeys, key)
+ resp.AbsentKeys = append(resp.AbsentKeys, keys[i])
}
return resp, nil
}
diff --git a/banyand/metadata/schema/property/node_status_test.go b/banyand/metadata/schema/property/node_status_test.go
index 7cca09533..1c43a24e0 100644
--- a/banyand/metadata/schema/property/node_status_test.go
+++ b/banyand/metadata/schema/property/node_status_test.go
@@ -24,6 +24,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
schemav1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/schema/v1"
@@ -215,7 +217,12 @@ func TestGetAbsentKeys_PostDelete(t *testing.T) {
assert.Contains(t, absentNames, "never")
}
-func TestGetAbsentKeys_NilCache_AllAbsent(t *testing.T) {
+func TestGetAbsentKeys_NilCache_AllStillPresent(t *testing.T) {
+ // A nil cache means the node has not observed any schema state. The node
+ // must therefore report every requested key as still-present so the
+ // liaison's AwaitSchemaDeleted barrier keeps polling rather than falsely
+ // concluding deletion has been applied. Mirrors the Phase 1 collectPresentKeys
+ // nil-cache contract in banyand/liaison/grpc/barrier.go.
srv := NewNodeSchemaStatusServer(func() *schemaCache { return nil })
req := &clusterv1.GetAbsentKeysRequest{
@@ -227,9 +234,35 @@ func TestGetAbsentKeys_NilCache_AllAbsent(t *testing.T) {
resp, err := srv.GetAbsentKeys(context.Background(), req)
require.NoError(t, err)
- assert.Len(t, resp.GetAbsentKeys(), 2,
- "nil cache reports every requested key as absent so the liaison fan-out retries")
- assert.Empty(t, resp.GetStillPresentKeys())
+ assert.Len(t, resp.GetStillPresentKeys(), 2,
+ "nil cache reports every requested key as still-present so the liaison barrier keeps polling")
+ assert.Empty(t, resp.GetAbsentKeys())
+}
+
+func TestGetKeyRevisions_OverLimit_RejectsWithInvalidArgument(t *testing.T) {
+ srv := nodeStatusFixture(t, map[string]*cacheEntry{})
+ keys := make([]*schemav1.SchemaKey, nodeStatusMaxKeys+1)
+ for i := range keys {
+ keys[i] = &schemav1.SchemaKey{Kind: "measure", Group: "g1", Name: strconv.Itoa(i)}
+ }
+
+ resp, err := srv.GetKeyRevisions(context.Background(), &clusterv1.GetKeyRevisionsRequest{Keys: keys})
+ require.Error(t, err)
+ assert.Nil(t, resp)
+ assert.Equal(t, codes.InvalidArgument, status.Code(err))
+}
+
+func TestGetAbsentKeys_OverLimit_RejectsWithInvalidArgument(t *testing.T) {
+ srv := nodeStatusFixture(t, map[string]*cacheEntry{})
+ keys := make([]*schemav1.SchemaKey, nodeStatusMaxKeys+1)
+ for i := range keys {
+ keys[i] = &schemav1.SchemaKey{Kind: "measure", Group: "g1", Name: strconv.Itoa(i)}
+ }
+
+ resp, err := srv.GetAbsentKeys(context.Background(), &clusterv1.GetAbsentKeysRequest{Keys: keys})
+ require.Error(t, err)
+ assert.Nil(t, resp)
+ assert.Equal(t, codes.InvalidArgument, status.Code(err))
}
func TestGetKeyRevisions_ChunkedAtLimit(t *testing.T) {
(skywalking-banyandb) 01/02: fix(schema): resolve lint warnings in NodeSchemaStatusService
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch phase-2-step-2-1-node-schema-status
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit fb457a74495dd26ec7b60319148fe1541418609d
Author: Hongtao Gao <ha...@gmail.com>
AuthorDate: Wed Apr 29 22:34:50 2026 +0000
fix(schema): resolve lint warnings in NodeSchemaStatusService
Fixes misspell "unrecognised"/"initialised" -> American English variants
and removes unused `group` parameter from the test helper that always
received "g1".
via [HAPI](https://hapi.run)
---
banyand/metadata/schema/property/converter.go | 2 +-
banyand/metadata/schema/property/node_status.go | 4 ++--
.../metadata/schema/property/node_status_test.go | 24 ++++++++++++----------
3 files changed, 16 insertions(+), 14 deletions(-)
diff --git a/banyand/metadata/schema/property/converter.go b/banyand/metadata/schema/property/converter.go
index 8164c4ca3..2155180bc 100644
--- a/banyand/metadata/schema/property/converter.go
+++ b/banyand/metadata/schema/property/converter.go
@@ -180,7 +180,7 @@ func kindFromProtoString(protoKind string) schema.Kind {
// BuildPropertyIDFromSchemaKey converts a SchemaKey from the cluster
// node-status proto into the property-ID string used internally by the
-// schema cache. Returns an empty string when the kind is unrecognised so
+// schema cache. Returns an empty string when the kind is unrecognized so
// the caller can short-circuit instead of looking up a malformed propID.
func BuildPropertyIDFromSchemaKey(key *schemav1.SchemaKey) string {
kind := kindFromProtoString(key.GetKind())
diff --git a/banyand/metadata/schema/property/node_status.go b/banyand/metadata/schema/property/node_status.go
index b48752cae..8ec61e31d 100644
--- a/banyand/metadata/schema/property/node_status.go
+++ b/banyand/metadata/schema/property/node_status.go
@@ -69,7 +69,7 @@ func NewNodeSchemaStatusServerForRegistry(reg *SchemaRegistry) *NodeSchemaStatus
}
// GetMaxRevision returns the highest mod_revision currently observed by the
-// node's local schema cache. When the cache is not yet initialised the
+// node's local schema cache. When the cache is not yet initialized the
// response carries 0, which the liaison's barrier loop interprets as
// "node not yet caught up" and continues polling.
func (s *NodeSchemaStatusServer) GetMaxRevision(_ context.Context, _ *clusterv1.GetMaxRevisionRequest) (*clusterv1.GetMaxRevisionResponse, error) {
@@ -120,7 +120,7 @@ func (s *NodeSchemaStatusServer) GetAbsentKeys(_ context.Context, req *clusterv1
}
c := s.cacheProvider()
if c == nil {
- // Cache not yet initialised — every requested key is treated as absent
+ // Cache not yet initialized — every requested key is treated as absent
// from the local perspective. The liaison's barrier will retry on the
// next backoff iteration once the cache is online.
resp.AbsentKeys = append([]*schemav1.SchemaKey(nil), keys...)
diff --git a/banyand/metadata/schema/property/node_status_test.go b/banyand/metadata/schema/property/node_status_test.go
index 3b0db7499..7cca09533 100644
--- a/banyand/metadata/schema/property/node_status_test.go
+++ b/banyand/metadata/schema/property/node_status_test.go
@@ -52,15 +52,17 @@ func nodeStatusFixture(t *testing.T, entries map[string]*cacheEntry) *NodeSchema
return NewNodeSchemaStatusServer(func() *schemaCache { return c })
}
-func makeEntry(kind schema.Kind, group, name string, rev int64) (string, *cacheEntry) {
+const testGroup = "g1"
+
+func makeEntry(kind schema.Kind, name string, rev int64) (string, *cacheEntry) {
propID := kind.String() + "_"
if kind == schema.KindGroup {
propID += name
} else {
- propID += group + "/" + name
+ propID += testGroup + "/" + name
}
return propID, &cacheEntry{
- group: group,
+ group: testGroup,
name: name,
kind: kind,
modRevision: rev,
@@ -69,7 +71,7 @@ func makeEntry(kind schema.Kind, group, name string, rev int64) (string, *cacheE
}
func TestGetMaxRevision_ReflectsLatestApply(t *testing.T) {
- id, entry := makeEntry(schema.KindMeasure, "g1", "m1", 42)
+ id, entry := makeEntry(schema.KindMeasure, "m1", 42)
srv := nodeStatusFixture(t, map[string]*cacheEntry{id: entry})
resp, err := srv.GetMaxRevision(context.Background(), &clusterv1.GetMaxRevisionRequest{})
@@ -78,7 +80,7 @@ func TestGetMaxRevision_ReflectsLatestApply(t *testing.T) {
}
func TestGetMaxRevision_UnchangedByQuery(t *testing.T) {
- id, entry := makeEntry(schema.KindStream, "g1", "s1", 100)
+ id, entry := makeEntry(schema.KindStream, "s1", 100)
srv := nodeStatusFixture(t, map[string]*cacheEntry{id: entry})
first, err := srv.GetMaxRevision(context.Background(), &clusterv1.GetMaxRevisionRequest{})
@@ -101,8 +103,8 @@ func TestGetMaxRevision_NilCache_ReturnsZero(t *testing.T) {
}
func TestGetKeyRevisions_PresentAndAbsent(t *testing.T) {
- idMeasure, entryMeasure := makeEntry(schema.KindMeasure, "g1", "m1", 50)
- idStream, entryStream := makeEntry(schema.KindStream, "g1", "s1", 70)
+ idMeasure, entryMeasure := makeEntry(schema.KindMeasure, "m1", 50)
+ idStream, entryStream := makeEntry(schema.KindStream, "s1", 70)
srv := nodeStatusFixture(t, map[string]*cacheEntry{
idMeasure: entryMeasure,
idStream: entryStream,
@@ -155,7 +157,7 @@ func TestGetKeyRevisions_NotYetNotified_ReportedAbsent(t *testing.T) {
// downstream handlers have not yet processed it. Per the gate semantics in
// GetKeyModRevision, the server must report it as not present.
c := newSchemaCache()
- id, entry := makeEntry(schema.KindMeasure, "g1", "m1", 100)
+ id, entry := makeEntry(schema.KindMeasure, "m1", 100)
c.entries[id] = entry
c.notifiedModRevision = 50 // below entry's rev
srv := NewNodeSchemaStatusServer(func() *schemaCache { return c })
@@ -179,8 +181,8 @@ func TestGetAbsentKeys_PostDelete(t *testing.T) {
// entry that never existed. Expectation: the deleted key and the
// never-existed key end up in absent_keys; the surviving key in
// still_present_keys.
- idAlive, entryAlive := makeEntry(schema.KindMeasure, "g1", "alive", 30)
- idDead, entryDead := makeEntry(schema.KindMeasure, "g1", "dead", 40)
+ idAlive, entryAlive := makeEntry(schema.KindMeasure, "alive", 30)
+ idDead, entryDead := makeEntry(schema.KindMeasure, "dead", 40)
c := newSchemaCache()
c.entries[idAlive] = entryAlive
c.entries[idDead] = entryDead
@@ -240,7 +242,7 @@ func TestGetKeyRevisions_ChunkedAtLimit(t *testing.T) {
keys := make([]*schemav1.SchemaKey, total)
for i := range total {
name := "m" + strconv.Itoa(i)
- id, entry := makeEntry(schema.KindMeasure, "g1", name, int64(i+1))
+ id, entry := makeEntry(schema.KindMeasure, name, int64(i+1))
entries[id] = entry
keys[i] = &schemav1.SchemaKey{Kind: "measure", Group: "g1", Name: name}
}