From 95c3e986115a4d3f1fbbdc64a9fc3a0dfad57770 Mon Sep 17 00:00:00 2001 From: Hungerdream <1710233908@qq.com> Date: Fri, 31 Jul 2026 18:49:34 +0800 Subject: [PATCH] =?UTF-8?q?fix(server):=20=E6=98=BE=E5=BC=8F=20target=5Fho?= =?UTF-8?q?sts=20=E6=97=B6=E4=B8=BB=E6=9C=BA=E6=8C=87=E5=90=91=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E6=8C=87=E5=AE=9A=E7=9A=84=E4=B8=BB=E8=8A=82=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 显式列表校验通过后令 host = nodes[0],task.TargetHost/IP、台账、 CloudDM 注册与凭据地址不再指向此前默认挑选的空闲主机 - 显式列表解析抽为纯函数 resolvePinnedPrimaryReplicaNodes, 行为与原内联逻辑一致(校验失败终态/单机上限 defer 重试) - 新增测试覆盖主节点非池首空闲节点及数量/池外/重复/占满场景 --- server/internal/service/delivery.go | 69 +++++++++++++++--------- server/internal/service/delivery_test.go | 29 ++++++++++ 2 files changed, 72 insertions(+), 26 deletions(-) diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 25ea5d4..0200da6 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -335,6 +335,40 @@ func hostTaskCount(occupied []string, name string) int { return count } +// resolvePinnedPrimaryReplicaNodes 按用户显式指定的有序 target_hosts 解析 primary_replica 节点列表， +// 首节点为主节点。返回值：节点列表、参数校验失败原因（终态）、延后调度错误（defer: 前缀，可重试）。 +func resolvePinnedPrimaryReplicaNodes(hosts []targetHost, requested []string, occupied []string, limit int, nodeCount int64, pinnedPrimary string) ([]*targetHost, string, error) { + if int64(len(requested)) != nodeCount { + return nil, fmt.Sprintf("primary_replica requires %d target_hosts, got %d", nodeCount, len(requested)), nil + } + nodes := make([]*targetHost, 0, len(requested)) + seen := map[string]struct{}{} + for index, requestedHost := range requested { + var selected *targetHost + for i := range hosts { + if hosts[i].Name == requestedHost { + selected = &hosts[i] + break + } + } + if selected == nil { + return nil, fmt.Sprintf("target_hosts[%d] %q is not in the candidate host pool", index, requestedHost), nil + } + if _, exists := seen[selected.Name]; exists { + return nil, fmt.Sprintf("target_hosts contains duplicate host %q", selected.Name), nil + } + if hostTaskCount(occupied, selected.Name) >= max(limit, 1) { + return nil, "", fmt.Errorf("defer: pinned host %s reached the per-host instance limit %d", selected.Name, max(limit, 1)) + } + seen[selected.Name] = struct{}{} + nodes = append(nodes, selected) + } + if pinnedPrimary != "" && nodes[0].Name != pinnedPrimary { + return nil, "target_host must match the first primary_replica target_hosts entry", nil + } + return nodes, "", nil +} + func topologyNodeCount(input MySQLDeliveryInput) int64 { if input.Topology == "primary_replica" { replicas := input.ReplicaCount @@ -1757,34 +1791,17 @@ func (s *DeliveryService) claimAndReserve(ctx context.Context) (*model.DeliveryT nodes := []*targetHost{host} if nodeCount > 1 { if len(payload.TargetHosts) > 0 { - if int64(len(payload.TargetHosts)) != nodeCount { - return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("primary_replica requires %d target_hosts, got %d", nodeCount, len(payload.TargetHosts))) + pinned, validationMsg, deferErr := resolvePinnedPrimaryReplicaNodes(meta.Hosts, payload.TargetHosts, occupied, limit, nodeCount, payload.TargetHost) + if deferErr != nil { + return deferErr } - nodes = nodes[:0] - seen := map[string]struct{}{} - for index, requestedHost := range payload.TargetHosts { - var selected *targetHost - for i := range meta.Hosts { - if meta.Hosts[i].Name == requestedHost { - selected = &meta.Hosts[i] - break - } - } - if selected == nil { - return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_hosts[%d] %q is not in the candidate host pool", index, requestedHost)) - } - if _, exists := seen[selected.Name]; exists { - return s.failInTransaction(tx, &task, model.TaskValidationFailed, fmt.Sprintf("target_hosts contains duplicate host %q", selected.Name)) - } - if hostTaskCount(occupied, selected.Name) >= max(limit, 1) { - return fmt.Errorf("defer: pinned host %s reached the per-host instance limit %d", selected.Name, max(limit, 1)) - } - seen[selected.Name] = struct{}{} - nodes = append(nodes, selected) - } - if payload.TargetHost != "" && nodes[0].Name != payload.TargetHost { - return s.failInTransaction(tx, &task, model.TaskValidationFailed, "target_host must match the first primary_replica target_hosts entry") + if validationMsg != "" { + return s.failInTransaction(tx, &task, model.TaskValidationFailed, validationMsg) } + // 显式列表以首节点为主节点：覆盖此前默认挑选的空闲主机， + // 避免 task.TargetHost/IP 与台账、凭据指向未部署节点。 + nodes = pinned + host = nodes[0] } else { for i := range meta.Hosts { candidate := &meta.Hosts[i] diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index 6a5c2ef..a197443 100644 --- a/server/internal/service/delivery_test.go +++ b/server/internal/service/delivery_test.go @@ -138,6 +138,35 @@ func TestFirstFreeHost(t *testing.T) { } } +func TestResolvePinnedPrimaryReplicaNodes(t *testing.T) { + pool := []targetHost{{Name: "node-a", IP: "10.0.0.1"}, {Name: "node-b", IP: "10.0.0.2"}, {Name: "node-c", IP: "10.0.0.3"}} + + // 主节点不是池首空闲节点（firstFreeHost 会选 node-a）：首节点必须以用户列表为准。 + nodes, validationMsg, err := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-c", "node-a"}, nil, 4, 2, "") + if err != nil || validationMsg != "" { + t.Fatalf("unexpected rejection: msg=%q err=%v", validationMsg, err) + } + if nodes[0].Name != "node-c" || nodes[0].IP != "10.0.0.3" || nodes[1].Name != "node-a" { + t.Fatalf("primary must follow the user-specified order, got %+v", nodes) + } + + if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a"}, nil, 4, 2, ""); msg == "" { + t.Fatal("host count mismatch was accepted") + } + if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-x"}, nil, 4, 2, ""); msg == "" { + t.Fatal("host outside the pool was accepted") + } + if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-a"}, nil, 4, 2, ""); msg == "" { + t.Fatal("duplicate host was accepted") + } + if _, msg, _ := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-b", "node-c"}, nil, 4, 2, "node-a"); msg == "" { + t.Fatal("target_host mismatching the first entry was accepted") + } + if _, _, err := resolvePinnedPrimaryReplicaNodes(pool, []string{"node-a", "node-b"}, []string{"node-b"}, 1, 2, ""); err == nil { + t.Fatal("saturated pinned host must defer scheduling") + } +} + func TestAllocatePort(t *testing.T) { if port, err := allocatePort(0, nil); err != nil || port != mysqlPortPoolStart { t.Fatalf("expected first pool port %d, got %d err=%v", mysqlPortPoolStart, port, err)