diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index fcdc50839..78cd2c234 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -389,6 +389,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy j.inboundService.ClearNodeOnlineClients(n.Id) return nil } + snap.ManagedAliases = rt.AdoptedInboundAliases() service.FilterNodeSnapshot(n, snap) _, _, dirty, _, _ := j.nodeService.NodeSyncState(n.Id) if !dirty { diff --git a/internal/web/runtime/remote.go b/internal/web/runtime/remote.go index 17828788f..c1625f39d 100644 --- a/internal/web/runtime/remote.go +++ b/internal/web/runtime/remote.go @@ -78,8 +78,9 @@ func (e *remoteAPIError) Error() string { return "remote: " + e.msg } type Remote struct { node *model.Node - mu sync.RWMutex - remoteIDByTag map[string]int + mu sync.RWMutex + remoteIDByTag map[string]int + adoptedAliases map[string]string // pushedFP holds the fingerprint of the last inbound wire payload successfully // pushed, keyed by panel-side tag, so reconcile can skip re-sending an // unchanged inbound. Guarded by mu; dropped with the Remote on node config change. @@ -99,8 +100,10 @@ type Remote struct { } type RemoteInboundOption struct { + Id int `json:"id"` Tag string `json:"tag"` Remark string `json:"remark"` + Listen string `json:"listen"` Protocol model.Protocol `json:"protocol"` Port int `json:"port"` } @@ -109,6 +112,7 @@ func NewRemote(n *model.Node, r NodeEgressResolver) *Remote { return &Remote{ node: n, remoteIDByTag: make(map[string]int), + adoptedAliases: make(map[string]string), pushedFP: make(map[string]string), egressResolver: r, } @@ -479,13 +483,33 @@ func (r *Remote) recordPushedInbound(ib *model.Inbound) { r.mu.Unlock() } -// RecordAdoptedInbound stamps the fingerprint when the master adopts the -// node's own settings serialization into its DB — direct knowledge of the -// exact payload the node holds. +// RecordAdoptedInbound stamps the exact payload fingerprint after the master +// adopts a node's settings serialization. func (r *Remote) RecordAdoptedInbound(ib *model.Inbound) { r.recordPushedInbound(ib) } +// AdoptInboundAlias records a deployed alias without mutating either panel. +// The runtime association is rediscovered after a master restart. +func (r *Remote) AdoptInboundAlias(ib *model.Inbound, remote RemoteInboundOption) { + r.mu.Lock() + r.remoteIDByTag[remote.Tag] = remote.Id + r.remoteIDByTag[ib.Tag] = remote.Id + r.adoptedAliases[ib.Tag] = remote.Tag + r.pushedFP[ib.Tag] = wireFingerprint(wireInbound(ib, r.node.Id)) + r.mu.Unlock() +} + +func (r *Remote) AdoptedInboundAliases() []string { + r.mu.RLock() + defer r.mu.RUnlock() + aliases := make([]string, 0, len(r.adoptedAliases)) + for _, alias := range r.adoptedAliases { + aliases = append(aliases, alias) + } + return aliases +} + // AdvancePushedInbound moves the reconcile-skip fingerprint from an inbound's // pre-edit payload to its post-edit payload once every per-client push for the // edit succeeded. It advances only when the recorded fingerprint proves the @@ -661,8 +685,9 @@ func (r *Remote) ResetInboundTraffic(ctx context.Context, ib *model.Inbound) err } type TrafficSnapshot struct { - Inbounds []*model.Inbound - OnlineEmails []string + Inbounds []*model.Inbound + OnlineEmails []string + ManagedAliases []string // OnlineTree is the node's GUID-keyed online subtree (its own clients under // its panelGuid plus every descendant under theirs). Preferred over the flat // OnlineEmails so the master can attribute deeply nested clients to the real diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index ef4c4adb9..252d71dda 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -96,13 +96,15 @@ func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote, if err := db.Model(model.Inbound{}).Where("node_id = ?", nodeID).Find(&inbounds).Error; err != nil { return err } - remoteTags, err := rt.ListRemoteTags(ctx) + remoteInbounds, err := rt.ListInboundOptions(ctx) if err != nil { return err } + remoteTags := make([]string, 0, len(remoteInbounds)) remoteTagSet := make(map[string]struct{}, len(remoteTags)) - for _, tag := range remoteTags { - remoteTagSet[tag] = struct{}{} + for _, remoteIb := range remoteInbounds { + remoteTags = append(remoteTags, remoteIb.Tag) + remoteTagSet[remoteIb.Tag] = struct{}{} } prefix := nodeTagPrefix(&nodeID) desiredTags := make(map[string]struct{}, len(inbounds)*2) @@ -129,6 +131,33 @@ func (s *InboundService) ReconcileNode(ctx context.Context, rt *runtime.Remote, if built, bErr := s.buildInboundForNodePush(db, ib); bErr == nil { runtimeIb = built } + if !existsOnNode && n.Guid != "" && ib.OriginNodeGuid == n.Guid { + var compatible []runtime.RemoteInboundOption + for _, remoteIb := range remoteInbounds { + if remoteIb.Port == runtimeIb.Port && + remoteIb.Protocol == runtimeIb.Protocol && + strings.TrimSpace(remoteIb.Listen) == strings.TrimSpace(runtimeIb.Listen) { + compatible = append(compatible, remoteIb) + } + } + switch len(compatible) { + case 1: + alias := compatible[0] + desiredTags[alias.Tag] = struct{}{} + rt.AdoptInboundAlias(runtimeIb, alias) + existsOnNode = true + logger.Infof("adopted compatible inbound %q on node %s as %q", alias.Tag, n.Name, ib.Tag) + case 0: + // No compatible occupant: keep the normal create path, which + // leaves a real port/protocol drift loud. + default: + for _, candidate := range compatible { + desiredTags[candidate.Tag] = struct{}{} + } + errs = append(errs, fmt.Errorf("reconcile inbound %q: ambiguous compatible remote inbounds", ib.Tag)) + continue + } + } if _, err := rt.ReconcileInbound(ctx, runtimeIb, existsOnNode); err != nil { errs = append(errs, fmt.Errorf("reconcile inbound %q: %w", ib.Tag, err)) } @@ -514,6 +543,29 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } c, ok := tagToCentral[snapIb.Tag] + if !ok { + origin := originGuidFor(snapIb) + var compatible []*model.Inbound + for i := range central { + candidate := ¢ral[i] + if candidate.OriginNodeGuid == origin && + candidate.Port == snapIb.Port && + candidate.Protocol == snapIb.Protocol && + strings.TrimSpace(candidate.Listen) == strings.TrimSpace(snapIb.Listen) { + compatible = append(compatible, candidate) + } + } + switch len(compatible) { + case 1: + c, ok = compatible[0], true + tagToCentral[snapIb.Tag] = c + snapTags[c.Tag] = struct{}{} + case 0: + // A genuinely new inbound follows the normal adoption path. + default: + return false, fmt.Errorf("setRemoteTraffic: inbound %q has ambiguous compatible central aliases", snapIb.Tag) + } + } if !ok { if dirty { continue diff --git a/internal/web/service/inbound_node_reconcile_test.go b/internal/web/service/inbound_node_reconcile_test.go index 337094949..9817652f1 100644 --- a/internal/web/service/inbound_node_reconcile_test.go +++ b/internal/web/service/inbound_node_reconcile_test.go @@ -251,6 +251,107 @@ func TestReconcileNode_ContinuesPastFailedInbound(t *testing.T) { } } +func TestReconcileNode_AdoptsCompatibleOriginInboundWithoutRemoteMutation(t *testing.T) { + setupConflictDB(t) + + var mu sync.Mutex + mutations := 0 + writeOK := func(w http.ResponseWriter, obj any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj}) + } + mux := http.NewServeMux() + mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) { + writeOK(w, []map[string]any{{"id": 41, "tag": "already-deployed", "listen": "", "port": 8443, "protocol": "vless"}}) + }) + mux.HandleFunc("/panel/api/inbounds/", func(w http.ResponseWriter, _ *http.Request) { + mu.Lock() + mutations++ + mu.Unlock() + writeOK(w, nil) + }) + ts := httptest.NewServer(mux) + t.Cleanup(ts.Close) + + node := reconcileTestNode(t, ts, "adopt-node", "all", nil) + node.Guid = "origin-guid" + if err := database.GetDB().Model(node).Update("guid", node.Guid).Error; err != nil { + t.Fatalf("update node guid: %v", err) + } + seedInboundConflictNode(t, "desired-name", "", 8443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id) + if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil { + t.Fatalf("set origin guid: %v", err) + } + + svc := InboundService{} + rt := runtime.NewRemote(node, nil) + if err := svc.ReconcileNode(context.Background(), rt, node); err != nil { + t.Fatalf("first ReconcileNode: %v", err) + } + if err := svc.ReconcileNode(context.Background(), rt, node); err != nil { + t.Fatalf("second ReconcileNode: %v", err) + } + mu.Lock() + got := mutations + mu.Unlock() + if got != 0 { + t.Fatalf("remote mutations = %d, want 0 while adopting compatible deployed inbound", got) + } +} + +func TestReconcileNode_AmbiguousCompatibleInboundsAreNotSwept(t *testing.T) { + setupConflictDB(t) + + ts, deletedIDs := fakeNodePanel(t, map[string]int{"alias-a": 51, "alias-b": 52}) + node := reconcileTestNode(t, ts, "ambiguous-node", "all", nil) + node.Guid = "origin-guid" + if err := database.GetDB().Model(node).Update("guid", node.Guid).Error; err != nil { + t.Fatalf("update node guid: %v", err) + } + seedInboundConflictNode(t, "desired-name", "", 0, model.Protocol(""), `{}`, `{"clients":[]}`, &node.Id) + if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil { + t.Fatalf("set origin guid: %v", err) + } + + err := (&InboundService{}).ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node) + if err == nil || !strings.Contains(err.Error(), "ambiguous compatible remote inbounds") { + t.Fatalf("ReconcileNode error = %v, want ambiguity error", err) + } + if got := deletedIDs(); len(got) != 0 { + t.Fatalf("deleted ambiguous candidates = %v, want none", got) + } +} + +func TestReconcileNode_IncompatiblePortOccupantRemainsLoud(t *testing.T) { + setupConflictDB(t) + + writeOK := func(w http.ResponseWriter, obj any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"success": true, "msg": "", "obj": obj}) + } + mux := http.NewServeMux() + mux.HandleFunc("/panel/api/inbounds/list", func(w http.ResponseWriter, _ *http.Request) { + writeOK(w, []map[string]any{{"id": 42, "tag": "port-owner", "listen": "", "port": 9443, "protocol": "trojan"}}) + }) + mux.HandleFunc("/panel/api/inbounds/add", func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"success": false, "msg": "port already occupied", "obj": nil}) + }) + ts := httptest.NewServer(mux) + t.Cleanup(ts.Close) + + node := reconcileTestNode(t, ts, "drift-node", "all", nil) + node.Guid = "origin-guid" + seedInboundConflictNode(t, "desired-name", "", 9443, model.VLESS, `{"network":"tcp"}`, `{"clients":[]}`, &node.Id) + if err := database.GetDB().Model(&model.Inbound{}).Where("tag = ?", "desired-name").Update("origin_node_guid", node.Guid).Error; err != nil { + t.Fatalf("set origin guid: %v", err) + } + + err := (&InboundService{}).ReconcileNode(context.Background(), runtime.NewRemote(node, nil), node) + if err == nil || !strings.Contains(err.Error(), "port already occupied") { + t.Fatalf("ReconcileNode error = %v, want loud incompatible-port error", err) + } +} + func TestEnsureInboundTagAllowed(t *testing.T) { setupConflictDB(t) db := database.GetDB() diff --git a/internal/web/service/node.go b/internal/web/service/node.go index a22a40d7a..8977f4d24 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -689,6 +689,9 @@ func FilterNodeSnapshot(n *model.Node, snap *runtime.TrafficSnapshot) { return } allowed := nodeSelectedTagSet(n) + for _, tag := range snap.ManagedAliases { + allowed[tag] = struct{}{} + } filtered := make([]*model.Inbound, 0, len(snap.Inbounds)) for _, inbound := range snap.Inbounds { if inbound == nil { diff --git a/internal/web/service/node_tag_sync_test.go b/internal/web/service/node_tag_sync_test.go index aa0f6a0fa..f97a12e9a 100644 --- a/internal/web/service/node_tag_sync_test.go +++ b/internal/web/service/node_tag_sync_test.go @@ -64,3 +64,37 @@ func TestSetRemoteTraffic_KeepsInboundOnPrefixMismatch(t *testing.T) { t.Fatalf("traffic not attributed across prefix mismatch: up=%d down=%d", rows[0].Up, rows[0].Down) } } + +func TestSetRemoteTraffic_AdoptsCompatibleOriginAliasWithoutDuplicate(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + + const nodeID = 1 + if err := db.Create(&model.Node{Id: nodeID, Name: "node", Address: "10.0.0.2", Port: 2053, ApiToken: "t", Guid: "node-guid"}).Error; err != nil { + t.Fatalf("create node: %v", err) + } + id := nodeID + central := &model.Inbound{UserId: 1, NodeID: &id, OriginNodeGuid: "node-guid", Tag: "desired-name", Enable: true, Port: 8443, Protocol: model.VLESS, Settings: `{"clients":[]}`} + if err := db.Create(central).Error; err != nil { + t.Fatalf("create central inbound: %v", err) + } + + snap := &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{{ + Tag: "already-deployed", Enable: true, Port: 8443, Protocol: model.VLESS, + Settings: `{"clients":[]}`, Up: 11, Down: 22, + }}} + if _, err := (&InboundService{}).setRemoteTrafficLocked(nodeID, snap, false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + + var rows []model.Inbound + if err := db.Where("node_id = ?", nodeID).Find(&rows).Error; err != nil { + t.Fatalf("list node inbounds: %v", err) + } + if len(rows) != 1 || rows[0].Id != central.Id || rows[0].Tag != "desired-name" { + t.Fatalf("alias adoption rows = %#v, want original central inbound only", rows) + } + if rows[0].Up != 11 || rows[0].Down != 22 { + t.Fatalf("alias traffic = %d/%d, want 11/22", rows[0].Up, rows[0].Down) + } +} diff --git a/internal/web/service/node_test.go b/internal/web/service/node_test.go index e3f255a26..ddddc0b2a 100644 --- a/internal/web/service/node_test.go +++ b/internal/web/service/node_test.go @@ -235,3 +235,18 @@ func TestFilterNodeSnapshotMatchesPrefixedSelectedTag(t *testing.T) { t.Fatalf("bare selected tag in-100-tcp was dropped; kept=%v", kept) } } + +func TestFilterNodeSnapshotKeepsAdoptedAlias(t *testing.T) { + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{Tag: "deployed-alias"}, {Tag: "unmanaged"}}, + ManagedAliases: []string{"deployed-alias"}, + } + FilterNodeSnapshot(&model.Node{ + InboundSyncMode: "selected", + InboundTags: []string{"desired-name"}, + }, snap) + + if len(snap.Inbounds) != 1 || snap.Inbounds[0].Tag != "deployed-alias" { + t.Fatalf("filtered snapshot = %#v, want adopted alias only", snap.Inbounds) + } +}