mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-16 00:01:02 +00:00
fix(node): stop client edits from tearing down node inbounds and harden reconcile fingerprints
A client save on the master always stamped a fresh updated_at, marked the node dirty, and let the 5s sync push a full inbounds/update to the node, where applying it removes and re-adds the Xray handler - killing live traffic on every edit, including no-op saves (open the editor, click Save). Nodes stayed online with Xray running while forwarding nothing until a manual Xray restart. - No-op client saves preserve the client's updated_at and return before any DB write, runtime RPC, or node dirty mark when the effective settings did not change. - Successful per-client add/update/delete pushes advance the node's reconcile-skip fingerprint only when the recorded fingerprint proves the node held the exact pre-edit payload and every push in the edit succeeded (Remote.AdvancePushedInbound). Anything unproven keeps the stale fingerprint so the dirty reconcile still sends the full inbound. Unconditional stamping would certify folded bulk changes (threshold, flow change, offline edit) or partially failed batches as delivered: a folded 41->6 bulk delete followed by one live edit left the node permanently serving all 41 clients in end-to-end testing, with the snapshot adoption then resurrecting the deleted clients on the master. - DeleteUser treats only an envelope-level not-found as already deleted; an HTTP 404 from an old node build without the detach endpoint surfaces as an error instead of certifying an undelivered delete. cacheDel drops the fingerprint alongside the id cache so DelInbound and tag renames leave no stale skip entry. - Adopting the node's own settings serialization into the master row now also stamps the fingerprint (RecordAdoptedInbound). Without it the serialization round-trip invalidated the fingerprint one sync tick after every push, so each edit degraded back to a full teardown push. - UpdateInboundClient applies the Shadowsocks method normalization before the no-op comparison (real method changes bump updated_at, SS no-op edits are detected) and syncs the generated subId into the pushed client so the node cannot mint a different one. Verified with a two-panel docker deployment: no-op saves produce zero node requests, real edits send one lightweight clients/update RPC with zero full inbound updates and zero handler teardowns, and folded bulk deletes still converge. Based on PR #5778 by @rqzbeh. Closes #5764 Closes #5771
This commit is contained in:
@@ -26,7 +26,15 @@ func clientsSettings(t *testing.T, clients []model.Client) string {
|
||||
if err != nil {
|
||||
t.Fatalf("marshal settings: %v", err)
|
||||
}
|
||||
return string(b)
|
||||
var out map[string]any
|
||||
if err := json.Unmarshal(b, &out); err != nil {
|
||||
t.Fatalf("unmarshal settings: %v", err)
|
||||
}
|
||||
b2, err := json.MarshalIndent(out, "", " ")
|
||||
if err != nil {
|
||||
t.Fatalf("marshal settings again: %v", err)
|
||||
}
|
||||
return string(b2)
|
||||
}
|
||||
|
||||
func emailsOf(clients []model.Client) []string {
|
||||
|
||||
@@ -645,6 +645,7 @@ func (s *ClientService) bulkAdjustInboundClients(
|
||||
}
|
||||
return res
|
||||
}
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
// A flow change rewrites the user's xray config, which the lightweight
|
||||
@@ -670,6 +671,7 @@ func (s *ClientService) bulkAdjustInboundClients(
|
||||
push = false
|
||||
}
|
||||
if push {
|
||||
pushFailed := false
|
||||
for email := range foundEmails {
|
||||
entry := plan[email]
|
||||
updated := *entry.record.ToClient()
|
||||
@@ -682,8 +684,12 @@ func (s *ClientService) bulkAdjustInboundClients(
|
||||
updated.UpdatedAt = nowMs
|
||||
if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil {
|
||||
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
||||
pushFailed = true
|
||||
}
|
||||
}
|
||||
if !pushFailed {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -960,6 +966,7 @@ func (s *ClientService) bulkDelInboundClients(
|
||||
}
|
||||
return res
|
||||
}
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
foundList := make([]string, 0, len(foundEmails))
|
||||
@@ -1060,11 +1067,16 @@ func (s *ClientService) bulkDelInboundClients(
|
||||
push = false
|
||||
}
|
||||
if push {
|
||||
pushFailed := false
|
||||
for email := range foundEmails {
|
||||
if err1 := rt.DeleteUser(context.Background(), oldInbound, email); err1 != nil {
|
||||
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
||||
pushFailed = true
|
||||
}
|
||||
}
|
||||
if !pushFailed {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1560,6 +1572,7 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService,
|
||||
}
|
||||
return res
|
||||
}
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
|
||||
@@ -1625,13 +1638,18 @@ func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService,
|
||||
}
|
||||
}
|
||||
} else if push {
|
||||
pushFailed := false
|
||||
for _, ch := range changed {
|
||||
updated := ch.client
|
||||
updated.UpdatedAt = nowMs
|
||||
if err1 := rt.UpdateUser(context.Background(), oldInbound, ch.email, updated); err1 != nil {
|
||||
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
||||
pushFailed = true
|
||||
}
|
||||
}
|
||||
if !pushFailed {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
|
||||
return res
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"maps"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -18,6 +19,28 @@ import (
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func sameClientConfigExceptUpdatedAt(a, b map[string]any) bool {
|
||||
aa := maps.Clone(a)
|
||||
bb := maps.Clone(b)
|
||||
delete(aa, "updated_at")
|
||||
delete(bb, "updated_at")
|
||||
an, aerr := json.Marshal(aa)
|
||||
bn, berr := json.Marshal(bb)
|
||||
return aerr == nil && berr == nil && string(an) == string(bn)
|
||||
}
|
||||
|
||||
// advancePushedInbound advances the node's reconcile-skip fingerprint from the
|
||||
// pre-edit settings to the saved ones after every per-client push succeeded.
|
||||
func advancePushedInbound(rt runtime.Runtime, prevSettings string, ib *model.Inbound) {
|
||||
rem, ok := rt.(*runtime.Remote)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
prev := *ib
|
||||
prev.Settings = prevSettings
|
||||
rem.AdvancePushedInbound(&prev, ib)
|
||||
}
|
||||
|
||||
// delInboundClients removes several clients from a single inbound in one pass:
|
||||
// one settings rewrite, one runtime sweep, one Save and one SyncInbound for the
|
||||
// whole batch, instead of repeating the full per-client cycle. It mirrors the
|
||||
@@ -89,6 +112,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
var sharedSet map[string]bool
|
||||
@@ -185,6 +209,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
||||
|
||||
// Apply runtime deletes after commit — outside the serialized writer so a
|
||||
// slow node call can't stall traffic accounting.
|
||||
nodePushFailed := false
|
||||
for _, t := range targets {
|
||||
if len(t.email) == 0 {
|
||||
continue
|
||||
@@ -203,9 +228,13 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
||||
} else if nodePush {
|
||||
if err1 := nodeRt.DeleteUser(context.Background(), oldInbound, t.email); err1 != nil {
|
||||
logger.Warning("Error in deleting client on", nodeRt.Name(), ":", err1)
|
||||
nodePushFailed = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if nodePush && !nodePushFailed {
|
||||
advancePushedInbound(nodeRt, prevSettings, oldInbound)
|
||||
}
|
||||
|
||||
return needRestart, nil
|
||||
}
|
||||
@@ -349,6 +378,7 @@ func (s *ClientService) addInboundClient(inboundSvc *InboundService, data *model
|
||||
return false, err
|
||||
}
|
||||
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
needRestart := false
|
||||
@@ -441,6 +471,9 @@ func (s *ClientService) addInboundClient(inboundSvc *InboundService, data *model
|
||||
}
|
||||
}
|
||||
}
|
||||
if push {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
|
||||
return needRestart, nil
|
||||
@@ -565,21 +598,25 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
||||
settingsClients, _ := oldSettings["clients"].([]any)
|
||||
var preservedCreated any
|
||||
var preservedSubID string
|
||||
var oldClientMap map[string]any
|
||||
if clientIndex >= 0 && clientIndex < len(settingsClients) {
|
||||
if oldMap, ok := settingsClients[clientIndex].(map[string]any); ok {
|
||||
oldClientMap = oldMap
|
||||
if v, ok2 := oldMap["created_at"]; ok2 {
|
||||
preservedCreated = v
|
||||
}
|
||||
preservedSubID, _ = oldMap["subId"].(string)
|
||||
}
|
||||
}
|
||||
if oldInbound.Protocol == model.Shadowsocks {
|
||||
applyShadowsocksClientMethod(interfaceClients, oldSettings)
|
||||
}
|
||||
if len(interfaceClients) > 0 {
|
||||
if newMap, ok := interfaceClients[0].(map[string]any); ok {
|
||||
if preservedCreated == nil {
|
||||
preservedCreated = time.Now().Unix() * 1000
|
||||
}
|
||||
newMap["created_at"] = preservedCreated
|
||||
newMap["updated_at"] = time.Now().Unix() * 1000
|
||||
newSub, _ := newMap["subId"].(string)
|
||||
if strings.TrimSpace(newSub) == "" {
|
||||
if strings.TrimSpace(preservedSubID) != "" {
|
||||
@@ -588,6 +625,9 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
||||
newMap["subId"] = random.NumLower(16)
|
||||
}
|
||||
}
|
||||
if v, ok2 := newMap["subId"].(string); ok2 {
|
||||
clients[0].SubID = v
|
||||
}
|
||||
if oldInbound.Protocol == model.WireGuard {
|
||||
newMap["privateKey"] = clients[0].PrivateKey
|
||||
newMap["publicKey"] = clients[0].PublicKey
|
||||
@@ -599,12 +639,18 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
||||
newMap["keepAlive"] = clients[0].KeepAlive
|
||||
}
|
||||
}
|
||||
if oldClientMap != nil && sameClientConfigExceptUpdatedAt(oldClientMap, newMap) {
|
||||
if v, ok2 := oldClientMap["updated_at"]; ok2 {
|
||||
newMap["updated_at"] = v
|
||||
} else {
|
||||
delete(newMap, "updated_at")
|
||||
}
|
||||
} else {
|
||||
newMap["updated_at"] = time.Now().Unix() * 1000
|
||||
}
|
||||
interfaceClients[0] = newMap
|
||||
}
|
||||
}
|
||||
if oldInbound.Protocol == model.Shadowsocks {
|
||||
applyShadowsocksClientMethod(interfaceClients, oldSettings)
|
||||
}
|
||||
settingsClients[clientIndex] = interfaceClients[0]
|
||||
oldSettings["clients"] = settingsClients
|
||||
|
||||
@@ -630,6 +676,11 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
||||
return false, err
|
||||
}
|
||||
|
||||
if string(newSettings) == oldInbound.Settings {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
needRestart := false
|
||||
@@ -768,6 +819,8 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
||||
} else if push {
|
||||
if err1 := rt.UpdateUser(context.Background(), oldInbound, oldEmail, clients[0]); err1 != nil {
|
||||
logger.Warning("Error in updating client on", rt.Name(), ":", err1)
|
||||
} else {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
@@ -828,6 +881,7 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
||||
return false, err
|
||||
}
|
||||
|
||||
prevSettings := oldInbound.Settings
|
||||
oldInbound.Settings = string(newSettings)
|
||||
|
||||
emailShared, err := inboundSvc.emailUsedByOtherInbounds(email, inboundId)
|
||||
@@ -922,6 +976,8 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
||||
if push {
|
||||
if err1 := rt.DeleteUser(context.Background(), oldInbound, email); err1 != nil {
|
||||
logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
|
||||
} else {
|
||||
advancePushedInbound(rt, prevSettings, oldInbound)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -240,6 +240,40 @@ func (s *InboundService) GetNodeInboundTrafficTotals() (map[string][2]int64, err
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func adoptedWireChanged(c, snapIb *model.Inbound, adoptedSettings string) bool {
|
||||
return c.Settings != adoptedSettings ||
|
||||
c.Enable != snapIb.Enable ||
|
||||
c.Remark != snapIb.Remark ||
|
||||
c.SubSortIndex != normalizeSubSortIndex(snapIb.SubSortIndex) ||
|
||||
c.Listen != snapIb.Listen ||
|
||||
c.Port != snapIb.Port ||
|
||||
c.Protocol != snapIb.Protocol ||
|
||||
c.Total != snapIb.Total ||
|
||||
c.ExpiryTime != snapIb.ExpiryTime ||
|
||||
c.StreamSettings != snapIb.StreamSettings ||
|
||||
c.Sniffing != snapIb.Sniffing ||
|
||||
c.TrafficReset != snapIb.TrafficReset
|
||||
}
|
||||
|
||||
// adoptedWireInbound is the central inbound as it reads after adopting the
|
||||
// node-reported wire fields — the payload the reconcile fingerprint must track.
|
||||
func adoptedWireInbound(c, snapIb *model.Inbound, adoptedSettings string) *model.Inbound {
|
||||
a := *c
|
||||
a.Enable = snapIb.Enable
|
||||
a.Remark = snapIb.Remark
|
||||
a.SubSortIndex = normalizeSubSortIndex(snapIb.SubSortIndex)
|
||||
a.Listen = snapIb.Listen
|
||||
a.Port = snapIb.Port
|
||||
a.Protocol = snapIb.Protocol
|
||||
a.Total = snapIb.Total
|
||||
a.ExpiryTime = snapIb.ExpiryTime
|
||||
a.Settings = adoptedSettings
|
||||
a.StreamSettings = snapIb.StreamSettings
|
||||
a.Sniffing = snapIb.Sniffing
|
||||
a.TrafficReset = snapIb.TrafficReset
|
||||
return &a
|
||||
}
|
||||
|
||||
func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty bool) (bool, error) {
|
||||
if snap == nil || nodeID <= 0 {
|
||||
return false, nil
|
||||
@@ -396,6 +430,8 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
|
||||
|
||||
structuralChange := false
|
||||
|
||||
var adoptedInbounds []*model.Inbound
|
||||
|
||||
newInboundIDs := make(map[int]struct{})
|
||||
|
||||
snapTags := make(map[string]struct{}, len(snap.Inbounds))
|
||||
@@ -514,6 +550,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
|
||||
updates["sniffing"] = snapIb.Sniffing
|
||||
updates["traffic_reset"] = snapIb.TrafficReset
|
||||
updates["last_traffic_reset_time"] = snapIb.LastTrafficResetTime
|
||||
if adoptedWireChanged(c, snapIb, adoptedSettings) {
|
||||
adoptedInbounds = append(adoptedInbounds, adoptedWireInbound(c, snapIb, adoptedSettings))
|
||||
}
|
||||
}
|
||||
if !inGrace || (snapIb.Up+snapIb.Down) <= (c.Up+c.Down) {
|
||||
updates["up"] = snapIb.Up
|
||||
@@ -889,6 +928,18 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
|
||||
}
|
||||
committed = true
|
||||
|
||||
if len(adoptedInbounds) > 0 {
|
||||
if mgr := runtime.GetManager(); mgr != nil {
|
||||
if rt, rtErr := mgr.RuntimeFor(&nodeID); rtErr == nil {
|
||||
if rem, ok := rt.(*runtime.Remote); ok {
|
||||
for _, ib := range adoptedInbounds {
|
||||
rem.RecordAdoptedInbound(ib)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if p != nil {
|
||||
tree := snap.OnlineTree
|
||||
switch {
|
||||
|
||||
@@ -141,6 +141,81 @@ func TestNodeBulk_SmallAddPushesLive(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeUpdateInboundClientNoopSkipsRuntimeAndDirty(t *testing.T) {
|
||||
setupBulkDB(t)
|
||||
nodeID, fake := setupNodeRuntime(t)
|
||||
client := model.Client{
|
||||
ID: uuid.NewString(),
|
||||
Email: "noop@x",
|
||||
SubID: "sub-noop",
|
||||
Enable: true,
|
||||
CreatedAt: 111,
|
||||
UpdatedAt: 222,
|
||||
}
|
||||
ib := nodeInbound(t, nodeID, 30020, []model.Client{client})
|
||||
|
||||
svc := &ClientService{}
|
||||
inboundSvc := &InboundService{}
|
||||
if _, err := svc.UpdateInboundClient(inboundSvc, &model.Inbound{
|
||||
Id: ib.Id,
|
||||
Protocol: model.VLESS,
|
||||
Settings: clientsSettings(t, []model.Client{client}),
|
||||
}, client.Email); err != nil {
|
||||
t.Fatalf("UpdateInboundClient: %v", err)
|
||||
}
|
||||
|
||||
if got := fake.updateUser.Load(); got != 0 {
|
||||
t.Fatalf("no-op update streamed %d UpdateUser RPCs, want 0", got)
|
||||
}
|
||||
if _, _, dirty, _, err := (&NodeService{}).NodeSyncState(nodeID); err != nil {
|
||||
t.Fatalf("NodeSyncState: %v", err)
|
||||
} else if dirty {
|
||||
t.Fatal("no-op update must not mark the node dirty")
|
||||
}
|
||||
reloaded, err := inboundSvc.GetInbound(ib.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("GetInbound: %v", err)
|
||||
}
|
||||
if reloaded.Settings != ib.Settings {
|
||||
t.Fatal("no-op update rewrote inbound settings")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeUpdateInboundClientLivePushKeepsDirtyBackup(t *testing.T) {
|
||||
setupBulkDB(t)
|
||||
nodeID, fake := setupNodeRuntime(t)
|
||||
client := model.Client{
|
||||
ID: uuid.NewString(),
|
||||
Email: "edit@x",
|
||||
SubID: "sub-edit",
|
||||
Enable: true,
|
||||
CreatedAt: 111,
|
||||
UpdatedAt: 222,
|
||||
}
|
||||
ib := nodeInbound(t, nodeID, 30021, []model.Client{client})
|
||||
|
||||
edited := client
|
||||
edited.Comment = "changed"
|
||||
svc := &ClientService{}
|
||||
inboundSvc := &InboundService{}
|
||||
if _, err := svc.UpdateInboundClient(inboundSvc, &model.Inbound{
|
||||
Id: ib.Id,
|
||||
Protocol: model.VLESS,
|
||||
Settings: clientsSettings(t, []model.Client{edited}),
|
||||
}, client.Email); err != nil {
|
||||
t.Fatalf("UpdateInboundClient: %v", err)
|
||||
}
|
||||
|
||||
if got := fake.updateUser.Load(); got != 1 {
|
||||
t.Fatalf("edit streamed %d UpdateUser RPCs, want 1", got)
|
||||
}
|
||||
if _, _, dirty, _, err := (&NodeService{}).NodeSyncState(nodeID); err != nil {
|
||||
t.Fatalf("NodeSyncState: %v", err)
|
||||
} else if !dirty {
|
||||
t.Fatal("successful live update should keep node dirty as reconcile backup")
|
||||
}
|
||||
}
|
||||
|
||||
// TestNodeBulk_LargeDeleteFoldsToDirty: deleting more than the threshold from an
|
||||
// online node inbound must fold into a reconcile rather than per-client deletes.
|
||||
func TestNodeBulk_LargeDeleteFoldsToDirty(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user