mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-15 07:40:59 +00:00
fix(node): adopt a matching deployed inbound instead of recreating it (#6197)
* fix(node): adopt compatible origin inbounds without mutation * fix(nodes): preserve ambiguous and adopted aliases --------- Co-authored-by: n0ctal <293235942+n0ctal@users.noreply.github.com>
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user