This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new da2513c9f fix(studio): scope ACL pushes, gate settings writes, fix
unmeasured values (#6851)
da2513c9f is described below
commit da2513c9f43e0b26b2540ba2bc56324e879576d0
Author: bai mengxuan <[email protected]>
AuthorDate: Tue Oct 27 20:06:09 2026 +0800
fix(studio): scope ACL pushes, gate settings writes, fix unmeasured values
(#6851)
* fix(acl): scope the broker push to the instance's cluster (#6851)
The ACL push resolved every master behind the instance's NameServer, so on a
NameServer shared by several clusters one instance's accounts and policies
were
written to brokers it is not bound to. masterAddresses() now takes the
instance
and filters through AclBrokerScope - the same scope the ACL read path
already
used - and an instance that names no cluster keeps every master, as before.
* fix(settings): prefer instance names in data source labels (#6267)
formatInstanceIds did a single find and OR-ed the results, so a data source
referencing both a name and a numeric key rendered the key. Look names up
first
and fall back to the id, which also fixes the exported CSV.
* fix(web): keep a reader out of the AI settings writes (#6865)
The AI, data source and cloud credential tabs had no role gate, so a reader
account reached the write handlers. Derive canWriteSettings once in a shared
useCanWriteSettings hook - the same expression GeneralSettingsTab already
used -
disable the write buttons and modal confirmations, and return early in the
handlers. The gate is checked before the submitting lock so a rejected
attempt
cannot leave the modal permanently locked.
* fix(clients): ignore source port changes in ID collision diagnostics
(#6203)
The client-id collision check compared full addresses, so a reconnect that
only
changed the TCP source port was reported as a critical collision. Compare
hosts
via the new connectionHost() helper.
* fix(cluster): count every listed connection in the client summary (#6321)
The summary tiles de-duplicated by type:clientId while the table, subtitle
and
footer counted rows, so the numbers disagreed. All five figures now come
from
the listed connection rows.
* fix(alert): invalidate test results when criteria change (#6297)
A dry-run result stayed on screen after the rule criteria changed, so an
operator
could act on a stale verdict. Track the 13 criteria fields, invalidate on
change,
and add a generation guard to handleTest so an in-flight response cannot
land on
a newer edit. Every clear path now goes through invalidateTestResult(),
which
also bumps the generation.
* fix(web): tolerate Git line endings in the license gate (#6189)
With core.autocrlf the vendored-icon comparison and the distributed LICENSE
were
byte-compared against LF-normalized manifests, so a Windows checkout failed
the
gate. Normalize both sides through eolInsensitive() and cover it with a CRLF
fixture that exercises the vendored SVG branch.
* fix(consumer): clear selection after confirmed empty refresh (#6352)
A refresh that confirmed the list is empty left the previous selection in
place,
so a following bulk action targeted rows that no longer exist.
* fix(instance): stamp the ownership reservation's bookkeeping columns
(#6668)
The two reservation INSERTs relied on database defaults for gmt_create and
gmt_modified, which left them null where no default applied and let the two
columns disagree. Write both explicitly from one LocalDateTime.now().
* fix(tencent): derive a topic consumer's subscription model from
MessageModel (#7033)
consumeType was read from the wrong field, inverting the clustered/broadcast
verdict for Tencent topic consumers. The parameter is renamed to
messageModel so
the signature does not contradict what it receives.
* fix(rmqctl): declare and enforce the paged tools' page size ceiling
(#6845)
Three paged tools documented pageSize with only a minimum, so the CLI
accepted
values the services reject with 400 (all three cap at 100). Declare
maximum: 100
in the catalog and carry it end to end; catalog_gen.go is regenerated and
`make catalog-verify` reports no drift. The constraint existed in
rmq-tools.yaml
before #4225 split the catalog and was dropped there, because the
generator's
allow-list did not include `maximum`.
Batch integration of eleven related fixes by the same contributor; #6189,
#6203,
#6267, #6297, #6321, #6352, #6668, #6845, #6865 and #7033 are closed as
merged
into this one. Verified on the combined branch: 4143 backend tests, 2009
frontend
tests across 166 files, license gate 8/8, production build, eslint with no
new
warnings, and catalog-verify / go test / go vet / make fmt all clean.
---
rmqctl/cmd/catalog.go | 3 +
rmqctl/cmd/catalog_bounds_test.go | 107 ++++++++++
rmqctl/cmd/catalog_validate.go | 8 +-
rmqctl/cmd/explain.go | 7 +
rmqctl/cmd/explain_test.go | 44 ++++
rmqctl/internal/catalog/catalog.go | 2 +
rmqctl/internal/catalog/catalog_gen.go | 8 +-
rmqctl/internal/catalog/generate/main.go | 9 +-
.../studio/instance/ResourceOwnershipGuard.java | 19 +-
.../studio/instance/acl/AclBrokerScope.java | 57 ++++++
.../rocketmq/studio/instance/acl/AclService.java | 15 +-
.../studio/instance/acl/ApacheAclReadService.java | 27 +--
.../provider/tencent/TencentInstanceProvider.java | 11 +-
.../src/main/resources/tool-catalog/tools/acl.yaml | 1 +
.../main/resources/tool-catalog/tools/audit.yaml | 1 +
.../main/resources/tool-catalog/tools/message.yaml | 1 +
.../instance/ResourceOwnershipGuardTest.java | 57 ++++++
.../studio/instance/acl/AclServiceTest.java | 62 ++++++
.../ops/ai/tool/catalog/ToolCatalogTest.java | 28 +++
.../tencent/TencentInstanceProviderTest.java | 20 ++
web/scripts/licenses.mjs | 13 +-
web/scripts/licenses.test.mjs | 83 +++++++-
web/src/hooks/useCanWriteSettings.test.tsx | 49 +++++
web/src/hooks/useCanWriteSettings.ts | 35 ++++
.../pages/cluster/__tests__/ClientsPage.test.tsx | 55 +++++
web/src/pages/cluster/clients.tsx | 22 +-
.../pages/instance/__tests__/ConsumerPage.test.tsx | 38 ++++
web/src/pages/instance/consumer.tsx | 2 +
.../ops/__tests__/AlertsTestResultResidue.test.tsx | 149 +++++++++++++-
web/src/pages/ops/alerts.tsx | 55 ++++-
web/src/pages/settings/AiAssistantTab.tsx | 18 +-
web/src/pages/settings/CloudCredentialTab.tsx | 16 +-
web/src/pages/settings/DataSourceTab.tsx | 25 ++-
web/src/pages/settings/GeneralSettingsTab.tsx | 12 +-
.../settings/__tests__/DataSourceTab.test.tsx | 121 +++++++++++
.../__tests__/SettingsWriteTabsReader.test.tsx | 227 +++++++++++++++++++++
web/src/utils/clientConnectionDiagnostics.test.ts | 45 ++++
web/src/utils/clientConnectionDiagnostics.ts | 12 +-
38 files changed, 1369 insertions(+), 95 deletions(-)
diff --git a/rmqctl/cmd/catalog.go b/rmqctl/cmd/catalog.go
index 735d2dce8..6f408801c 100644
--- a/rmqctl/cmd/catalog.go
+++ b/rmqctl/cmd/catalog.go
@@ -315,6 +315,9 @@ func schemaFlagUsage(field toolcatalog.Field) string {
if field.HasMinimum {
usage += fmt.Sprintf(" (minimum: %g)", field.Minimum)
}
+ if field.HasMaximum {
+ usage += fmt.Sprintf(" (maximum: %g)", field.Maximum)
+ }
if field.Required {
usage += " (required)"
}
diff --git a/rmqctl/cmd/catalog_bounds_test.go
b/rmqctl/cmd/catalog_bounds_test.go
new file mode 100644
index 000000000..0545964ed
--- /dev/null
+++ b/rmqctl/cmd/catalog_bounds_test.go
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the 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.
+ * The 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 cmd
+
+import (
+ "errors"
+ "strings"
+ "testing"
+
+ toolcatalog
"github.com/apache/rocketmq-dashboard/rmqctl/internal/catalog"
+ "github.com/apache/rocketmq-dashboard/rmqctl/internal/types"
+)
+
+// The paged tools hand pageSize to a service that rejects anything above 100
with a 400
+// ("pageSize must be between 1 and 100"), so the bounds belong to the tool
contract and not only to
+// the server: a caller that follows the schema must not be able to build a
call that fails at
+// execution, and `--help` has to name the ceiling the same way it names the
minimum.
+func TestPagedToolPageSizeCeilingTest(t *testing.T) {
+ for _, toolName := range []string{"rmq.acl.list", "rmq.audit.list",
"rmq.message.query_dlq"} {
+ t.Run(toolName, func(t *testing.T) {
+ tool, ok := toolcatalog.LookupTool(toolName)
+ if !ok {
+ t.Fatalf("tool %q is missing", toolName)
+ }
+ pageSize, ok := nestedField(tool, "page", "pageSize")
+ if !ok {
+ t.Fatalf("%s has no page.pageSize field",
toolName)
+ }
+ if !pageSize.HasMaximum || pageSize.Maximum != 100 {
+ t.Fatalf("%s page.pageSize = %+v; want maximum
100", toolName, pageSize)
+ }
+
+ if err := validateValue(tool, pageSize, int64(100));
err != nil {
+ t.Fatalf("%s rejects the boundary value 100:
%v", toolName, err)
+ }
+ for _, value := range []int64{101, 1000} {
+ err := validateValue(tool, pageSize, value)
+ if err == nil {
+ t.Fatalf("%s accepts pageSize %d, which
the service rejects at execution", toolName, value)
+ }
+ assertInvalidArgument(t, err, "must be at most
100")
+ }
+ assertInvalidArgument(t, validateValue(tool, pageSize,
int64(0)), "must be at least 1")
+
+ usage := schemaFlagUsage(pageSize)
+ if !strings.Contains(usage, "(maximum: 100)") {
+ t.Fatalf("%s --help usage = %q; want the
ceiling advertised next to the minimum", toolName, usage)
+ }
+ if !strings.Contains(usage, "(minimum: 1)") {
+ t.Fatalf("%s --help usage = %q; want the
minimum still advertised", toolName, usage)
+ }
+ })
+ }
+}
+
+func nestedField(tool toolcatalog.Tool, path ...string) (toolcatalog.Field,
bool) {
+ fields := tool.InputSchema.Fields
+ for index, name := range path {
+ var found *toolcatalog.Field
+ for position := range fields {
+ if fields[position].Name == name {
+ found = &fields[position]
+ break
+ }
+ }
+ if found == nil {
+ return toolcatalog.Field{}, false
+ }
+ if index == len(path)-1 {
+ return *found, true
+ }
+ if found.Object == nil {
+ return toolcatalog.Field{}, false
+ }
+ fields = found.Object.Fields
+ }
+ return toolcatalog.Field{}, false
+}
+
+func assertInvalidArgument(t *testing.T, err error, want string) {
+ t.Helper()
+ if err == nil {
+ t.Fatalf("err = nil, want INVALID_ARGUMENT containing %q", want)
+ }
+ var cliError *types.CLIError
+ if !errors.As(err, &cliError) || cliError.Code !=
types.CodeInvalidArgument {
+ t.Fatalf("err = %v, want INVALID_ARGUMENT", err)
+ }
+ if !strings.Contains(cliError.Message, want) {
+ t.Fatalf("message = %q, want it to contain %q",
cliError.Message, want)
+ }
+}
diff --git a/rmqctl/cmd/catalog_validate.go b/rmqctl/cmd/catalog_validate.go
index def11f17d..7f18fd5fd 100644
--- a/rmqctl/cmd/catalog_validate.go
+++ b/rmqctl/cmd/catalog_validate.go
@@ -110,7 +110,7 @@ func validateValue(tool toolcatalog.Tool, field
toolcatalog.Field, value any) er
tool.CommandPath(), field.Flag,
strings.Join(field.Enum, ", ")))
}
}
- if !field.HasMinimum {
+ if !field.HasMinimum && !field.HasMaximum {
return nil
}
var number float64
@@ -122,10 +122,14 @@ func validateValue(tool toolcatalog.Tool, field
toolcatalog.Field, value any) er
default:
return nil
}
- if number < field.Minimum {
+ if field.HasMinimum && number < field.Minimum {
return invalidArgument(fmt.Sprintf(
"%s --%s must be at least %g", tool.CommandPath(),
field.Flag, field.Minimum))
}
+ if field.HasMaximum && number > field.Maximum {
+ return invalidArgument(fmt.Sprintf(
+ "%s --%s must be at most %g", tool.CommandPath(),
field.Flag, field.Maximum))
+ }
return nil
}
diff --git a/rmqctl/cmd/explain.go b/rmqctl/cmd/explain.go
index 8a2c447ae..4bb3d98d1 100644
--- a/rmqctl/cmd/explain.go
+++ b/rmqctl/cmd/explain.go
@@ -98,6 +98,7 @@ type fieldExplanation struct {
Global bool `json:"global,omitempty" yaml:"global,omitempty"`
Enum []string `json:"enum,omitempty" yaml:"enum,omitempty"`
Minimum *float64 `json:"minimum,omitempty"
yaml:"minimum,omitempty"`
+ Maximum *float64 `json:"maximum,omitempty"
yaml:"maximum,omitempty"`
MinLength int `json:"minLength,omitempty"
yaml:"minLength,omitempty"`
}
@@ -162,6 +163,9 @@ func explainSchemaFields(schema toolcatalog.InputSchema,
prefix string) []fieldE
if field.HasMinimum {
explained.Minimum = new(field.Minimum)
}
+ if field.HasMaximum {
+ explained.Maximum = new(field.Maximum)
+ }
fields = append(fields, explained)
}
return fields
@@ -205,6 +209,9 @@ func renderResourceExplanation(out interface{ Write([]byte)
(int, error) }, expl
if field.Minimum != nil {
constraints = append(constraints,
fmt.Sprintf("minimum=%g", *field.Minimum))
}
+ if field.Maximum != nil {
+ constraints = append(constraints,
fmt.Sprintf("maximum=%g", *field.Maximum))
+ }
if field.MinLength > 0 {
constraints = append(constraints,
fmt.Sprintf("minLength=%d", field.MinLength))
}
diff --git a/rmqctl/cmd/explain_test.go b/rmqctl/cmd/explain_test.go
index 4325053b0..25cc3e8ff 100644
--- a/rmqctl/cmd/explain_test.go
+++ b/rmqctl/cmd/explain_test.go
@@ -60,6 +60,50 @@ func TestExplainResourceJSONComesFromCatalog(t *testing.T) {
}
}
+// TestExplainReportsTheDeclaredPageSizeCeiling verifies that explain reports
every bound the
+// catalog declares for a field. It is the machine-readable view agents build
calls from, so it
+// cannot be the one surface that omits the ceiling --help prints and the CLI
enforces.
+func TestExplainReportsTheDeclaredPageSizeCeiling(t *testing.T) {
+ stdout, stderr := &bytes.Buffer{}, &bytes.Buffer{}
+ exitCode := NewApp(stdout, stderr).Execute([]string{"--output", "json",
"explain", "acl"})
+ if exitCode != 0 {
+ t.Fatalf("explain failed: %s", stderr)
+ }
+ var explanation resourceExplanation
+ if err := json.Unmarshal(stdout.Bytes(), &explanation); err != nil {
+ t.Fatalf("decode output: %v", err)
+ }
+ var found *fieldExplanation
+ for _, action := range explanation.Actions {
+ for index := range action.Fields {
+ if action.Fields[index].Name == "page.pageSize" {
+ found = &action.Fields[index]
+ break
+ }
+ }
+ if found != nil {
+ break
+ }
+ }
+ if found == nil {
+ t.Fatalf("explain acl does not report page.pageSize: %#v",
explanation.Actions)
+ }
+ if found.Minimum == nil || *found.Minimum != 1 {
+ t.Fatalf("page.pageSize minimum = %v, want 1", found.Minimum)
+ }
+ if found.Maximum == nil || *found.Maximum != 100 {
+ t.Fatalf("page.pageSize maximum = %v, want 100", found.Maximum)
+ }
+
+ tableOut, tableErr := &bytes.Buffer{}, &bytes.Buffer{}
+ if code := NewApp(tableOut, tableErr).Execute([]string{"explain",
"acl"}); code != 0 {
+ t.Fatalf("explain acl failed: %s", tableErr)
+ }
+ if !strings.Contains(tableOut.String(), "minimum=1, maximum=100") {
+ t.Fatalf("the constraints column does not carry the
ceiling:\n%s", tableOut.String())
+ }
+}
+
// TestExplainMarksInstanceFieldGlobal verifies that the instance identifier
// field stays visible in explain output even though it is no longer a per-tool
// flag: it must be presented as the global --instance-id flag.
diff --git a/rmqctl/internal/catalog/catalog.go
b/rmqctl/internal/catalog/catalog.go
index cc0a86efb..edd58557a 100644
--- a/rmqctl/internal/catalog/catalog.go
+++ b/rmqctl/internal/catalog/catalog.go
@@ -113,6 +113,8 @@ type Field struct {
Enum []string
Minimum float64
HasMinimum bool
+ Maximum float64
+ HasMaximum bool
MinLength int
// ClientDefault records the x-client-default catalog extension. The
only
// supported value today is ClientDefaultNow ("NOW"): rmqctl fills the
diff --git a/rmqctl/internal/catalog/catalog_gen.go
b/rmqctl/internal/catalog/catalog_gen.go
index 39871e7fb..ffdc4f97a 100644
--- a/rmqctl/internal/catalog/catalog_gen.go
+++ b/rmqctl/internal/catalog/catalog_gen.go
@@ -21,7 +21,7 @@ package catalog
var defaultDocument = Document{
Version: "2.0.0",
MinimumClientVersion: "2.0.0",
- Digest:
"de70ac57695e5fb9db7b1a16d995c745eaa89c1e33f339954a9fb7831e7e9fd7",
+ Digest:
"e2ebdf66500503301025ed12f41ada2176e63249dce4732ee0e98c57300be682",
Tools: []Tool{
{
Name: "rmq.acl.list",
@@ -40,7 +40,7 @@ var defaultDocument = Document{
{Name: "page", Kind: ObjectField,
Object: &InputSchema{
Fields: []Field{
{Name: "page", Flag:
"page", Kind: IntegerField, Minimum: 1, HasMinimum: true},
- {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true},
+ {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true, Maximum:
100, HasMaximum: true},
},
}},
},
@@ -163,7 +163,7 @@ var defaultDocument = Document{
{Name: "page", Kind: ObjectField,
Object: &InputSchema{
Fields: []Field{
{Name: "page", Flag:
"page", Kind: IntegerField, Minimum: 1, HasMinimum: true},
- {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true},
+ {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true, Maximum:
100, HasMaximum: true},
},
}},
},
@@ -515,7 +515,7 @@ var defaultDocument = Document{
{Name: "page", Kind: ObjectField,
Object: &InputSchema{
Fields: []Field{
{Name: "page", Flag:
"page", Kind: IntegerField, Minimum: 1, HasMinimum: true},
- {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true},
+ {Name: "pageSize",
Flag: "page-size", Kind: IntegerField, Minimum: 1, HasMinimum: true, Maximum:
100, HasMaximum: true},
},
}},
},
diff --git a/rmqctl/internal/catalog/generate/main.go
b/rmqctl/internal/catalog/generate/main.go
index dc4fc78a8..c3a2c4f6f 100644
--- a/rmqctl/internal/catalog/generate/main.go
+++ b/rmqctl/internal/catalog/generate/main.go
@@ -75,6 +75,7 @@ type property struct {
Description string `yaml:"description"
json:"description,omitempty"`
Enum []string `yaml:"enum"
json:"enum,omitempty"`
Minimum *float64 `yaml:"minimum"
json:"minimum,omitempty"`
+ Maximum *float64 `yaml:"maximum"
json:"maximum,omitempty"`
MinLength int `yaml:"minLength"
json:"minLength,omitempty"`
Items *property `yaml:"items"
json:"items,omitempty"`
Required []string `yaml:"required,omitempty"
json:"required,omitempty"`
@@ -130,6 +131,7 @@ type goField struct {
Required bool
Enum []string
Minimum *float64
+ Maximum *float64
MinLength int
ClientDefault string
Object *goInputSchema
@@ -198,7 +200,7 @@ var defaultDocument = Document{
{{- end }}
},
}
-{{ define "field" }}{Name: {{ quote .Name }}{{ if .Flag }}, Flag: {{ quote
.Flag }}{{ end }}{{ if .Description }}, Description: {{ quote .Description }}{{
end }}, Kind: {{ .Kind }}{{ if .Required }}, Required: true{{ end }}{{ if .Enum
}}, Enum: {{ stringSlice .Enum }}{{ end }}{{ if .Minimum }}, Minimum: {{ number
.Minimum }}, HasMinimum: true{{ end }}{{ if .MinLength }}, MinLength: {{
.MinLength }}{{ end }}{{ if .ClientDefault }}, ClientDefault: {{ quote
.ClientDefault }}{{ end }}{{ with [...]
+{{ define "field" }}{Name: {{ quote .Name }}{{ if .Flag }}, Flag: {{ quote
.Flag }}{{ end }}{{ if .Description }}, Description: {{ quote .Description }}{{
end }}, Kind: {{ .Kind }}{{ if .Required }}, Required: true{{ end }}{{ if .Enum
}}, Enum: {{ stringSlice .Enum }}{{ end }}{{ if .Minimum }}, Minimum: {{ number
.Minimum }}, HasMinimum: true{{ end }}{{ if .Maximum }}, Maximum: {{ number
.Maximum }}, HasMaximum: true{{ end }}{{ if .MinLength }}, MinLength: {{
.MinLength }}{{ end }}{{ if [...]
Fields: []Field{
{{ range .Fields }}{{ template "field" . }},
{{ end }}},
@@ -236,7 +238,7 @@ func (schema *property) UnmarshalYAML(node *yaml.Node)
error {
// Reject unsupported input keywords instead of silently dropping
constraints.
for index := 0; index+1 < len(node.Content); index += 2 {
switch key := node.Content[index].Value; key {
- case "type", "description", "enum", "minimum", "minLength",
"items",
+ case "type", "description", "enum", "minimum", "maximum",
"minLength", "items",
"required", "properties", "additionalProperties",
"anyOf", "x-cli-flag",
"x-client-default":
default:
@@ -757,6 +759,9 @@ func compileSchema(schema inputSchema) (goInputSchema,
error) {
if field.Minimum != nil {
compiledField.Minimum = new(*field.Minimum)
}
+ if field.Maximum != nil {
+ compiledField.Maximum = new(*field.Maximum)
+ }
if kind == "ObjectField" {
object, err := compileSchema(field)
if err != nil {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
index b311253d7..0fec7e692 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuard.java
@@ -31,6 +31,7 @@ import
org.springframework.transaction.support.TransactionSynchronizationManager
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StringUtils;
+import java.time.LocalDateTime;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
@@ -195,14 +196,20 @@ public class ResourceOwnershipGuard {
Ownership existing = checkOwner(instance, resource,
read(domain, resource, target,
true).stream().findFirst().orElse(null), !create);
if (existing == null) {
+ // The reservation must stamp its own bookkeeping columns:
leaving them to the
+ // `DEFAULT CURRENT_TIMESTAMP` of the schema would store a
creation time taken from the
+ // database clock, while every later write (the create
path included) stamps
+ // gmt_modified from the JVM clock, so a row whose
database and studio run in different
+ // zones would report a creation time offset from its own
update time.
+ LocalDateTime stampedAt = LocalDateTime.now();
if (resource.kind() == Kind.TOPIC) {
- jdbc.update("INSERT INTO rmq_instance_topic (name,
instance_id, cluster_id, domain, status, topic_type)"
- + " VALUES (?, ?, ?, ?, 'PENDING',
?)", resource.name(), instance.getName(), target,
- domain, topicType == null ? "NORMAL" :
topicType);
+ jdbc.update("INSERT INTO rmq_instance_topic (name,
instance_id, cluster_id, domain, status, topic_type, gmt_create, gmt_modified)"
+ + " VALUES (?, ?, ?, ?, 'PENDING', ?,
?, ?)", resource.name(), instance.getName(), target,
+ domain, topicType == null ? "NORMAL" :
topicType, stampedAt, stampedAt);
} else {
- jdbc.update("INSERT INTO rmq_instance_group (name,
instance_id, cluster_id, domain, status)"
- + " VALUES (?, ?, ?, ?, 'PENDING')",
resource.name(), instance.getName(), target,
- domain);
+ jdbc.update("INSERT INTO rmq_instance_group (name,
instance_id, cluster_id, domain, status, gmt_create, gmt_modified)"
+ + " VALUES (?, ?, ?, ?, 'PENDING', ?,
?)", resource.name(), instance.getName(), target,
+ domain, stampedAt, stampedAt);
}
} else {
requireTopicType(existing, topicType);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclBrokerScope.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclBrokerScope.java
new file mode 100644
index 000000000..aa62bdc1d
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclBrokerScope.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.rocketmq.studio.instance.acl;
+
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+/**
+ * The brokers of an instance's endpoint that one ACL operation is allowed to
touch.
+ *
+ * <p>An instance bound to a single cluster on a NameServer shared by several
must neither surface nor
+ * mutate another cluster's brokers, so the topology is narrowed the same way
broker discovery narrows
+ * it ({@code RocketMQClusterProvider.discoverBrokers}). An instance that
names no cluster keeps every
+ * broker of the endpoint, and a configured cluster whose membership cannot be
resolved fails closed
+ * instead of falling back to the whole cluster table.
+ *
+ * <p>The ACL read and the ACL push both resolve their broker set through this
method: if they
+ * disagreed, the page would show one cluster's rules while a push wrote the
dashboard's accounts and
+ * policies into another cluster's brokers.
+ */
+final class AclBrokerScope {
+
+ private AclBrokerScope() {
+ }
+
+ static Set<String> scopedBrokerNames(ClusterInfo clusterInfo, String
configuredCluster) {
+ Set<String> names = new
TreeSet<>(clusterInfo.getBrokerAddrTable().keySet());
+ if (configuredCluster == null) {
+ return names;
+ }
+ Map<String, Set<String>> clusters = clusterInfo.getClusterAddrTable();
+ if (clusters == null || clusters.containsKey(configuredCluster)
+ && clusters.get(configuredCluster) == null) {
+ throw new BusinessException(503, "Broker cluster membership is
unavailable");
+ }
+ names.retainAll(clusters.getOrDefault(configuredCluster, Set.of()));
+ return names;
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
index 224d2482d..666350a90 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/AclService.java
@@ -369,7 +369,7 @@ public class AclService {
log.info("Pushing {} ACL account(s) of instance {} to the brokers",
plans.size(), instanceId);
return runtimeAdminClientResolver.execute(instanceId, admin -> {
DefaultMQAdminExt adminExt = (DefaultMQAdminExt) admin;
- List<String> masters = masterAddresses(adminExt);
+ List<String> masters = masterAddresses(adminExt, instanceId);
if (!masters.isEmpty() && !acl2Enabled(adminExt, masters.get(0))) {
throw new BusinessException(409, "Broker ACL 2.0 is not
enabled on "
+ masters.get(0) + "; set authenticationEnabled=true
and "
@@ -501,12 +501,21 @@ public class AclService {
}
}
- private List<String> masterAddresses(DefaultMQAdminExt adminExt) throws
Exception {
+ /**
+ * The master brokers this push may write to. The instance's configured
cluster decides the set,
+ * through the same scope the ACL read uses ({@link AclBrokerScope}): a
NameServer shared by
+ * several clusters must not receive this instance's accounts and policies
on the clusters it is
+ * not bound to. An instance that names no cluster keeps every master of
the endpoint.
+ */
+ private List<String> masterAddresses(DefaultMQAdminExt adminExt, String
instanceId) throws Exception {
ClusterInfo clusterInfo = adminExt.examineBrokerClusterInfo();
if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
return List.of();
}
- return clusterInfo.getBrokerAddrTable().values().stream()
+ String configuredCluster = StringUtils.hasText(instanceId)
+ ? runtimeAdminClientResolver.configuredClusterName(instanceId)
: null;
+ return AclBrokerScope.scopedBrokerNames(clusterInfo,
configuredCluster).stream()
+ .map(clusterInfo.getBrokerAddrTable()::get)
.filter(broker -> broker != null && broker.getBrokerAddrs() !=
null)
.map(broker -> broker.getBrokerAddrs().get(MixAll.MASTER_ID))
.filter(Objects::nonNull)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
index a4b729333..fe813cabd 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/acl/ApacheAclReadService.java
@@ -22,7 +22,6 @@ import org.apache.rocketmq.remoting.protocol.body.AclInfo;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
-import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
@@ -30,8 +29,6 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
-import java.util.TreeSet;
@Service
@RequiredArgsConstructor
@@ -48,7 +45,7 @@ public class ApacheAclReadService {
if (clusterInfo == null || clusterInfo.getBrokerAddrTable() ==
null) {
return result(policies, failures);
}
- for (String brokerName : scopedBrokerNames(clusterInfo,
configuredCluster)) {
+ for (String brokerName :
AclBrokerScope.scopedBrokerNames(clusterInfo, configuredCluster)) {
String address =
masterAddress(clusterInfo.getBrokerAddrTable().get(brokerName));
if (address == null || policies.containsKey(address) ||
failures.containsKey(address)) {
continue;
@@ -63,28 +60,6 @@ public class ApacheAclReadService {
});
}
- /**
- * The broker set this read is allowed to touch. An instance bound to one
cluster on a NameServer
- * shared by several must not surface another cluster's ACL, so the
topology is narrowed the same
- * way {@code RocketMQClusterProvider.discoverBrokers} narrows it; an
instance that names no
- * cluster keeps every broker of the endpoint. A configured cluster whose
membership is missing
- * from the topology cannot be resolved, so the read fails closed instead
of reading the whole
- * cluster table (mirrors the 503 that broker discovery raises for the
same condition).
- */
- private Set<String> scopedBrokerNames(ClusterInfo clusterInfo, String
configuredCluster) {
- Set<String> names = new
TreeSet<>(clusterInfo.getBrokerAddrTable().keySet());
- if (configuredCluster == null) {
- return names;
- }
- Map<String, Set<String>> clusters = clusterInfo.getClusterAddrTable();
- if (clusters == null || clusters.containsKey(configuredCluster)
- && clusters.get(configuredCluster) == null) {
- throw new BusinessException(503, "Broker cluster membership is
unavailable");
- }
- names.retainAll(clusters.getOrDefault(configuredCluster, Set.of()));
- return names;
- }
-
private List<RemoteAclPolicyVO> mapPolicies(List<AclInfo> brokerPolicies) {
if (brokerPolicies == null) {
return List.of();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
index c3d0d7632..5b6de1d07 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProvider.java
@@ -1024,7 +1024,9 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
return TopicConsumerVO.builder()
.group(subscription.getConsumerGroup())
- .consumeType(toConsumeType(subscription.getConsumeType(),
messageModel))
+ // ConsumeType is the vendor's consumption mechanism
(PUSH/PULL/POP); the
+ // subscription model the contract carries is MessageModel.
+ .consumeType(toConsumeType(messageModel,
subscription.getConsumeType()))
.messageModel(messageModel)
.diffTotal(subscription.getConsumerLag() == null ? 0L :
subscription.getConsumerLag())
.build();
@@ -1155,8 +1157,11 @@ public class TencentInstanceProvider implements
InstanceProvider {
}
}
- private static ConsumeType toConsumeType(String consumeType, String
messageModel) {
- String value = StringUtils.hasText(consumeType) ? consumeType :
messageModel;
+ private static ConsumeType toConsumeType(String messageModel, String
consumeType) {
+ // Tencent's ConsumeType is the consumption mechanism (PUSH/PULL/POP)
and never carries the
+ // subscription model, so MessageModel is the one to read; the
mechanism stays as the fallback
+ // for rows that only report it.
+ String value = StringUtils.hasText(messageModel) ? messageModel :
consumeType;
if (!StringUtils.hasText(value)) {
return null;
}
diff --git a/server/src/main/resources/tool-catalog/tools/acl.yaml
b/server/src/main/resources/tool-catalog/tools/acl.yaml
index 9f649af3e..bf8e8694e 100644
--- a/server/src/main/resources/tool-catalog/tools/acl.yaml
+++ b/server/src/main/resources/tool-catalog/tools/acl.yaml
@@ -37,6 +37,7 @@ tools:
pageSize:
type: integer
minimum: 1
+ maximum: 100
outputSchema:
type: object
required:
diff --git a/server/src/main/resources/tool-catalog/tools/audit.yaml
b/server/src/main/resources/tool-catalog/tools/audit.yaml
index 8609e7ef8..e64279dec 100644
--- a/server/src/main/resources/tool-catalog/tools/audit.yaml
+++ b/server/src/main/resources/tool-catalog/tools/audit.yaml
@@ -35,6 +35,7 @@ tools:
pageSize:
type: integer
minimum: 1
+ maximum: 100
outputSchema:
type: object
required:
diff --git a/server/src/main/resources/tool-catalog/tools/message.yaml
b/server/src/main/resources/tool-catalog/tools/message.yaml
index 52fa0f448..c43e04b15 100644
--- a/server/src/main/resources/tool-catalog/tools/message.yaml
+++ b/server/src/main/resources/tool-catalog/tools/message.yaml
@@ -300,6 +300,7 @@ tools:
pageSize:
type: integer
minimum: 1
+ maximum: 100
outputSchema:
type: object
required:
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
index 5a8d5c57b..57b5ab340 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/ResourceOwnershipGuardTest.java
@@ -23,6 +23,7 @@ import
org.apache.rocketmq.studio.provider.apache.RocketMQLiteTopicProvider;
import
org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.studio.persistence.entity.RmqGroup;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.common.util.JdbcRowValues;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Kind;
import org.apache.rocketmq.studio.instance.ResourceOwnershipGuard.Resource;
import
org.apache.rocketmq.studio.provider.apache.RocketMQDefaultClusterResolver;
@@ -60,6 +61,7 @@ import
org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.support.TransactionTemplate;
+import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
@@ -688,6 +690,61 @@ class ResourceOwnershipGuardTest {
return info;
}
+ /**
+ * A reservation has to carry its own bookkeeping stamps. The shipped
columns are declared
+ * {@code DEFAULT CURRENT_TIMESTAMP}, which the database evaluates on its
own clock, while the create
+ * path stamps {@code gmt_modified} from the JVM clock — so a reservation
that leans on the column
+ * default stores a creation time taken from a different clock than the
modification time, and the
+ * two can disagree by the offset between the database and the studio
(eight hours when the database
+ * runs in UTC and the studio in Asia/Shanghai). The two columns are
recreated without their defaults
+ * here, so that "the reservation supplied the value" is what this test
observes.
+ */
+ @Test
+ void topicReservationSuppliesItsOwnBookkeepingStampsTest() {
+ dropTimestampDefaults("topic");
+ LocalDateTime before = LocalDateTime.now();
+ guard.write(first, topic, "cluster", true, "NORMAL", () -> null);
+ LocalDateTime after = LocalDateTime.now();
+ Map<String, Object> row = jdbc.queryForMap(
+ "SELECT gmt_create, gmt_modified FROM rmq_instance_topic WHERE
name = ?", topic.name());
+ assertStampedByTheStudioClock(row, before, after);
+ }
+
+ @Test
+ void groupReservationSuppliesItsOwnBookkeepingStampsTest() {
+ dropTimestampDefaults("group");
+ Resource group = new Resource(Kind.GROUP, "orders-group");
+ LocalDateTime before = LocalDateTime.now();
+ guard.write(first, group, "cluster", true, null, () -> null);
+ LocalDateTime after = LocalDateTime.now();
+ Map<String, Object> row = jdbc.queryForMap(
+ "SELECT gmt_create, gmt_modified FROM rmq_instance_group WHERE
name = ?", group.name());
+ assertStampedByTheStudioClock(row, before, after);
+ }
+
+ /**
+ * Both columns have to carry the value the reservation supplied, not
merely a non-null one. Equality
+ * locks the single {@code stampedAt} the two columns share, and the
window proves the studio clock
+ * produced them — a database default would land outside it whenever the
two clocks disagree, which is
+ * the hours-wide zone offset this fix removes. The one-second slack
absorbs the second precision the
+ * DATETIME columns are stored with, which floors the stamped value.
+ */
+ private static void assertStampedByTheStudioClock(Map<String, Object> row,
LocalDateTime before, LocalDateTime after) {
+ LocalDateTime created = JdbcRowValues.dateTimeValue(row, "gmt_create");
+ LocalDateTime modified = JdbcRowValues.dateTimeValue(row,
"gmt_modified");
+ assertThat(created).as("gmt_create").isNotNull();
+ assertThat(modified).as("gmt_modified").isNotNull();
+ assertThat(created).as("gmt_create and gmt_modified come from one
stampedAt").isEqualTo(modified);
+ assertThat(created).isAfterOrEqualTo(before.minusSeconds(1));
+ assertThat(modified).isBeforeOrEqualTo(after.plusSeconds(1));
+ }
+
+ /** Same table, but the two bookkeeping columns no longer fall back to the
database clock. */
+ private void dropTimestampDefaults(String kind) {
+ jdbc.execute("ALTER TABLE rmq_instance_" + kind + " ALTER COLUMN
gmt_create DROP DEFAULT");
+ jdbc.execute("ALTER TABLE rmq_instance_" + kind + " ALTER COLUMN
gmt_modified DROP DEFAULT");
+ }
+
private void legacyIndexes() {
for (String kind : List.of("topic", "group")) {
jdbc.execute("ALTER TABLE rmq_instance_" + kind + " DROP INDEX
uk_" + kind + "_domain");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
index c91836d45..0547fa804 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/acl/AclServiceTest.java
@@ -55,6 +55,8 @@ import java.util.Locale;
import java.util.Optional;
import java.util.Properties;
import java.util.TimeZone;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
@@ -1504,6 +1506,66 @@ class AclServiceTest {
verify(admin, never()).updateAcl(any(), any(AclInfo.class));
}
+ @Test
+ void
pushAclToBrokerTouchesOnlyTheConfiguredClusterOfASharedNameServerTest() throws
Exception {
+ AclUserVO stored =
AclUserVO.builder().id(11L).username("orders").secretKey("secret-4").build();
+ when(aclRepository.findUsers()).thenReturn(List.of(stored));
+ when(aclRepository.findUserById(11L)).thenReturn(Optional.of(stored));
+ when(aclRepository.findRulePage("user:orders", null, null, null, null,
1, 200))
+ .thenReturn(PageResult.of(List.of(), 0, 1, 200));
+ when(clusterProvider.discoverBrokers("instance-a",
null)).thenReturn(List.of());
+
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+
when(admin.examineBrokerClusterInfo()).thenReturn(sharedNameServerTopology());
+
when(runtimeAdminClientResolver.configuredClusterName("instance-a")).thenReturn("cluster-a");
+ when(runtimeAdminClientResolver.execute(eq("instance-a"),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(call -> call.getArgument(1,
MqAdminExtFactory.AdminAction.class).apply(admin));
+
+ PushAclToBrokerResultVO result =
aclService.pushAclToBroker("instance-a");
+
+ // The instance is bound to cluster-a; the other cluster behind the
same NameServer must not
+ // receive this instance's accounts, policies or pre-flight config
reads.
+ assertThat(result.getBrokers()).containsExactly("10.0.0.1:10911");
+ assertThat(result.getAccounts().get(0).getBrokers())
+ .allSatisfy(broker ->
assertThat(broker.getBroker()).isEqualTo("10.0.0.1:10911"));
+ verify(admin).createUser(eq("10.0.0.1:10911"), any());
+ verify(admin, never()).createUser(eq("10.0.0.2:10911"), any());
+ verify(admin, never()).updateUser(eq("10.0.0.2:10911"), any());
+ verify(admin, never()).updateAcl(eq("10.0.0.2:10911"),
any(AclInfo.class));
+ verify(admin, never()).getBrokerConfig("10.0.0.2:10911");
+ }
+
+ @Test
+ void pushAclToBrokerKeepsEveryMasterWhenTheInstanceNamesNoClusterTest()
throws Exception {
+ AclUserVO stored =
AclUserVO.builder().id(12L).username("reader").secretKey("secret-5").build();
+ when(aclRepository.findUsers()).thenReturn(List.of(stored));
+ when(aclRepository.findUserById(12L)).thenReturn(Optional.of(stored));
+ when(aclRepository.findRulePage("user:reader", null, null, null, null,
1, 200))
+ .thenReturn(PageResult.of(List.of(), 0, 1, 200));
+ when(clusterProvider.discoverBrokers("instance-a",
null)).thenReturn(List.of());
+
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+
when(admin.examineBrokerClusterInfo()).thenReturn(sharedNameServerTopology());
+
when(runtimeAdminClientResolver.configuredClusterName("instance-a")).thenReturn(null);
+ when(runtimeAdminClientResolver.execute(eq("instance-a"),
any(MqAdminExtFactory.AdminAction.class)))
+ .thenAnswer(call -> call.getArgument(1,
MqAdminExtFactory.AdminAction.class).apply(admin));
+
+ PushAclToBrokerResultVO result =
aclService.pushAclToBroker("instance-a");
+
+
assertThat(result.getBrokers()).containsExactlyInAnyOrder("10.0.0.1:10911",
"10.0.0.2:10911");
+ }
+
+ private ClusterInfo sharedNameServerTopology() {
+ BrokerData first = new BrokerData("cluster-a", "broker-a", new
HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ BrokerData second = new BrokerData("cluster-b", "broker-b", new
HashMap<>(Map.of(0L, "10.0.0.2:10911")));
+ ClusterInfo clusterInfo = new ClusterInfo();
+ clusterInfo.setBrokerAddrTable(new HashMap<>(Map.of("broker-a", first,
"broker-b", second)));
+ clusterInfo.setClusterAddrTable(new HashMap<>(Map.of(
+ "cluster-a", new TreeSet<>(Set.of("broker-a")),
+ "cluster-b", new TreeSet<>(Set.of("broker-b")))));
+ return clusterInfo;
+ }
+
private ClusterInfo clusterInfoWithMaster(String address) {
ClusterInfo clusterInfo = new ClusterInfo();
BrokerData broker = new BrokerData("cluster-a", "broker-a", new
HashMap<>(Map.of(0L, address)));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
index f6178e71b..bad84f9a5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/ai/tool/catalog/ToolCatalogTest.java
@@ -27,6 +27,7 @@ import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.core.io.Resource;
import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import tools.jackson.databind.json.JsonMapper;
@@ -107,6 +108,33 @@ class ToolCatalogTest {
}
}
+ @Test
+ void rejectsPageSizesAboveTheCeilingThePagedServicesEnforceTest() {
+ ToolCatalog catalog = new ToolCatalog(new DefaultResourceLoader());
+ ToolSchemaValidator validator = new ToolSchemaValidator(catalog,
+ new LegacyJackson2Config().jackson2ObjectMapper(),
JsonMapper.builder().build());
+ // The audit feed is platform level, the other two are scoped to one
instance.
+ Map<String, Map<String, Object>> scopes = Map.of(
+ "rmq.acl.list", Map.of("instanceId", "instance-a"),
+ "rmq.audit.list", Map.of(),
+ "rmq.message.query_dlq", Map.of("instanceId", "instance-a"));
+ for (Map.Entry<String, Map<String, Object>> scoped :
scopes.entrySet()) {
+ ToolDefinition definition = catalog.getDefinition(scoped.getKey());
+ Map<String, Object> overLimit = new HashMap<>(scoped.getValue());
+ overLimit.put("page", Map.of("page", 1, "pageSize", 101));
+ Map<String, Object> atLimit = new HashMap<>(scoped.getValue());
+ atLimit.put("page", Map.of("page", 1, "pageSize", 100));
+
+ // The services answer 400 ("pageSize must be between 1 and 100")
for anything above the
+ // ceiling, so a schema-valid call must not be able to ask for it
in the first place. The
+ // boundary value stays valid: rejecting 100 would invent a
ceiling the service does not have.
+ assertThatThrownBy(() -> validator.validateInput(definition,
overLimit))
+ .isInstanceOfSatisfying(ToolExecutionException.class,
+ error ->
assertThat(error.getCode()).isEqualTo(400));
+ validator.validateInput(definition, atLimit);
+ }
+ }
+
@Test
void rejectsCatalogThatDoesNotMatchItsJsonSchema() {
Resource invalid = utf8Resource("""
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
index 5ef54fb12..656da42a7 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/tencent/TencentInstanceProviderTest.java
@@ -575,6 +575,26 @@ class TencentInstanceProviderTest {
assertThat(consumers.get(0).getDiffTotal()).isEqualTo(42L);
}
+ @Test
+ void
getTopicConsumersShouldDeriveTheSubscriptionModelFromTheMessageModelTest()
throws Exception {
+ SubscriptionData subscription = new SubscriptionData();
+ subscription.setConsumerGroup("GID_orders");
+ // The vendor reports the consumption mechanism in ConsumeType
(PUSH/PULL/POP); the
+ // subscription model the contract carries lives in MessageModel.
+ subscription.setConsumeType("PUSH");
+ subscription.setMessageModel("CLUSTERING");
+ subscription.setConsumerLag(42L);
+ DescribeTopicResponse response = new DescribeTopicResponse();
+ response.setSubscriptionData(new SubscriptionData[]{subscription});
+ when(client.DescribeTopic(any())).thenReturn(response);
+
+ List<TopicConsumerVO> consumers =
provider.getTopicConsumers(STUDIO_INSTANCE_ID, "orders");
+
+ assertThat(consumers).hasSize(1);
+
assertThat(consumers.get(0).getConsumeType()).isEqualTo(ConsumeType.CLUSTERING);
+ assertThat(consumers.get(0).getMessageModel()).isEqualTo("CLUSTERING");
+ }
+
@Test
void getTopicConsumersShouldReadEverySubscriptionPageTest() throws
Exception {
DescribeTopicResponse firstPage = new DescribeTopicResponse();
diff --git a/web/scripts/licenses.mjs b/web/scripts/licenses.mjs
index 42278db04..85f94bfa7 100644
--- a/web/scripts/licenses.mjs
+++ b/web/scripts/licenses.mjs
@@ -28,6 +28,8 @@ const read = (name) => {
if (!data.toString().trim()) throw new Error(`empty license material:
${name}`);
return data;
};
+// Git may hand back CRLF for tracked text; compare and distribute the
canonical upstream LF bytes.
+const eolInsensitive = (data) => Buffer.from(data.toString().replace(/\r\n/g,
'\n'));
const json = (name) => JSON.parse(read(name));
const slash = (name) => name.split(path.sep).join('/');
@@ -62,7 +64,7 @@ export function collectLicenses(moduleIds, base = root) {
const packages = new Map();
const files = new Map();
const components = [];
- const baseText = (name) => read(path.join(base,
name)).toString().split('\nThird-party source materials\n')[0];
+ const baseText = (name) => read(path.join(base,
name)).toString().replace(/\r\n/g, '\n').split('\nThird-party source
materials\n')[0];
let license = baseText('LICENSE');
let notice = baseText('NOTICE');
const ids = [...new Set(moduleIds)].sort();
@@ -120,8 +122,10 @@ export function collectLicenses(moduleIds, base = root) {
if (identity === '[email protected]') {
// The 1.0.6 npm package omits LICENSE; the index.js from the immutable
upstream commit below is identical to the package, and the MIT text is supplied.
const fallback = path.join(base,
'licenses/toggle-selection-1.0.6/LICENSE');
+ // Git may convert this tracked text to CRLF; verify and distribute the
canonical upstream LF bytes.
+ sourceContent[fallback] =
Buffer.from(read(fallback).toString().replace(/\r\n/g, '\n'));
if (sha(read(path.join(dir, 'index.js'))) !==
'd1a1caf366f8ae5ed3cf4a87c42c46e73bb5536acbcbc5ca479c180d4d3e7756'
- || sha(read(fallback)) !==
'5149051aed807f78acfbf9a43ac66368374a8fa1f9dfc092b73de5a67d42673a') {
+ || sha(sourceContent[fallback]) !==
'5149051aed807f78acfbf9a43ac66368374a8fa1f9dfc092b73de5a67d42673a') {
throw new Error('toggle-selection code or upstream license text does
not match the verified source');
}
sources.push(fallback, path.join(dir, 'README.md'));
@@ -134,12 +138,15 @@ export function collectLicenses(moduleIds, base = root) {
const icons = readdirSync(vendoredDir).filter((name) =>
name.endsWith('.svg')).sort();
if (icons.length !== 17) throw new Error('the vendored SVG list changed;
the source must be re-approved');
for (const name of icons) {
- if (!read(path.join(vendoredDir, name)).equals(read(path.join(dir,
'icons', name)))) {
+ // Both sides are text: the tracked copy may be CRLF, so compare the
normalized bytes.
+ if (!eolInsensitive(read(path.join(vendoredDir,
name))).equals(eolInsensitive(read(path.join(dir, 'icons', name))))) {
throw new Error(`a vendored icon does not match the locked upstream
text: ${name}`);
}
}
const fallback = path.join(vendoredDir, 'LICENSE');
sources = [...new Set([...sources, fallback])];
+ // Keep the distributed text and its manifest checksum on the canonical
LF bytes.
+ sourceContent[fallback] = eolInsensitive(read(fallback));
sourceURLs[fallback] =
'https://github.com/lobehub/lobe-icons/blob/v1.95.0/LICENSE';
notice += '\nModel brand icons from LobeHub
(https://github.com/lobehub/lobe-icons):\nCopyright (c) 2023 LobeHub.\n';
}
diff --git a/web/scripts/licenses.test.mjs b/web/scripts/licenses.test.mjs
index 35c100e87..5ee749060 100644
--- a/web/scripts/licenses.test.mjs
+++ b/web/scripts/licenses.test.mjs
@@ -15,7 +15,7 @@
* limitations under the License.
*/
import assert from 'node:assert/strict';
-import { mkdtempSync, mkdirSync, readFileSync, writeFileSync, rmSync } from
'node:fs';
+import { mkdtempSync, mkdirSync, readFileSync, readdirSync, writeFileSync,
rmSync } from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import test from 'node:test';
@@ -51,6 +51,81 @@ test('upstreamFallbacksKeepCompleteTextTest', () => {
assert.match(result.files.get('legal/licenses/@ant-design/[email protected]/LICENSE.txt').toString(),
/2018-present Ant UED/);
});
+function lineEndingFixture(t, ending = '\n') {
+ const directory = temporary(t);
+ const license = readFileSync(path.join(root,
'licenses/toggle-selection-1.0.6/LICENSE'), 'utf8').replace(/\r\n/g, '\n');
+ for (const name of ['LICENSE', 'NOTICE']) {
+ const text = readFileSync(path.join(root, name), 'utf8').replace(/\r\n/g,
'\n');
+ write(directory, name, text.replace(/\n/g, ending));
+ }
+ for (const name of ['package.json', 'index.js', 'README.md']) {
+ write(directory, `node_modules/toggle-selection/${name}`,
readFileSync(path.join(root, 'node_modules/toggle-selection', name)));
+ }
+ write(directory, 'licenses/toggle-selection-1.0.6/LICENSE',
license.replace(/\n/g, ending));
+ // The vendored LobeHub icons are tracked too, so Git can hand them back as
CRLF. The installed
+ // package has no root LICENSE: the vendored one is the only license text
this component carries.
+ const logoLicense = readFileSync(path.join(root,
'src/assets/model-logos/LICENSE'), 'utf8').replace(/\r\n/g, '\n');
+ write(directory, 'src/assets/model-logos/LICENSE',
logoLicense.replace(/\n/g, ending));
+ write(directory, 'node_modules/@lobehub/icons-static-svg/package.json',
+ readFileSync(path.join(root,
'node_modules/@lobehub/icons-static-svg/package.json')));
+ const icons = readdirSync(path.join(root,
'src/assets/model-logos')).filter((name) => name.endsWith('.svg')).sort();
+ for (const name of icons) {
+ const svg = readFileSync(path.join(root, 'src/assets/model-logos', name),
'utf8').replace(/\r\n/g, '\n');
+ write(directory, `src/assets/model-logos/${name}`, svg.replace(/\n/g,
ending));
+ write(directory, `node_modules/@lobehub/icons-static-svg/icons/${name}`,
+ readFileSync(path.join(root,
'node_modules/@lobehub/icons-static-svg/icons', name)));
+ }
+ return { directory, license, logoLicense };
+}
+
+test('gitLineEndingsPreserveVerifiedLicenseAndBinaryAttributionTest', (t) => {
+ const lf = lineEndingFixture(t);
+ const crlf = lineEndingFixture(t, '\r\n');
+ const modules = ['node_modules/toggle-selection/index.js'];
+ const expected = collectLicenses(modules, lf.directory);
+ const actual = collectLicenses(modules, crlf.directory);
+ assert.deepEqual(actual.components, expected.components);
+ assert.deepEqual(actual.files, expected.files);
+
assert.equal(actual.files.get('legal/licenses/[email protected]/LICENSE.txt').toString(),
lf.license);
+ for (const name of ['LICENSE', 'NOTICE']) {
+ assert.doesNotMatch(actual.files.get(name).toString(), /Third-party source
materials|src\/assets\/model-logos/);
+ }
+});
+
+test('gitLineEndingsPreserveTheVendoredIconAttributionTest', (t) => {
+ const lf = lineEndingFixture(t);
+ const crlf = lineEndingFixture(t, '\r\n');
+ const modules = ['src/assets/model-logos/openai.svg',
'node_modules/toggle-selection/index.js'];
+ const expected = collectLicenses(modules, lf.directory);
+ const actual = collectLicenses(modules, crlf.directory);
+ assert.deepEqual(actual.components, expected.components);
+ assert.deepEqual(actual.files, expected.files);
+
assert.equal(actual.files.get('legal/licenses/@lobehub/[email protected]/LICENSE.txt').toString(),
lf.logoLicense);
+ assert.match(actual.files.get('NOTICE').toString(), /Copyright \(c\) 2023
LobeHub/);
+ assert.doesNotMatch(actual.files.get('LICENSE').toString(), /Third-party
source materials|src\/assets\/model-logos/);
+ // Every tracked icon is a single line today, so Git leaves them alone and
the byte comparison above
+ // proves nothing on its own. Re-wrap one pair to keep the icon gate honest
about line endings.
+ const icon = 'openai.svg';
+ const upstream = readFileSync(path.join(root,
'node_modules/@lobehub/icons-static-svg/icons', icon), 'utf8')
+ .replace(/\r\n/g, '\n').replace('</svg>', '\n</svg>');
+ write(crlf.directory,
`node_modules/@lobehub/icons-static-svg/icons/${icon}`, upstream);
+ write(crlf.directory, `src/assets/model-logos/${icon}`,
upstream.replace(/\n/g, '\r\n'));
+ const rewrapped = collectLicenses(modules, crlf.directory);
+ assert.deepEqual(rewrapped.components, expected.components);
+ assert.deepEqual(rewrapped.files, expected.files);
+});
+
+test('lineEndingToleranceStillRejectsChangedLicenseAndCodeTest', (t) => {
+ const { directory, license } = lineEndingFixture(t, '\r\n');
+ const fallback = 'licenses/toggle-selection-1.0.6/LICENSE';
+ write(directory, fallback, license.replace('Copyright', 'Altered
copyright').replace(/\n/g, '\r\n'));
+ assert.throws(() =>
collectLicenses(['node_modules/toggle-selection/index.js'], directory), /does
not match the verified source/);
+ write(directory, fallback, license);
+ const code = readFileSync(path.join(directory,
'node_modules/toggle-selection/index.js'));
+ write(directory, 'node_modules/toggle-selection/index.js',
Buffer.concat([code, Buffer.from('\n// changed code')]));
+ assert.throws(() =>
collectLicenses(['node_modules/toggle-selection/index.js'], directory), /does
not match the verified source/);
+});
+
test('missingOrUnknownLicenseFailsTest', (t) => {
const directory = temporary(t);
for (const name of ['LICENSE', 'NOTICE']) write(directory, name,
readFileSync(path.join(root, name)));
@@ -65,6 +140,7 @@ test('missingOrUnknownLicenseFailsTest', (t) => {
test('vitePackagingAndTamperGateTest', async (t) => {
const directory = temporary(t);
const licensePlugin = distributionLicenses();
+ const fixturePath = (name) => JSON.stringify(path.join(root,
name).split(path.sep).join('/'));
// Build only fixtures for React, CSS, SVG and small dependencies; do not
build the app or run app tests.
const result = await build({
root,
@@ -75,7 +151,10 @@ test('vitePackagingAndTamperGateTest', async (t) => {
name: 'license-fixture',
resolveId(id) { if (id === 'license-fixture') return
'\0license-fixture'; },
load(id) {
- if (id === '\0license-fixture') return `import React from
'${root}node_modules/react/index.js'; import logo from
'${root}src/assets/model-logos/openai.svg'; import '${root}src/index.css';
import toggle from '${root}node_modules/toggle-selection/index.js';
console.log(React, logo, toggle);`;
+ if (id === '\0license-fixture') return `import React from
${fixturePath('node_modules/react/index.js')};
+ import logo from
${fixturePath('src/assets/model-logos/openai.svg')};
+ import ${fixturePath('src/index.css')};
+ import toggle from
${fixturePath('node_modules/toggle-selection/index.js')}; console.log(React,
logo, toggle);`;
},
},
licensePlugin,
diff --git a/web/src/hooks/useCanWriteSettings.test.tsx
b/web/src/hooks/useCanWriteSettings.test.tsx
new file mode 100644
index 000000000..c33c87d05
--- /dev/null
+++ b/web/src/hooks/useCanWriteSettings.test.tsx
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the 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.
+ * The 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.
+ */
+
+import { renderHook } from '@testing-library/react';
+import { describe, expect, it, vi } from 'vitest';
+import { useCanWriteSettings } from './useCanWriteSettings';
+
+type MockAuthState = { admin: boolean | null; userId: number | null };
+
+// Read inside the selector, not when the factory runs, so the cases below can
swap the session
+// without re-registering the module mock.
+let authState: MockAuthState = { admin: false, userId: 42 };
+
+vi.mock('../stores/authStore', () => ({
+ default: (selector: (state: MockAuthState) => unknown) =>
selector(authState),
+}));
+
+const rendered = () => renderHook(() => useCanWriteSettings()).result.current;
+
+const sessions: Array<[string, MockAuthState, boolean]> = [
+ ['a signed-in reader', { userId: 42, admin: false }, false],
+ ['a signed-in admin', { userId: 42, admin: true }, true],
+ // Auth disabled: nobody is signed in, the caller counts as system and the
server lets the write
+ // through. This row is why the guard reads `!userId ||` first — dropping it
would lock every
+ // deployment that runs without studio authentication.
+ ['a deployment with studio auth disabled', { userId: null, admin: null },
true],
+ ['a signed-in operator whose admin flag never arrived', { userId: 42, admin:
null }, false],
+];
+
+describe('useCanWriteSettings', () => {
+ it.each(sessions)('decides the settings writes for %sTest', (_label, state,
expected) => {
+ authState = state;
+ expect(rendered()).toBe(expected);
+ });
+});
diff --git a/web/src/hooks/useCanWriteSettings.ts
b/web/src/hooks/useCanWriteSettings.ts
new file mode 100644
index 000000000..f2427cdb7
--- /dev/null
+++ b/web/src/hooks/useCanWriteSettings.ts
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the 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.
+ * The 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.
+ */
+
+import useAuthStore from '../stores/authStore';
+
+/**
+ * 设置类写操作的角色闸门：当前登录者能否写设置。
+ *
+ * Mirrors the server policy: the settings writes (POST
/api/settings/general/save, the LLM
+ * configuration POST, and the data source / cloud credential mutations) are
admin-only, and the
+ * server also hands redacted secret values to readers. Without a signed-in
studio user (auth
+ * disabled) the caller counts as system, so writes stay allowed — that
fallback is deliberate:
+ * dropping it would lock out every deployment that runs without studio
authentication.
+ *
+ * 四个设置 tab 共用这一份派生，避免权限口径在复制中分叉。
+ */
+export function useCanWriteSettings(): boolean {
+ const authUserId = useAuthStore((state) => state.userId);
+ const authAdmin = useAuthStore((state) => state.admin);
+ return !authUserId || authAdmin === true;
+}
diff --git a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
index acee44de1..7291f4db5 100644
--- a/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClientsPage.test.tsx
@@ -168,6 +168,29 @@ async function expectInteractive(element: Element) {
}
describe('Clients page', () => {
+ it('shows one producer connected to two brokers without a false ID collision
Test', async () => {
+ vi.mocked(connectionsService.listConnections).mockResolvedValue([
+ {
+ ...connection,
+ clientId: '192.0.2.10@4242',
+ protocol: 'Remoting',
+ address: '192.0.2.10:51001',
+ },
+ {
+ ...connection,
+ clientId: '192.0.2.10@4242',
+ protocol: 'Remoting',
+ address: '192.0.2.10:51002',
+ },
+ ]);
+ renderWithProviders(<ClientsPage />);
+ await screen.findAllByText('192.0.2.10:51001');
+ await screen.findAllByText('192.0.2.10:51002');
+ const diagnostics = screen.getByTestId('client-connection-diagnostics');
+ expect(within(diagnostics).queryByText('Client ID
连接到多个地址')).not.toBeInTheDocument();
+ expect(within(diagnostics).getByText('客户端连接健康')).toBeInTheDocument();
+ });
+
it('returns to the first page when the connection search changes', async ()
=> {
const pagedConnections = Array.from({ length: 21 }, (_, index) => ({
...connection,
@@ -274,6 +297,38 @@ describe('Clients page', () => {
expect(within(languageVersions).getByText('C++ 4.9.8:
1')).toBeInTheDocument();
});
+ it('countsEveryListedConnectionWhenAClientServesSeveralGroupsTest', async ()
=> {
+ // A RocketMQ client registers one entry per group it serves - its own
group plus the shared
+ // CLIENT_INNER_PRODUCER - so the same clientId appears on several rows.
The summary tiles
+ // describe the connections the table lists; the de-duplicated clients
stay the diagnostics
+ // panel's job ("唯一 Client / 唯一地址" below).
+ vi.mocked(connectionsService.listConnections).mockResolvedValue([
+ { ...connection, clientId: 'shared-client', groupOrTopic: 'order-create'
},
+ { ...connection, clientId: 'shared-client', groupOrTopic:
'CLIENT_INNER_PRODUCER' },
+ {
+ ...connection,
+ clientId: 'order-consumer',
+ type: 'Consumer',
+ groupOrTopic: 'cg-order',
+ address: '10.0.1.40:49152',
+ },
+ ]);
+ renderWithProviders(<ClientsPage />);
+
+ await waitFor(() => {
+
expect(within(screen.getByTestId('connection-total')).getByText('3')).toBeInTheDocument();
+ });
+
expect(within(screen.getByTestId('producer-total')).getByText('2')).toBeInTheDocument();
+
expect(within(screen.getByTestId('consumer-total')).getByText('1')).toBeInTheDocument();
+ expect(
+ within(screen.getByTestId('protocol-distribution')).getByText('gRPC: 3'),
+ ).toBeInTheDocument();
+
+ const diagnostics = screen.getByTestId('client-connection-diagnostics');
+ expect(diagnostics.textContent).toContain('唯一 Client: 2');
+ expect(diagnostics.textContent).toContain('唯一地址: 2');
+ });
+
it('renders client connection diagnostics for risky inventories', async ()
=> {
vi.mocked(connectionsService.listConnections).mockResolvedValue([
{
diff --git a/web/src/pages/cluster/clients.tsx
b/web/src/pages/cluster/clients.tsx
index 73aa5139f..9740332e0 100644
--- a/web/src/pages/cluster/clients.tsx
+++ b/web/src/pages/cluster/clients.tsx
@@ -292,21 +292,17 @@ const ClientsPage = () => {
const clusterConnections = useMemo(() => connections, [connections]);
const connectionStats = useMemo(() => {
- const instances = Array.from(
- new Map(
- clusterConnections.map((connection) => [
- `${connection.type}:${connection.clientId ?? connection.address ??
connection.groupOrTopic}`,
- connection,
- ]),
- ).values(),
- );
+ // Every tile describes the connections the table below lists: the page
header and the table
+ // footer both count these rows, and the diagnostics panel already reports
the de-duplicated
+ // view (unique clients / addresses). Collapsing a client that serves
several groups into one
+ // entry here made the summary disagree with both, e.g. 5 in the header of
a 7-row page.
return {
- total: instances.length,
- producers: instances.filter((connection) => connection.type ===
'Producer').length,
- consumers: instances.filter((connection) => connection.type ===
'Consumer').length,
- protocols: countBy(instances.map((connection) => connection.protocol)),
+ total: clusterConnections.length,
+ producers: clusterConnections.filter((connection) => connection.type ===
'Producer').length,
+ consumers: clusterConnections.filter((connection) => connection.type ===
'Consumer').length,
+ protocols: countBy(clusterConnections.map((connection) =>
connection.protocol)),
languageVersions: countBy(
- instances.map(
+ clusterConnections.map(
(connection) =>
`${connection.language ?? UNKNOWN_LANGUAGE} ${connection.version
?? UNKNOWN_LANGUAGE}`,
),
diff --git a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
index 312796132..89f671747 100644
--- a/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
+++ b/web/src/pages/instance/__tests__/ConsumerPage.test.tsx
@@ -411,6 +411,44 @@ describe('Consumer page', () => {
errorSpy.mockRestore();
});
+ it.each(['empty', 'unchanged', 'failure'])(
+ 'keepsBatchSelectionBoundToConfirmedConsumerInventoryTest: %s',
+ async (outcome) => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ let finishRefresh!: (value: ReturnType<typeof groupPage>) => void;
+ let failRefresh!: (error: Error) => void;
+ const refreshedPage = new Promise<ReturnType<typeof
groupPage>>((resolve, reject) => {
+ finishRefresh = resolve;
+ failRefresh = reject;
+ });
+ vi.mocked(consumerService.listConsumerGroupPage)
+ .mockResolvedValueOnce(groupPage([group]))
+ .mockReturnValue(refreshedPage);
+ renderWithProviders(<ConsumerPage />);
+ const row = await screen.findByRole('row', { name: /remote-cg/ });
+ await user.click(within(row).getByRole('checkbox'));
+ expect(screen.getByRole('button', { name: /删除 \(1\)$/
})).toBeInTheDocument();
+
+ // Enabling the ordinary list refresh immediately reads the same
instance/query.
+ await user.click(screen.getByRole('button', { name: /自动刷新/ }));
+ await waitFor(() =>
expect(consumerService.listConsumerGroupPage).toHaveBeenCalledTimes(2));
+ await act(async () => {
+ if (outcome === 'failure') failRefresh(new Error('list read failed'));
+ else finishRefresh(groupPage(outcome === 'empty' ? [] : [group]));
+ });
+ expect(consumerService.deleteConsumerGroup).not.toHaveBeenCalled();
+ expect(consumerService.batchDeleteConsumerGroups).not.toHaveBeenCalled();
+ if (outcome === 'empty') {
+ expect(screen.queryByText('remote-cg')).not.toBeInTheDocument();
+ expect(screen.queryByRole('button', { name: /删除 \(1\)$/
})).not.toBeInTheDocument();
+ } else {
+ const retained = screen.getByText('remote-cg').closest('tr') as
HTMLElement;
+ expect(within(retained).getByRole('checkbox')).toBeChecked();
+ expect(screen.getByRole('button', { name: /删除 \(1\)$/
})).toBeInTheDocument();
+ }
+ },
+ );
+
afterEach(async () => {
await act(async () => {
cleanup();
diff --git a/web/src/pages/instance/consumer.tsx
b/web/src/pages/instance/consumer.tsx
index bede02a48..1a02b099a 100644
--- a/web/src/pages/instance/consumer.tsx
+++ b/web/src/pages/instance/consumer.tsx
@@ -368,6 +368,8 @@ const ConsumerPageContent = ({
if (requestId === groupRequestIdRef.current) {
setGroups(result.items);
setTotalGroups(result.total);
+ // A current successful empty result confirms there are no targets
in this query.
+ if (result.total === 0) setSelectedRowKeys([]);
if (result.items.length === 0 && result.total > 0 && pageToLoad > 1)
{
setPage(Math.max(1, Math.ceil(result.total / pageSizeToLoad)));
}
diff --git a/web/src/pages/ops/__tests__/AlertsTestResultResidue.test.tsx
b/web/src/pages/ops/__tests__/AlertsTestResultResidue.test.tsx
index 67bb06ae0..e93e8c2da 100644
--- a/web/src/pages/ops/__tests__/AlertsTestResultResidue.test.tsx
+++ b/web/src/pages/ops/__tests__/AlertsTestResultResidue.test.tsx
@@ -16,10 +16,10 @@
*/
import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
-import { fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
+import { act, fireEvent, render, screen, waitFor, within } from
'@testing-library/react';
import userEvent from '@testing-library/user-event';
import { App } from 'antd';
-import type { AlertRule, PageResult } from '../../../api/ops';
+import type { AlertRule, AlertRuleDomain, AlertRuleTestResult, PageResult }
from '../../../api/ops';
import { LangProvider } from '../../../i18n/LangContext';
import { listInstances } from '../../../services/instanceService';
import {
@@ -88,11 +88,11 @@ const pageResult = (rules: AlertRule[]):
PageResult<AlertRule> => ({
size: 20,
});
-const renderPage = () =>
+const renderPage = (domain?: AlertRuleDomain) =>
render(
<App>
<LangProvider>
- <AlertsPage />
+ <AlertsPage domain={domain} />
</LangProvider>
</App>,
);
@@ -103,6 +103,49 @@ function getRuleRow(ruleName: string) {
return row;
}
+function mockBusinessLagRule() {
+ const lagRule: AlertRule = {
+ ...buildRule(3, 'Lag guard'),
+ metric: 'consumer.lag.total',
+ threshold: 1000,
+ thresholdUnit: 'messages',
+ channels: ['email'],
+ consumerGroup: 'orders',
+ aggregation: 'LAST',
+ windowSeconds: 0,
+ consecutiveSamples: 1,
+ };
+ vi.mocked(listAlertRulesPage).mockResolvedValue(pageResult([lagRule]));
+ vi.mocked(listNativeAlertMetrics).mockResolvedValue([
+ {
+ key: 'consumer.lag.total',
+ label: 'Consumer lag total',
+ thresholdUnit: 'messages',
+ supportsConsumerGroup: true,
+ },
+ ]);
+}
+
+function lagTestResult(conditionMet = true): AlertRuleTestResult {
+ return {
+ samples: [
+ {
+ labels: { consumerGroup: 'orders' },
+ availability: 'AVAILABLE',
+ currentValue: 1200,
+ conditionMet,
+ },
+ ],
+ };
+}
+
+async function openLagEditor(user: ReturnType<typeof userEvent.setup>) {
+ renderPage('BUSINESS');
+ await screen.findByText('Lag guard');
+ await user.click(within(getRuleRow('Lag guard')).getByRole('button', { name:
'编辑' }));
+ return screen.findByRole('dialog');
+}
+
describe('AlertsPage rule test results', () => {
beforeEach(() => {
vi.clearAllMocks();
@@ -172,4 +215,102 @@ describe('AlertsPage rule test results', () => {
expect(within(reopened).queryByText('规则试运行结果')).toBeNull();
expect(within(reopened).queryByText('broker=stale-broker-sample')).toBeNull();
}, 30_000);
+
+ it.each(['completed', 'pending', 'unchanged', 'metadata'])(
+ 'keepsTestResultsBoundToTheCurrentThresholdTest: %s',
+ async (completion) => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ mockBusinessLagRule();
+ const samples = lagTestResult();
+ let finishTest!: (value: AlertRuleTestResult) => void;
+ const pendingTest = new Promise<AlertRuleTestResult>((resolve) => {
+ finishTest = resolve;
+ });
+ vi.mocked(testAlertRule).mockReset();
+ vi.mocked(testAlertRule).mockReturnValue(pendingTest);
+ const editor = await openLagEditor(user);
+ await user.click(within(editor).getByRole('button', { name: '试运行' }));
+ await waitFor(() =>
+ expect(testAlertRule).toHaveBeenCalledWith(
+ expect.objectContaining({ metric: 'consumer.lag.total', threshold:
1000 }),
+ 'BUSINESS',
+ ),
+ );
+ if (completion !== 'pending') {
+ await act(async () => finishTest(samples));
+ expect(await within(editor).findByText('规则试运行结果')).toBeInTheDocument();
+ expect(within(editor).getByText('命中')).toBeInTheDocument();
+ }
+ if (completion === 'completed' || completion === 'pending') {
+ const threshold = within(editor).getByPlaceholderText('阈值');
+ await user.clear(threshold);
+ await user.type(threshold, '2000');
+ expect(threshold).toHaveValue('2000');
+ }
+ if (completion === 'metadata') {
+ await user.type(within(editor).getByRole('textbox', { name: '规则名称' }),
' updated');
+ }
+ if (completion === 'pending') {
+ await act(async () => finishTest(samples));
+ }
+ expect(testAlertRule).toHaveBeenCalledTimes(1);
+ if (completion === 'unchanged' || completion === 'metadata') {
+ expect(within(editor).getByText('规则试运行结果')).toBeInTheDocument();
+ expect(within(editor).getByText('命中')).toBeInTheDocument();
+ } else {
+ expect(within(editor).queryByText('规则试运行结果')).not.toBeInTheDocument();
+ expect(within(editor).queryByText('命中')).not.toBeInTheDocument();
+ }
+ },
+ );
+
+ it.each(['success', 'failure'])(
+ 'keepsTheNewCriteriaTestActiveWhenAnOldRequestSettlesTest: %s',
+ async (outcome) => {
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ mockBusinessLagRule();
+ let finishEarlier!: (value: AlertRuleTestResult) => void;
+ let failEarlier!: (error: Error) => void;
+ const earlier = new Promise<AlertRuleTestResult>((resolve, reject) => {
+ finishEarlier = resolve;
+ failEarlier = reject;
+ });
+ let finishCurrent!: (value: AlertRuleTestResult) => void;
+ const current = new Promise<AlertRuleTestResult>((resolve) => {
+ finishCurrent = resolve;
+ });
+ vi.mocked(testAlertRule).mockReset();
+
vi.mocked(testAlertRule).mockReturnValueOnce(earlier).mockReturnValueOnce(current);
+ const editor = await openLagEditor(user);
+ const testButton = within(editor).getByRole('button', { name: '试运行' });
+ await user.click(testButton);
+ await waitFor(() => expect(testAlertRule).toHaveBeenCalledTimes(1));
+ const threshold = within(editor).getByPlaceholderText('阈值');
+ await user.clear(threshold);
+ await user.type(threshold, '2000');
+ await user.click(testButton);
+ await waitFor(() =>
+ expect(testAlertRule).toHaveBeenNthCalledWith(
+ 2,
+ expect.objectContaining({ threshold: 2000 }),
+ 'BUSINESS',
+ ),
+ );
+ await act(async () => {
+ if (outcome === 'success') finishEarlier(lagTestResult());
+ else failEarlier(new Error('old criteria test failed'));
+ });
+ expect(testButton).toHaveClass('ant-btn-loading');
+ expect(within(editor).queryByText('规则试运行结果')).not.toBeInTheDocument();
+ expect(screen.queryByText(/规则试运行失败/)).not.toBeInTheDocument();
+ expect(screen.queryByText(/1 个命中：1200/)).not.toBeInTheDocument();
+ await act(async () => finishCurrent(lagTestResult(false)));
+ expect(await within(editor).findByText('规则试运行结果')).toBeInTheDocument();
+ expect(within(editor).getByText('未命中')).toBeInTheDocument();
+ expect(testButton).not.toHaveClass('ant-btn-loading');
+ },
+ // Opening the editor, running two tests and retyping the threshold is a
long chain of
+ // userEvent interactions; the case above needs the same budget for the
same reason.
+ 30_000,
+ );
});
diff --git a/web/src/pages/ops/alerts.tsx b/web/src/pages/ops/alerts.tsx
index 266474041..6b56ebdb7 100644
--- a/web/src/pages/ops/alerts.tsx
+++ b/web/src/pages/ops/alerts.tsx
@@ -134,6 +134,21 @@ const aggregationLabelKeys: Record<string, string> = {
SUM: 'alerts.aggregation.SUM',
};
const notificationTemplateVariables = ALERT_NOTIFICATION_TEMPLATE_VARIABLES;
+const testCriteriaFields = new Set([
+ 'instanceId',
+ 'metric',
+ 'operator',
+ 'threshold',
+ 'thresholdUnit',
+ 'consumerGroup',
+ 'topic',
+ 'clusterName',
+ 'brokerName',
+ 'duration',
+ 'aggregation',
+ 'windowSeconds',
+ 'consecutiveSamples',
+]);
const availabilityMetrics = new Set([
'nameserver.availability',
'broker.availability',
@@ -245,6 +260,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
const [instances, setInstances] = useState<Instance[]>([]);
const [transferringRules, setTransferringRules] = useState(false);
const metricRequestVersion = useRef(0);
+ const testRequestVersion = useRef(0);
const importInputRef = useRef<HTMLInputElement>(null);
const notificationTemplateRef = useRef<TextAreaRef>(null);
const supportsUnavailableCondition =
supportsUnavailableOperator(selectedMetric);
@@ -468,7 +484,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
setSelectedInstanceId(undefined);
setMetricOptions([]);
setMetricLoading(false);
- setTestResult(null);
+ invalidateTestResult();
form.resetFields();
setModalVisible(true);
};
@@ -510,6 +526,12 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
}
};
+ const invalidateTestResult = () => {
+ testRequestVersion.current += 1;
+ setTestResult(null);
+ setTesting(false);
+ };
+
const loadMetricCapabilities = async (instanceId?: string, resetMetric =
true) => {
if (!instanceId?.trim()) return;
const requestVersion = ++metricRequestVersion.current;
@@ -519,7 +541,12 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
const metrics = await listNativeAlertMetrics(instanceId.trim(), domain);
if (requestVersion !== metricRequestVersion.current) return;
setMetricOptions(metrics);
- if (resetMetric) form.setFieldValue('metric', undefined);
+ if (resetMetric) {
+ // Programmatic setFieldValue does not fire onValuesChange, so
clearing the metric here
+ // has to invalidate the stale run itself.
+ if (form.getFieldValue('metric') !== undefined) invalidateTestResult();
+ form.setFieldValue('metric', undefined);
+ }
if (metrics.length === 0) message.warning(t('alerts.metricUnavailable'));
} catch {
if (requestVersion === metricRequestVersion.current) {
@@ -532,7 +559,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
const openEditModal = (rule: AlertRule) => {
setEditingRule(rule);
- setTestResult(null);
+ invalidateTestResult();
form.setFieldsValue({
...rule,
metric: normalizeMetric(rule.metric),
@@ -551,7 +578,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
const openDuplicateModal = (rule: AlertRule) => {
setEditingRule(null);
- setTestResult(null);
+ invalidateTestResult();
setSelectedInstanceId(rule.instanceId);
const { id: _id, lastTriggered: _lastTriggered, ...copy } = rule;
form.setFieldsValue({
@@ -889,8 +916,10 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
};
const handleTest = async () => {
+ const requestVersion = ++testRequestVersion.current;
try {
const values = await form.validateFields();
+ if (requestVersion !== testRequestVersion.current) return;
const payload = {
...attachThresholdUnit(values),
...(nativeRatioMetrics.has(normalizeMetric(values.metric)) ? {
thresholdUnit: '%' } : {}),
@@ -899,6 +928,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
const result = await (domain === 'CLUSTER'
? testAlertRule(payload)
: testAlertRule(payload, domain));
+ if (requestVersion !== testRequestVersion.current) return;
setTestResult(result);
if (result.samples.length === 0) {
message.warning(t('alerts.testNoSamples'));
@@ -936,10 +966,11 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
}),
);
} catch (error) {
+ if (requestVersion !== testRequestVersion.current) return;
if (error && typeof error === 'object' && 'errorFields' in error) return;
message.error(t('alerts.testRunFailed'));
} finally {
- setTesting(false);
+ if (requestVersion === testRequestVersion.current) setTesting(false);
}
};
@@ -1125,7 +1156,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
metricRequestVersion.current += 1;
setModalVisible(false);
setEditingRule(null);
- setTestResult(null);
+ invalidateTestResult();
setSelectedInstanceId(undefined);
setMetricOptions([]);
setMetricLoading(false);
@@ -1143,7 +1174,7 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
metricRequestVersion.current += 1;
setModalVisible(false);
setEditingRule(null);
- setTestResult(null);
+ invalidateTestResult();
setSelectedInstanceId(undefined);
setMetricOptions([]);
setMetricLoading(false);
@@ -1158,7 +1189,15 @@ const AlertsPageContent = ({ domain = 'CLUSTER' }:
AlertsPageProps) => {
</Flex>
}
>
- <Form form={form} layout="vertical">
+ <Form
+ form={form}
+ layout="vertical"
+ onValuesChange={(changedValues) => {
+ if (Object.keys(changedValues).some((field) =>
testCriteriaFields.has(field))) {
+ invalidateTestResult();
+ }
+ }}
+ >
<div
style={{
display: 'grid',
diff --git a/web/src/pages/settings/AiAssistantTab.tsx
b/web/src/pages/settings/AiAssistantTab.tsx
index d4bcd589a..abb524862 100644
--- a/web/src/pages/settings/AiAssistantTab.tsx
+++ b/web/src/pages/settings/AiAssistantTab.tsx
@@ -45,6 +45,7 @@ import {
} from '../../api/llm';
import { fallbackModelOptions } from '../studio/llmModelOptions';
import { describeThrownMessage } from '../../utils/apiError';
+import { useCanWriteSettings } from '../../hooks/useCanWriteSettings';
const DEFAULT_BASE_URL: Record<string, string> = {
tongyi: 'https://dashscope.aliyuncs.com/compatible-mode/v1',
@@ -108,6 +109,11 @@ interface TestState {
export const AiAssistantTab = () => {
const { t } = useLang();
const { message } = App.useApp();
+ // The LLM configuration surface is admin-only on both ends (the GET is an
admin-only read, the
+ // POST falls through to the admin requirement), so a reader's save, test or
key clear could only
+ // ever come back 403. The controls stay disabled instead, the same way the
other settings tabs
+ // treat the writes they own.
+ const canWriteSettings = useCanWriteSettings();
const [form] = Form.useForm();
const selectedProvider = Form.useWatch('provider', form);
@@ -239,6 +245,7 @@ export const AiAssistantTab = () => {
};
const handleClearApiKey = async () => {
+ if (!canWriteSettings) return;
const payload = await buildPayload();
if (!payload) return;
const confirmed = window.confirm(t('settings.clearApiKeyConfirm'));
@@ -277,6 +284,7 @@ export const AiAssistantTab = () => {
};
const handleTest = async () => {
+ if (!canWriteSettings) return;
const payload = await buildPayload();
if (!payload) return;
const requestId = testRequestIdRef.current + 1;
@@ -304,6 +312,7 @@ export const AiAssistantTab = () => {
};
const handleSave = async () => {
+ if (!canWriteSettings) return;
const payload = await buildPayload();
if (!payload) return;
const configGeneration = configGenerationRef.current;
@@ -444,6 +453,7 @@ export const AiAssistantTab = () => {
size="small"
style={{ marginLeft: 8 }}
loading={saving}
+ disabled={!canWriteSettings}
onClick={() => void handleClearApiKey()}
>
{t('settings.clearApiKey')}
@@ -554,12 +564,16 @@ export const AiAssistantTab = () => {
<Button
type="primary"
loading={saving}
- disabled={loading}
+ disabled={loading || !canWriteSettings}
onClick={() => void handleSave()}
>
{t('common.save')}
</Button>
- <Button loading={testing} disabled={loading} onClick={() => void
handleTest()}>
+ <Button
+ loading={testing}
+ disabled={loading || !canWriteSettings}
+ onClick={() => void handleTest()}
+ >
{t('settings.testConnection')}
</Button>
</Space>
diff --git a/web/src/pages/settings/CloudCredentialTab.tsx
b/web/src/pages/settings/CloudCredentialTab.tsx
index a404bbe67..69c379187 100644
--- a/web/src/pages/settings/CloudCredentialTab.tsx
+++ b/web/src/pages/settings/CloudCredentialTab.tsx
@@ -35,6 +35,7 @@ import { DeleteOutlined, DownloadOutlined, EditOutlined,
PlusOutlined } from '@a
import type { ColumnsType } from 'antd/es/table';
import { MagnifyingGlass } from '@phosphor-icons/react';
import { useLang } from '../../i18n/LangContext';
+import { useCanWriteSettings } from '../../hooks/useCanWriteSettings';
import {
createCloudCredential,
@@ -65,6 +66,10 @@ interface CredentialFormValues {
export const CloudCredentialTab = () => {
const { t } = useLang();
+ // Credential writes are admin-only (POST/PUT/DELETE fall through to the
admin requirement in
+ // AuthInterceptor), so a reader's save or delete could only ever come back
403. The controls stay
+ // disabled instead, the same way the other settings tabs treat the writes
they own.
+ const canWriteSettings = useCanWriteSettings();
const [credentials, setCredentials] = useState<CloudCredential[]>([]);
const [total, setTotal] = useState(0);
const [page, setPage] = useState(1);
@@ -194,6 +199,7 @@ export const CloudCredentialTab = () => {
};
const handleSubmit = async () => {
+ if (!canWriteSettings) return;
if (submitInFlightRef.current) return;
submitInFlightRef.current = true;
setSubmitting(true);
@@ -233,6 +239,7 @@ export const CloudCredentialTab = () => {
};
const handleDelete = async (credential: CloudCredential) => {
+ if (!canWriteSettings) return;
if (loading || loadFailed || search.trim() !== debouncedSearch ||
loadedQueryKey !== queryKey)
return;
try {
@@ -277,7 +284,7 @@ export const CloudCredentialTab = () => {
type="link"
size="small"
icon={<EditOutlined />}
- disabled={staleActionsDisabled}
+ disabled={staleActionsDisabled || !canWriteSettings}
onClick={() => openEditModal(record)}
>
{t('common.edit')}
@@ -293,7 +300,7 @@ export const CloudCredentialTab = () => {
size="small"
danger
icon={<DeleteOutlined />}
- disabled={staleActionsDisabled}
+ disabled={staleActionsDisabled || !canWriteSettings}
>
{t('common.delete')}
</Button>
@@ -339,7 +346,9 @@ export const CloudCredentialTab = () => {
type="primary"
icon={<PlusOutlined />}
onClick={openCreateModal}
- disabled={loading || filtersPending || (queryMismatch &&
!loadFailed)}
+ disabled={
+ loading || filtersPending || (queryMismatch && !loadFailed) ||
!canWriteSettings
+ }
>
{t('settings.addCredential')}
</Button>
@@ -385,6 +394,7 @@ export const CloudCredentialTab = () => {
open={modalOpen}
onCancel={closeModal}
onOk={() => void handleSubmit()}
+ okButtonProps={{ disabled: !canWriteSettings }}
confirmLoading={submitting}
closable={!submitting}
maskClosable={!submitting}
diff --git a/web/src/pages/settings/DataSourceTab.tsx
b/web/src/pages/settings/DataSourceTab.tsx
index 1c2755a26..5d1e45466 100644
--- a/web/src/pages/settings/DataSourceTab.tsx
+++ b/web/src/pages/settings/DataSourceTab.tsx
@@ -35,6 +35,7 @@ import { ApiOutlined, DeleteOutlined, EditOutlined,
PlusOutlined } from '@ant-de
import type { ColumnsType } from 'antd/es/table';
import { useLang } from '../../i18n/LangContext';
+import { useCanWriteSettings } from '../../hooks/useCanWriteSettings';
import StatusBadge from '../../components/StatusBadge';
import {
createDataSource,
@@ -117,6 +118,10 @@ const withoutSecrets = (values: DataSourceFormValues):
Partial<DataSource> => {
export const DataSourceTab = () => {
const { t } = useLang();
+ // Data source writes are admin-only (POST/PUT/DELETE fall through to the
admin requirement in
+ // AuthInterceptor), so a reader's test, save or delete could only ever come
back 403. The controls
+ // stay disabled instead, the same way the other settings tabs treat the
writes they own.
+ const canWriteSettings = useCanWriteSettings();
const [dataSources, setDataSources] = useState<DataSource[]>([]);
const [total, setTotal] = useState(0);
const [page, setPage] = useState(1);
@@ -225,6 +230,7 @@ export const DataSourceTab = () => {
const isCurrent = () => key !== 'modal' || modalSession ===
modalTestSessionRef.current;
if (!isCurrent()) return;
if (key !== 'modal' && listActionsDisabled) return;
+ if (!canWriteSettings) return;
if (key !== 'modal' && authNeedsSecret(data.auth)) {
message.warning(t('settings.dataSourceAuthTestHint'));
return;
@@ -282,6 +288,10 @@ export const DataSourceTab = () => {
};
const handleSubmit = async () => {
+ // The role guard runs before the re-entrancy lock: a reader who reaches
this handler (keyboard
+ // or assistive paths bypass `disabled`) must not take the lock and never
release it, wedging
+ // the dialog for admins too. CloudCredentialTab orders its guards the
same way.
+ if (!canWriteSettings) return;
if (submittingRef.current) return;
submittingRef.current = true;
setSubmitting(true);
@@ -318,6 +328,7 @@ export const DataSourceTab = () => {
const handleDelete = async (dataSource: DataSource) => {
if (listActionsDisabled) return;
+ if (!canWriteSettings) return;
try {
await deleteDataSource(dataSource.key);
// Refresh through the effect so a completed deletion uses the current
filters/page,
@@ -334,8 +345,9 @@ export const DataSourceTab = () => {
return instanceIds
.map(
(instanceId) =>
- instances.find(
- (instance) => instance.name === instanceId || String(instance.id)
=== instanceId,
+ (
+ instances.find((instance) => instance.name === instanceId) ??
+ instances.find((instance) => String(instance.id) === instanceId)
)?.name ?? instanceId,
)
.join('、');
@@ -406,7 +418,7 @@ export const DataSourceTab = () => {
size="small"
icon={<ApiOutlined />}
loading={testingKeys.has(record.key)}
- disabled={listActionsDisabled || authNeedsSecret(record.auth)}
+ disabled={listActionsDisabled || authNeedsSecret(record.auth) ||
!canWriteSettings}
title={authNeedsSecret(record.auth) ?
t('settings.dataSourceAuthTestHint') : undefined}
onClick={() => void handleTestConnection(record, record.key)}
>
@@ -416,7 +428,7 @@ export const DataSourceTab = () => {
type="link"
size="small"
icon={<EditOutlined />}
- disabled={listActionsDisabled}
+ disabled={listActionsDisabled || !canWriteSettings}
onClick={() => openEditModal(record)}
>
{t('common.edit')}
@@ -432,7 +444,7 @@ export const DataSourceTab = () => {
size="small"
danger
icon={<DeleteOutlined />}
- disabled={listActionsDisabled}
+ disabled={listActionsDisabled || !canWriteSettings}
>
{t('common.delete')}
</Button>
@@ -482,7 +494,7 @@ export const DataSourceTab = () => {
type="primary"
icon={<PlusOutlined />}
onClick={openCreateModal}
- disabled={loading}
+ disabled={loading || !canWriteSettings}
>
{t('settings.addDataSource')}
</Button>
@@ -518,6 +530,7 @@ export const DataSourceTab = () => {
open={modalOpen}
onCancel={closeModal}
onOk={() => void handleSubmit()}
+ okButtonProps={{ disabled: !canWriteSettings }}
confirmLoading={submitting}
destroyOnHidden
>
diff --git a/web/src/pages/settings/GeneralSettingsTab.tsx
b/web/src/pages/settings/GeneralSettingsTab.tsx
index 66cac55ff..f7f8d9d52 100644
--- a/web/src/pages/settings/GeneralSettingsTab.tsx
+++ b/web/src/pages/settings/GeneralSettingsTab.tsx
@@ -35,7 +35,7 @@ import type { GeneralSettings, GeneralSettingsUpdate } from
'../../api/settings'
import { useTheme } from '../../theme/useTheme';
import type { ThemeMode } from '../../theme/themePreference';
import { useLang } from '../../i18n/LangContext';
-import useAuthStore from '../../stores/authStore';
+import { useCanWriteSettings } from '../../hooks/useCanWriteSettings';
const { Text } = Typography;
@@ -71,12 +71,10 @@ export const GeneralSettingsTab = () => {
const [testingChannel, setTestingChannel] = useState<string>();
const securityInFlightRef = useRef(false);
const notifyInFlightRef = useRef(false);
- const authUserId = useAuthStore((state) => state.userId);
- const authAdmin = useAuthStore((state) => state.admin);
- // Mirrors the server policy: POST /api/settings/general/save is admin-only,
and the
- // server also hands redacted webhook values to readers. Without a signed-in
studio
- // user (auth disabled) the caller counts as system, so writes stay allowed.
- const canWriteSettings = !authUserId || authAdmin === true;
+ // Mirrors the server policy: POST /api/settings/general/save is admin-only,
and the server also
+ // hands redacted webhook values to readers. The shared hook documents why a
deployment without
+ // studio authentication keeps its writes enabled.
+ const canWriteSettings = useCanWriteSettings();
const translationRef = useRef(t);
const [securityForm] = Form.useForm();
const [notifyForm] = Form.useForm();
diff --git a/web/src/pages/settings/__tests__/DataSourceTab.test.tsx
b/web/src/pages/settings/__tests__/DataSourceTab.test.tsx
index 599a32f9d..281d4c029 100644
--- a/web/src/pages/settings/__tests__/DataSourceTab.test.tsx
+++ b/web/src/pages/settings/__tests__/DataSourceTab.test.tsx
@@ -20,6 +20,7 @@ import { act, fireEvent, render, screen, waitFor, within }
from '@testing-librar
import userEvent from '@testing-library/user-event';
import { App, ConfigProvider, message } from 'antd';
import type { DataSource, DataSourcePage } from '../../../api/settings';
+import type { Instance } from '../../../api/instance';
import {
createDataSource,
deleteDataSource,
@@ -32,6 +33,9 @@ import { LangProvider } from '../../../i18n/LangContext';
import { LANGUAGE_STORAGE_KEY } from '../../../i18n/languagePreference';
import { downloadCsv } from '../../../utils/download';
import { DataSourceTab } from '../DataSourceTab';
+import { listInstances } from '../../../services/instanceService';
+
+vi.mock('../../../services/instanceService', () => ({ listInstances: vi.fn()
}));
vi.mock('../../../api/settings', () => ({
createDataSource: vi.fn(),
@@ -78,6 +82,45 @@ const sourcePage: DataSourcePage = {
size: 20,
};
+const bindingInstances: Instance[] = [
+ {
+ id: 1,
+ name: 'apache-main',
+ vendor: 'APACHE',
+ type: 'DIRECT',
+ endpoint: 'nameserver.example:9876',
+ remark: null,
+ topicCount: 0,
+ consumerGroupCount: 0,
+ gmtCreate: '2026-10-11T00:00:00',
+ gmtModified: '2026-10-11T00:00:00',
+ },
+ {
+ id: 7,
+ name: 'legacy-instance',
+ vendor: 'APACHE',
+ type: 'DIRECT',
+ endpoint: 'legacy.example:9876',
+ remark: null,
+ topicCount: 0,
+ consumerGroupCount: 0,
+ gmtCreate: '2026-10-11T00:00:00',
+ gmtModified: '2026-10-11T00:00:00',
+ },
+ {
+ id: 2,
+ name: '1',
+ vendor: 'ALIYUN',
+ type: 'CLOUD',
+ endpoint: 'cloud.example:8080',
+ remark: null,
+ topicCount: 0,
+ consumerGroupCount: 0,
+ gmtCreate: '2026-10-11T00:00:00',
+ gmtModified: '2026-10-11T00:00:00',
+ },
+];
+
const deferred = <T,>() => {
let resolve!: (value: T) => void;
let reject!: (reason?: unknown) => void;
@@ -114,6 +157,84 @@ describe('DataSourceTab', () => {
localStorage.removeItem(LANGUAGE_STORAGE_KEY);
vi.mocked(listDataSourcesPage).mockResolvedValue(sourcePage);
vi.mocked(listAllDataSources).mockResolvedValue(sources);
+ vi.mocked(listInstances).mockResolvedValue([]);
+ });
+
+ it('keeps a numeric instance name in the table and edit selection Test',
async () => {
+ const boundSource = { ...sources[0], instanceIds: ['1'] };
+ vi.mocked(listInstances).mockResolvedValue(bindingInstances);
+ vi.mocked(listDataSourcesPage).mockResolvedValue({
+ ...sourcePage,
+ items: [boundSource],
+ total: 1,
+ });
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ render(
+ <App>
+ <DataSourceTab />
+ </App>,
+ );
+
+ const row = await screen.findByRole('row', { name: /Prometheus prod/ });
+ await user.click(within(row).getByRole('button', { name: /编\s*辑/ }));
+ const dialog = await screen.findByRole('dialog');
+ expect(
+ within(dialog).getByText('1', { selector:
'.ant-select-selection-item-content' }),
+ ).toBeInTheDocument();
+ expect(within(row).getAllByRole('cell')[3]).toHaveTextContent(/^1$/);
+ });
+
+ it('exports the numeric instance name instead of another instance with that
ID Test', async () => {
+ const boundSource = { ...sources[0], instanceIds: ['1'] };
+ vi.mocked(listInstances).mockResolvedValue(bindingInstances);
+ vi.mocked(listDataSourcesPage).mockResolvedValue({
+ ...sourcePage,
+ items: [boundSource],
+ total: 1,
+ });
+ vi.mocked(listAllDataSources).mockResolvedValue([boundSource]);
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ render(
+ <App>
+ <DataSourceTab />
+ </App>,
+ );
+ await screen.findByText('Prometheus prod');
+ await user.click(screen.getByRole('button', { name: /编\s*辑/ }));
+ const dialog = await screen.findByRole('dialog');
+ await within(dialog).findByText('1', { selector:
'.ant-select-selection-item-content' });
+ await user.click(within(dialog).getByRole('button', { name: /Cancel|取\s*消/
}));
+ await user.click(screen.getByRole('button', { name: '导出' }));
+
+ await waitFor(() => expect(downloadCsv).toHaveBeenCalled());
+ const csv = vi.mocked(downloadCsv).mock.calls[0][1];
+ expect(csv).toContain('"Applicable Instances"');
+ expect(csv).toContain('"http://prometheus:9090","1","None";');
+ expect(csv).not.toContain('apache-main');
+ });
+
+ it('keeps legacy numeric bindings, unknown names, and global sources
readable Test', async () => {
+ vi.mocked(listInstances).mockResolvedValue(bindingInstances);
+ const boundSource = { ...sources[0], instanceIds: ['7',
'missing-instance'] };
+ vi.mocked(listDataSourcesPage).mockResolvedValue({
+ ...sourcePage,
+ items: [boundSource, sources[1]],
+ });
+ const user = userEvent.setup({ pointerEventsCheck: 0 });
+ render(
+ <App>
+ <DataSourceTab />
+ </App>,
+ );
+
+ const row = await screen.findByRole('row', { name: /Prometheus prod/ });
+ await user.click(within(row).getByRole('button', { name: /编\s*辑/ }));
+ await screen.findByRole('dialog');
+ expect(within(row).getAllByRole('cell')[3]).toHaveTextContent(
+ 'legacy-instance、missing-instance',
+ );
+ const globalRow = screen.getByRole('row', { name: /Thanos DR/ });
+ expect(within(globalRow).getAllByRole('cell')[3]).toHaveTextContent('全局');
});
it.each([false, true])(
diff --git a/web/src/pages/settings/__tests__/SettingsWriteTabsReader.test.tsx
b/web/src/pages/settings/__tests__/SettingsWriteTabsReader.test.tsx
new file mode 100644
index 000000000..b78783d73
--- /dev/null
+++ b/web/src/pages/settings/__tests__/SettingsWriteTabsReader.test.tsx
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the 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.
+ * The 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.
+ */
+
+import type { ReactElement } from 'react';
+import { beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { render, screen } from '@testing-library/react';
+import { App } from 'antd';
+import { getLlmConfig, getLlmModels } from '../../../api/llm';
+import { listDataSourcesPage } from '../../../api/settings';
+import { listCloudCredentials } from '../../../api/cloudCredential';
+import { ThemeProvider } from '../../../theme/ThemeProvider';
+import { LangProvider } from '../../../i18n/LangContext';
+import { LANGUAGE_STORAGE_KEY } from '../../../i18n/languagePreference';
+import { AiAssistantTab } from '../AiAssistantTab';
+import { CloudCredentialTab } from '../CloudCredentialTab';
+import { DataSourceTab } from '../DataSourceTab';
+
+type MockAuthState = { admin: boolean | null; userId: number | null; logout:
() => void };
+const authState = vi.hoisted(() => ({
+ admin: false as boolean | null,
+ userId: 42 as number | null,
+ logout: vi.fn(),
+}));
+vi.mock('../../../stores/authStore', () => ({
+ default: (selector: (state: MockAuthState) => unknown) =>
selector(authState),
+}));
+
+vi.mock('../../../api/llm', () => ({
+ getLlmConfig: vi.fn(),
+ getLlmModels: vi.fn(),
+ saveLlmConfig: vi.fn(),
+ testLlmConnection: vi.fn(),
+}));
+
+vi.mock('../../../api/settings', () => ({
+ createDataSource: vi.fn(),
+ deleteDataSource: vi.fn(),
+ listAllDataSources: vi.fn(),
+ listDataSourcesPage: vi.fn(),
+ testDataSource: vi.fn(),
+ updateDataSource: vi.fn(),
+}));
+
+vi.mock('../../../api/cloudCredential', () => ({
+ createCloudCredential: vi.fn(),
+ deleteCloudCredential: vi.fn(),
+ exportCloudCredentials: vi.fn(),
+ listCloudCredentials: vi.fn(),
+ updateCloudCredential: vi.fn(),
+}));
+
+vi.mock('../../../services/instanceService', () => ({
+ listInstances: vi.fn().mockResolvedValue([]),
+}));
+
+beforeAll(() => {
+ Object.defineProperty(window, 'matchMedia', {
+ writable: true,
+ value: vi.fn().mockImplementation((query: string) => ({
+ matches: false,
+ media: query,
+ onchange: null,
+ addListener: vi.fn(),
+ removeListener: vi.fn(),
+ addEventListener: vi.fn(),
+ removeEventListener: vi.fn(),
+ dispatchEvent: vi.fn(),
+ })),
+ });
+});
+
+const renderTab = (tab: ReactElement) =>
+ render(
+ <App>
+ <LangProvider>
+ <ThemeProvider>{tab}</ThemeProvider>
+ </LangProvider>
+ </App>,
+ );
+
+beforeEach(() => {
+ vi.clearAllMocks();
+ localStorage.clear();
+ localStorage.setItem(LANGUAGE_STORAGE_KEY, 'en');
+ authState.admin = false;
+ authState.userId = 42;
+});
+
+describe('AiAssistantTab reader access', () => {
+ beforeEach(() => {
+ vi.mocked(getLlmModels).mockResolvedValue({ status: 0, data: [] });
+ });
+
+ it('disables the admin-only save and test controls for readers', async () =>
{
+ // The whole LLM configuration surface is admin-only: GET /api/llm/config
is an admin-only
+ // read, so a reader lands on the load error, and POST /api/llm/config
rejects the write with
+ // 403 (AuthInterceptor). Neither button can ever complete for this role.
+ vi.mocked(getLlmConfig).mockRejectedValue({
+ response: { status: 403, data: { message: 'Forbidden' } },
+ });
+
+ renderTab(<AiAssistantTab />);
+
+ // The load error is what a reader always sees here, and it is also the
point at which the form
+ // is out of its loading state.
+ await screen.findByText('Failed to load AI config');
+ expect(screen.getByRole('button', { name: 'Save' })).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Test connection/
})).toBeDisabled();
+ });
+
+ it('keeps the save and test controls usable for an admin', async () => {
+ authState.admin = true;
+ vi.mocked(getLlmConfig).mockResolvedValue({
+ provider: 'openai',
+ apiBase: 'https://example.com/v1',
+ model: 'gpt-4o-mini',
+ apiKeyConfigured: true,
+ maxTokens: 4096,
+ temperature: 0.7,
+ enabled: true,
+ });
+
+ renderTab(<AiAssistantTab />);
+
+ // The clear-key control renders only once the loaded config reports a
stored key, so awaiting it
+ // is also the wait for the load to settle (loading alone disables both
buttons).
+ expect(await screen.findByRole('button', { name: 'Clear key'
})).toBeEnabled();
+ expect(screen.getByRole('button', { name: 'Save' })).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Test connection/
})).toBeEnabled();
+ });
+});
+
+describe('DataSourceTab reader access', () => {
+ const dataSource = {
+ key: 'ds-1',
+ name: 'prod-prom',
+ type: 'PROMETHEUS',
+ url: 'http://prom:9090',
+ auth: 'NONE',
+ status: 'HEALTHY',
+ };
+
+ beforeEach(() => {
+ vi.mocked(listDataSourcesPage).mockResolvedValue({
+ items: [dataSource],
+ total: 1,
+ page: 1,
+ size: 10,
+ } as never);
+ });
+
+ it('disables the admin-only writes and the connection test for readers',
async () => {
+ renderTab(<DataSourceTab />);
+
+ // POST /api/settings/datasources/* falls through to the admin
requirement, the connection test
+ // included, so every one of these would answer 403 for this role.
+ expect(await screen.findByText('prod-prom')).toBeInTheDocument();
+ expect(screen.getByRole('button', { name: /Add data source/
})).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Test connection/
})).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Edit/ })).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Delete/ })).toBeDisabled();
+ });
+
+ it('keeps them usable for an admin', async () => {
+ authState.admin = true;
+ renderTab(<DataSourceTab />);
+
+ await screen.findByText('prod-prom');
+ expect(screen.getByRole('button', { name: /Add data source/
})).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Test connection/
})).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Edit/ })).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Delete/ })).toBeEnabled();
+ });
+});
+
+describe('CloudCredentialTab reader access', () => {
+ beforeEach(() => {
+ vi.mocked(listCloudCredentials).mockResolvedValue({
+ items: [
+ {
+ id: 1,
+ name: 'aliyun-prod',
+ vendor: 'ALIYUN',
+ accessKey: 'LTAI****',
+ remark: '',
+ gmtCreate: '2026-08-01T10:00:00Z',
+ },
+ ],
+ total: 1,
+ page: 1,
+ size: 10,
+ } as never);
+ });
+
+ it('disables the admin-only writes for readers', async () => {
+ renderTab(<CloudCredentialTab />);
+
+ expect(await screen.findByText('aliyun-prod')).toBeInTheDocument();
+ expect(screen.getByRole('button', { name: /Add credential/
})).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Edit/ })).toBeDisabled();
+ expect(screen.getByRole('button', { name: /Delete/ })).toBeDisabled();
+ });
+
+ it('keeps them usable for an admin', async () => {
+ authState.admin = true;
+ renderTab(<CloudCredentialTab />);
+
+ await screen.findByText('aliyun-prod');
+ expect(screen.getByRole('button', { name: /Add credential/
})).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Edit/ })).toBeEnabled();
+ expect(screen.getByRole('button', { name: /Delete/ })).toBeEnabled();
+ });
+});
diff --git a/web/src/utils/clientConnectionDiagnostics.test.ts
b/web/src/utils/clientConnectionDiagnostics.test.ts
index 5bf2ab3be..698f34046 100644
--- a/web/src/utils/clientConnectionDiagnostics.test.ts
+++ b/web/src/utils/clientConnectionDiagnostics.test.ts
@@ -33,6 +33,51 @@ const connection = (overrides: Partial<ClientConnection>):
ClientConnection => (
});
describe('client connection diagnostics', () => {
+ it.each([
+ ['IPv4', '192.0.2.10:51001', '192.0.2.10:51002'],
+ ['IPv4 socket notation', '/192.0.2.10:51001', '192.0.2.10:51002'],
+ ['IPv6', '[2001:db8::10]:51001', '[2001:db8::10]:51002'],
+ ])(
+ 'keeps one client connected to multiple brokers without a collision (%s)
Test',
+ (_, first, second) => {
+ const diagnostics = analyzeClientConnections([
+ connection({ clientId: '192.0.2.10@4242', protocol: 'Remoting',
address: first }),
+ connection({ clientId: '192.0.2.10@4242', protocol: 'Remoting',
address: second }),
+ ]);
+ expect(diagnostics.issues.map((item) =>
item.code)).not.toContain('CLIENT_ID_COLLISION');
+ expect(diagnostics.issues.map((item) => item.code)).not.toContain(
+ 'EXACT_DUPLICATE_CONNECTION',
+ );
+ expect(diagnostics.status).toBe('healthy');
+ expect(diagnostics.summary).toMatchObject({ totalConnections: 2,
uniqueAddressCount: 2 });
+ },
+ );
+
+ it.each([
+ ['192.0.2.10:51001', '192.0.2.11:51002'],
+ ['[2001:db8::10]:51001', '[2001:db8::11]:51002'],
+ ])('still reports the same client ID on different hosts (%s, %s) Test',
(first, second) => {
+ const diagnostics = analyzeClientConnections([
+ connection({ clientId: 'shared-client', address: first }),
+ connection({ clientId: 'shared-client', address: second }),
+ ]);
+ expect(diagnostics.issues).toContainEqual(
+ expect.objectContaining({
+ code: 'CLIENT_ID_COLLISION',
+ severity: 'critical',
+ evidence: [first, second],
+ }),
+ );
+ });
+
+ it('keeps distinct unrecognized addresses distinct Test', () => {
+ const diagnostics = analyzeClientConnections([
+ connection({ clientId: 'shared-client', address: 'unknown-a' }),
+ connection({ clientId: 'shared-client', address: 'unknown-b' }),
+ ]);
+ expect(diagnostics.issues.map((item) =>
item.code)).toContain('CLIENT_ID_COLLISION');
+ });
+
it('marks a consistent multi-client inventory as healthy', () => {
const diagnostics = analyzeClientConnections([
connection({ clientId: 'producer-a', address: '10.0.1.10:49152' }),
diff --git a/web/src/utils/clientConnectionDiagnostics.ts
b/web/src/utils/clientConnectionDiagnostics.ts
index 094f70c68..77cad4ec9 100644
--- a/web/src/utils/clientConnectionDiagnostics.ts
+++ b/web/src/utils/clientConnectionDiagnostics.ts
@@ -164,6 +164,13 @@ const connectionIdentity = (connection: ClientConnection):
string =>
const resourceKey = (connection: ClientConnection): string =>
`${normalizeText(connection.type)}:${normalizeText(connection.groupOrTopic)}`;
+const connectionHost = (address: string | null | undefined): string => {
+ const normalized = normalizeText(address).replace(/^\//, '');
+ // Broker connections use different TCP source ports for the same client
process.
+ const socket = /^(?:\[([^\]]+)\]|([^:]+)):\d+$/.exec(normalized);
+ return socket ? (socket[1] ?? socket[2]) : normalized;
+};
+
const groupConnections = (connections: ClientConnection[]): ConnectionGroup[]
=> {
const groups = new Map<string, ConnectionGroup>();
@@ -245,7 +252,10 @@ const addClientIdIssues = (connections:
ClientConnection[], issues: ClientConnec
connectionsByClient.forEach((clientConnections, clientId) => {
const addresses = uniqueSorted(clientConnections.map((connection) =>
connection.address));
- if (addresses.length > 1) {
+ const hosts = uniqueSorted(
+ clientConnections.map((connection) =>
connectionHost(connection.address)),
+ );
+ if (hosts.length > 1) {
issues.push(
issue(
'CLIENT_ID_COLLISION',
