From f29590b1c6b86781e43bdb54a33d07201f669276 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Fri, 14 Aug 2026 15:08:02 +0800 Subject: [PATCH] fix(zookeeper): handle large children responses safely Closes #6200 --- .github/workflows/ci.yml | 4 ++ agents/drivers/zookeeper/agent_test.go | 20 ++++++ agents/drivers/zookeeper/connection.go | 43 ++++++++++-- agents/drivers/zookeeper/integration_test.go | 68 +++++++++++++++++++ .../src/components/kv/KvKeyBrowser.vue | 14 ++++ .../KvKeyBrowserSearchAndSplit.spec.ts | 6 ++ 6 files changed, 149 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fc57b200e8..ab47f3ae67 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -823,6 +823,10 @@ jobs: fi DBX_ZOOKEEPER_TEST_CONNECT_STRING=127.0.0.1:2181 \ go test -run '^TestZooKeeperIntegration$' -count=1 ./... + if [ "$version" = "3.7.0" ]; then + DBX_ZOOKEEPER_TEST_CONNECT_STRING=127.0.0.1:2181 \ + go test -run '^TestZooKeeperLargeChildrenIntegration$' -count=1 ./... + fi cleanup trap - EXIT done diff --git a/agents/drivers/zookeeper/agent_test.go b/agents/drivers/zookeeper/agent_test.go index 58508ef638..358cc2a746 100644 --- a/agents/drivers/zookeeper/agent_test.go +++ b/agents/drivers/zookeeper/agent_test.go @@ -66,6 +66,23 @@ func TestConnectionConfiguration(t *testing.T) { if resolveAuthScheme(connectionConfig{URLParams: "?foo=1;auth_scheme=SASL_DIGEST"}) != saslDigestAuthScheme { t.Fatal("URL auth_scheme was not resolved") } + for _, test := range []struct { + name string + config connectionConfig + expected int + }{ + {"default", connectionConfig{}, defaultMaxBufferSize}, + {"field", connectionConfig{MaxBufferSize: intPointer(64 * 1024 * 1024)}, 64 * 1024 * 1024}, + {"URL param", connectionConfig{URLParams: "?foo=1;max_buffer_size=16777216"}, 16 * 1024 * 1024}, + {"field precedence", connectionConfig{MaxBufferSize: intPointer(8 * 1024 * 1024), URLParams: "max_buffer_size=16777216"}, 8 * 1024 * 1024}, + } { + t.Run("max buffer "+test.name, func(t *testing.T) { + got, err := resolveMaxBufferSize(test.config) + if err != nil || got != test.expected { + t.Fatalf("resolveMaxBufferSize()=%d,%v want %d", got, err, test.expected) + } + }) + } if !hasTLSOptions(connectionConfig{CACertPath: "/tmp/ca.pem"}) { t.Fatal("TLS path was not detected") } @@ -88,6 +105,9 @@ func TestConnectionValidationHappensBeforeNetwork(t *testing.T) { {"sasl password", connectionConfig{AuthScheme: saslDigestAuthScheme, Username: "user"}, `password is required when auth_scheme = "sasl_digest"`}, {"negative base sleep", connectionConfig{BaseSleepTimeMS: intPointer(-1)}, "base_sleep_time_ms must be non-negative"}, {"negative retries", connectionConfig{MaxRetries: intPointer(-1)}, "max_retries must be non-negative"}, + {"invalid max buffer", connectionConfig{URLParams: "max_buffer_size=large"}, "max_buffer_size must be an integer number of bytes"}, + {"zero max buffer", connectionConfig{MaxBufferSize: intPointer(0)}, "max_buffer_size must be between 1 and"}, + {"oversized max buffer", connectionConfig{MaxBufferSize: intPointer(maximumMaxBufferSize + 1)}, "max_buffer_size must be between 1 and"}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { diff --git a/agents/drivers/zookeeper/connection.go b/agents/drivers/zookeeper/connection.go index a525438df3..1d033e977c 100644 --- a/agents/drivers/zookeeper/connection.go +++ b/agents/drivers/zookeeper/connection.go @@ -21,9 +21,12 @@ const ( defaultProbeTimeout = 2 * time.Second defaultBaseSleepTime = 250 * time.Millisecond defaultMaxRetries = 2 + defaultMaxBufferSize = 32 * 1024 * 1024 + maximumMaxBufferSize = 256 * 1024 * 1024 defaultPort = 2181 defaultAuthScheme = "digest" saslDigestAuthScheme = "sasl_digest" + maxBufferSizeParam = "max_buffer_size" statLookupConcurrencyEnv = "DBX_ZOOKEEPER_STAT_LOOKUP_CONCURRENCY" defaultStatLookupWorkers = 16 minimumStatLookupWorkers = 1 @@ -46,6 +49,7 @@ type connectionConfig struct { ConnectionTimeoutMS int `json:"connection_timeout_ms"` BaseSleepTimeMS *int `json:"base_sleep_time_ms"` MaxRetries *int `json:"max_retries"` + MaxBufferSize *int `json:"max_buffer_size"` SSL bool `json:"ssl"` CACertPath string `json:"ca_cert_path"` ClientCertPath string `json:"client_cert_path"` @@ -148,6 +152,10 @@ func openClient(config connectionConfig) (*clientSession, error) { if config.MaxRetries != nil && *config.MaxRetries < 0 { return nil, errors.New("max_retries must be non-negative") } + maxBufferSize, err := resolveMaxBufferSize(config) + if err != nil { + return nil, err + } target, err := parseConnectTarget(connectionString(config)) if err != nil { @@ -173,6 +181,7 @@ func openClient(config connectionConfig) (*clientSession, error) { sessionTimeout, zk.WithDialer(dialer), zk.WithLogInfo(false), + zk.WithMaxBufferSize(maxBufferSize), ) if err != nil { return nil, err @@ -360,16 +369,38 @@ func resolveAuthScheme(config connectionConfig) string { if strings.TrimSpace(config.AuthScheme) != "" { return strings.ToLower(strings.TrimSpace(config.AuthScheme)) } - params := strings.TrimPrefix(strings.TrimSpace(config.URLParams), "?") - params = strings.ReplaceAll(params, ";", "&") - if parsed, err := url.ParseQuery(params); err == nil { - if configured := strings.TrimSpace(parsed.Get("auth_scheme")); configured != "" { - return strings.ToLower(configured) - } + if configured := strings.TrimSpace(connectionURLParams(config).Get("auth_scheme")); configured != "" { + return strings.ToLower(configured) } return defaultAuthScheme } +func resolveMaxBufferSize(config connectionConfig) (int, error) { + configured := config.MaxBufferSize + if configured == nil { + value := strings.TrimSpace(connectionURLParams(config).Get(maxBufferSizeParam)) + if value == "" { + return defaultMaxBufferSize, nil + } + parsed, err := strconv.Atoi(value) + if err != nil { + return 0, fmt.Errorf("%s must be an integer number of bytes", maxBufferSizeParam) + } + configured = &parsed + } + if *configured <= 0 || *configured > maximumMaxBufferSize { + return 0, fmt.Errorf("%s must be between 1 and %d bytes", maxBufferSizeParam, maximumMaxBufferSize) + } + return *configured, nil +} + +func connectionURLParams(config connectionConfig) url.Values { + params := strings.TrimPrefix(strings.TrimSpace(config.URLParams), "?") + params = strings.ReplaceAll(params, ";", "&") + parsed, _ := url.ParseQuery(params) + return parsed +} + func hasTLSOptions(config connectionConfig) bool { return config.SSL || firstNonBlank( config.CACertPath, diff --git a/agents/drivers/zookeeper/integration_test.go b/agents/drivers/zookeeper/integration_test.go index 6dd9d9b13d..c2697d623e 100644 --- a/agents/drivers/zookeeper/integration_test.go +++ b/agents/drivers/zookeeper/integration_test.go @@ -4,8 +4,12 @@ import ( "encoding/json" "fmt" "os" + "strings" + "sync" "testing" "time" + + "github.com/go-zookeeper/zk" ) const integrationConnectStringEnv = "DBX_ZOOKEEPER_TEST_CONNECT_STRING" @@ -59,6 +63,70 @@ func TestZooKeeperIntegration(t *testing.T) { } } +func TestZooKeeperLargeChildrenIntegration(t *testing.T) { + connectString := os.Getenv(integrationConnectStringEnv) + if connectString == "" { + t.Skip(integrationConnectStringEnv + " is not set") + } + const ( + childCount = 1600 + childNameSize = 800 + workerCount = 32 + listLimit = 100 + ) + if childCount*(childNameSize+4) <= 1024*1024 { + t.Fatal("large children fixture must exceed the Java client's default 1 MiB receive limit") + } + + service := &server{statLookupConcurrency: 16} + connection := map[string]any{"zookeeper_connect_string": connectString} + if _, err := service.connect(mustJSON(map[string]any{"connection": connection})); err != nil { + t.Fatal(err) + } + t.Cleanup(service.closeClient) + root := fmt.Sprintf("/dbx-go-large-children-%d", time.Now().UnixNano()) + if _, err := service.activeClient.Create(root, nil, 0); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = service.delete(mustJSON(map[string]any{"key": root, "recursive": true})) }) + + jobs := make(chan int, childCount) + for index := 0; index < childCount; index++ { + jobs <- index + } + close(jobs) + errors := make(chan error, workerCount) + var waitGroup sync.WaitGroup + suffix := strings.Repeat("x", childNameSize-7) + for worker := 0; worker < workerCount; worker++ { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + for index := range jobs { + name := fmt.Sprintf("%06d-%s", index, suffix) + if _, err := service.activeClient.Create(root+"/"+name, nil, 0); err != nil && err != zk.ErrNodeExists { + errors <- err + return + } + } + }() + } + waitGroup.Wait() + close(errors) + for err := range errors { + t.Fatal(err) + } + + recursive := false + listed, err := service.listPrefix(mustJSON(listRequest{Prefix: root, Recursive: &recursive, Limit: listLimit})) + if err != nil { + t.Fatal(err) + } + if len(listed.Keys) != listLimit || listed.Continuation == nil { + t.Fatalf("list keys=%d continuation=%t", len(listed.Keys), listed.Continuation != nil) + } +} + func BenchmarkZooKeeperOperations(b *testing.B) { connectString := os.Getenv(integrationConnectStringEnv) if connectString == "" { diff --git a/apps/desktop/src/components/kv/KvKeyBrowser.vue b/apps/desktop/src/components/kv/KvKeyBrowser.vue index c450634209..028ffff0d8 100644 --- a/apps/desktop/src/components/kv/KvKeyBrowser.vue +++ b/apps/desktop/src/components/kv/KvKeyBrowser.vue @@ -250,6 +250,7 @@ const listRevision = ref(null); const listFilteredByAcls = ref(false); const loading = ref(false); const loadingMore = ref(false); +const listError = ref(""); const expandedGroupIds = ref>(new Set()); const selectedKey = ref(null); const selectedKeyIdentity = ref(null); @@ -592,6 +593,7 @@ async function loadKeys(reset = true, options: LoadKeysOptions = {}) { const keyIdentityToRestore = options.preserveSelection ? selectedKeyIdentity.value : null; if (reset) { loading.value = true; + listError.value = ""; continuation.value = null; listRevision.value = null; listFilteredByAcls.value = false; @@ -620,6 +622,10 @@ async function loadKeys(reset = true, options: LoadKeysOptions = {}) { clearSelectedKey(); } } + } catch (error) { + if (reset && generation === keyLoadGeneration && props.connectionId === connectionId) { + listError.value = error instanceof Error ? error.message : String(error); + } } finally { if (generation === keyLoadGeneration && props.connectionId === connectionId) { loading.value = false; @@ -642,6 +648,7 @@ async function loadLazyRoot(reset = true, options: LoadKeysOptions = {}) { }; const previousExpanded = new Set(expandedGroupIds.value); loading.value = true; + listError.value = ""; loadingMore.value = false; listFilteredByAcls.value = false; if (!options.preserveSelection) { @@ -695,6 +702,10 @@ async function loadLazyRoot(reset = true, options: LoadKeysOptions = {}) { } else { expandedGroupIds.value = focusedRootExpansion(rootPath); } + } catch (error) { + if (lazyLoadContextValid(context)) { + listError.value = error instanceof Error ? error.message : String(error); + } } finally { if (lazyLoadContextValid(context)) loading.value = false; } @@ -1867,6 +1878,9 @@ defineExpose({ {{ labels.loadingKeys }} +
+ {{ listError }} +
{{ labels.empty }}
diff --git a/apps/desktop/src/components/kv/__tests__/KvKeyBrowserSearchAndSplit.spec.ts b/apps/desktop/src/components/kv/__tests__/KvKeyBrowserSearchAndSplit.spec.ts index f395ccfc0d..f93c18a79e 100644 --- a/apps/desktop/src/components/kv/__tests__/KvKeyBrowserSearchAndSplit.spec.ts +++ b/apps/desktop/src/components/kv/__tests__/KvKeyBrowserSearchAndSplit.spec.ts @@ -40,6 +40,12 @@ describe("KvKeyBrowser search and split layout", () => { expect(browserSource).toContain("keyLoadGeneration++;"); }); + it("shows list request errors instead of the empty-state message", () => { + expect(browserSource).toContain('const listError = ref("");'); + expect(browserSource.match(/listError\.value = error instanceof Error \? error\.message : String\(error\);/g)).toHaveLength(2); + expect(browserSource.indexOf('v-else-if="listError"')).toBeLessThan(browserSource.indexOf('v-else-if="visibleRows.length === 0"')); + }); + it("highlights the matching search query only for the selected search result", () => { expect(browserSource).toContain("interface KvSearchHighlight"); expect(browserSource).toMatch(/const activeSearchHighlight = computed\(\(\) => \(?props\.searchHighlight\?\.key === selectedKey\.value/);