mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-10-13 17:35:02 +03:00
fix(node): apply a pushed inbound's client changes in place instead of rebuilding it
A master pushes the whole inbound to its node whenever a client on it is auto-disabled, renewed or reset, and the node-sync reconcile re-sends every inbound whose payload changed (all of them after a master restart). The node's UpdateInbound answered every push with DelInbound + AddInbound, so Xray closed the listener and every open connection on that port dropped. On a node inbound with ~2,700 clients that was dozens of times a day. UpdateInbound now diffs the config Xray is running against the new one. When only settings.clients differ (vless, vmess, trojan, hysteria, masque) it applies per-user RemoveUser/AddUser through the runtime, and when nothing the core sees changed it does nothing. Any other change, or any per-user failure (a handler the core lost included), still rebuilds as before. AmneziaWG joins MTProto and TUIC on rt.UpdateInbound, whose in-place Ensure keeps the device and its peers; Del+Add tore it down. A pushed removal does not consult the node's own restartXrayOnClientDisable: it defaults to on, and the master already restarts its nodes from its own setting (restartRemoteNodesOnDisable). Nodes on an older release keep rebuilding until they update; the master side needs no change. Verified: the new service tests, TestXrayAPI_E2E_Users against a real core (pins the missing-user vs missing-handler error texts) and make node-e2e. make race not run: no goroutine, lock or shared state changed. Closes #6795
This commit is contained in:
@@ -26,6 +26,7 @@ import (
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/util/maskcompat"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"
|
||||
wgutil "github.com/mhsanaei/3x-ui/v3/internal/util/wireguard"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -1964,6 +1965,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
if err := tx.First(stored, inbound.Id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
running := *stored
|
||||
oldInbound = stored
|
||||
// The form posts back the clients and enable it loaded; both have their
|
||||
// own endpoints, so only a master's push may change them here.
|
||||
@@ -2113,7 +2115,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
}
|
||||
if !push {
|
||||
needRestart = true
|
||||
} else if oldProtocol == model.MTProto || oldInbound.Protocol == model.MTProto || oldProtocol == model.TUIC || oldInbound.Protocol == model.TUIC {
|
||||
} else if servedOutsideXray(oldProtocol) || servedOutsideXray(oldInbound.Protocol) {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
oldSnapshot.Protocol = oldProtocol
|
||||
@@ -2127,7 +2129,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
pushable = false
|
||||
}
|
||||
}
|
||||
newProtocolIsSidecar := oldInbound.Protocol == model.MTProto || oldInbound.Protocol == model.TUIC
|
||||
newProtocolIsSidecar := servedOutsideXray(oldInbound.Protocol)
|
||||
if pushable {
|
||||
postCommitApply = func() {
|
||||
if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, payload); err2 == nil {
|
||||
@@ -2143,7 +2145,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
} else {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
var runtimeInbound *model.Inbound
|
||||
var runtimeInbound, runningInbound *model.Inbound
|
||||
if inbound.Enable {
|
||||
var err2 error
|
||||
runtimeInbound, err2 = s.buildInboundForLocalRuntime(tx, oldInbound)
|
||||
@@ -2152,7 +2154,15 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
if runtimeInbound != nil && running.Enable {
|
||||
// Unbuildable means unknown running state: the rebuild below covers it.
|
||||
runningInbound, _ = s.buildInboundForLocalRuntime(tx, &running)
|
||||
}
|
||||
postCommitApply = func() {
|
||||
if runningInbound != nil && applyInboundUserOps(rt, runningInbound, runtimeInbound) {
|
||||
logger.Debug("Inbound clients updated in place on", rt.Name(), ":", oldInbound.Tag)
|
||||
return
|
||||
}
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
|
||||
logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
|
||||
}
|
||||
@@ -2217,6 +2227,41 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
return inbound, needRestart, nil
|
||||
}
|
||||
|
||||
// servedOutsideXray: these protocols run under their own manager, whose update
|
||||
// reconfigures in place where a Del+Add would drop every live session.
|
||||
func servedOutsideXray(protocol model.Protocol) bool {
|
||||
return protocol == model.MTProto || protocol == model.TUIC || protocol == model.AmneziaWG
|
||||
}
|
||||
|
||||
// applyInboundUserOps reports whether per-user core ops moved running to next, so
|
||||
// its listener and live connections survive; false means rebuild the handler.
|
||||
func applyInboundUserOps(rt runtime.Runtime, running, next *model.Inbound) bool {
|
||||
diff, ok := xray.InboundUserOps(running.GenXrayInboundConfig(), next.GenXrayInboundConfig())
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
ctx := context.Background()
|
||||
for _, u := range diff.RemovedUsers {
|
||||
if err := rt.RemoveUser(ctx, next, u.Email); err != nil && !xray.IsMissingUserErr(err) {
|
||||
logger.Debug("Unable to remove client", u.Email, "in place on", rt.Name(), ":", err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
for _, u := range diff.AddedUsers {
|
||||
err := rt.AddUser(ctx, next, u.User)
|
||||
if xray.IsUserExistsErr(err) {
|
||||
if err = rt.RemoveUser(ctx, next, u.Email); err == nil {
|
||||
err = rt.AddUser(ctx, next, u.User)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
logger.Debug("Unable to add client", u.Email, "in place on", rt.Name(), ":", err)
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// A node mirrors this payload into its own DB, so every client must survive:
|
||||
// filtering one out makes the node delete it, and the master then mirrors that.
|
||||
func (s *InboundService) buildInboundForNodePush(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) {
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||
)
|
||||
|
||||
// inboundOpsProbe records every handler and user operation the panel sends the
|
||||
// core, in order, and fails a remove or the first add the way xray does.
|
||||
type inboundOpsProbe struct {
|
||||
fakeNodeRuntime
|
||||
mu sync.Mutex
|
||||
ops []string
|
||||
removeErr error
|
||||
addErr error
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) record(op string) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.ops = append(p.ops, op)
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) AddInbound(context.Context, *model.Inbound) error {
|
||||
p.record("add-inbound")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) DelInbound(context.Context, *model.Inbound) error {
|
||||
p.record("del-inbound")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) UpdateInbound(context.Context, *model.Inbound, *model.Inbound) error {
|
||||
p.record("update-inbound")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) AddUser(_ context.Context, _ *model.Inbound, user map[string]any) error {
|
||||
email, _ := user["email"].(string)
|
||||
p.record("add-user " + email)
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
err := p.addErr
|
||||
p.addErr = nil
|
||||
return err
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) RemoveUser(_ context.Context, _ *model.Inbound, email string) error {
|
||||
p.record("remove-user " + email)
|
||||
return p.removeErr
|
||||
}
|
||||
|
||||
func (p *inboundOpsProbe) recorded() []string {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return append([]string(nil), p.ops...)
|
||||
}
|
||||
|
||||
func hotUsersClient(email, id string, enable bool) model.Client {
|
||||
return model.Client{Email: email, ID: id, Enable: enable, CreatedAt: 1, UpdatedAt: 1}
|
||||
}
|
||||
|
||||
// A master pushes the whole inbound for every client it disables or renews; the
|
||||
// node used to delete and re-add the handler, dropping every live connection.
|
||||
func TestNodeInboundPushChangingOnlyClientsKeepsTheHandler(t *testing.T) {
|
||||
const (
|
||||
idA = "5f2eb9d6-3a2f-4a55-9812-6ea1e2f7a301"
|
||||
idB = "5f2eb9d6-3a2f-4a55-9812-6ea1e2f7a302"
|
||||
)
|
||||
active := []model.Client{hotUsersClient("a", idA, true), hotUsersClient("b", idB, true)}
|
||||
bDisabled := []model.Client{hotUsersClient("a", idA, true), hotUsersClient("b", idB, false)}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
stored, pushed []model.Client
|
||||
portShift int
|
||||
enabling bool
|
||||
removeErr error
|
||||
addErr error
|
||||
wantOps []string
|
||||
}{
|
||||
{
|
||||
name: "disabling one client removes only that user", stored: active, pushed: bDisabled,
|
||||
wantOps: []string{"remove-user b"},
|
||||
},
|
||||
{
|
||||
name: "renewing a disabled client adds only that user", stored: bDisabled, pushed: active,
|
||||
wantOps: []string{"add-user b"},
|
||||
},
|
||||
{
|
||||
name: "a port change still rebuilds the handler", stored: active, pushed: active, portShift: 1,
|
||||
wantOps: []string{"del-inbound", "add-inbound"},
|
||||
},
|
||||
{
|
||||
name: "enabling a disabled inbound adds its handler", stored: active, pushed: active, enabling: true,
|
||||
wantOps: []string{"del-inbound", "add-inbound"},
|
||||
},
|
||||
{
|
||||
name: "a user the core no longer holds is not a rebuild", stored: active, pushed: bDisabled,
|
||||
removeErr: errors.New("failed to remove user: User b not found."),
|
||||
wantOps: []string{"remove-user b"},
|
||||
},
|
||||
{
|
||||
name: "a handler the core lost is rebuilt", stored: active, pushed: bDisabled,
|
||||
removeErr: errors.New("failed to remove user: failed to get handler: hot-users > handler not found: hot-users"),
|
||||
wantOps: []string{"remove-user b", "del-inbound", "add-inbound"},
|
||||
},
|
||||
{
|
||||
name: "a user the core already holds is replaced", stored: bDisabled, pushed: active,
|
||||
addErr: errors.New("User b already exists."),
|
||||
wantOps: []string{"add-user b", "remove-user b", "add-user b"},
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
setupConflictDB(t)
|
||||
// A node keeps this default; the master's own setting decides node restarts.
|
||||
setRestartOnClientDisable(t, true)
|
||||
probe := &inboundOpsProbe{removeErr: tc.removeErr, addErr: tc.addErr}
|
||||
useTestRuntimeManager(t).SetLocalRuntimeOverride(probe)
|
||||
|
||||
seedInboundConflict(t, "hot-users", "0.0.0.0", 50081, model.VLESS, `{"network":"tcp"}`, clientsSettings(t, tc.stored))
|
||||
stored := loadInboundByTag(t, "hot-users")
|
||||
if err := (&ClientService{}).SyncInbound(nil, stored.Id, tc.stored); err != nil {
|
||||
t.Fatalf("SyncInbound: %v", err)
|
||||
}
|
||||
if tc.enabling {
|
||||
if err := database.GetDB().Model(stored).Update("enable", false).Error; err != nil {
|
||||
t.Fatalf("disable stored inbound: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
pushed := *stored
|
||||
pushed.Settings = clientsSettings(t, tc.pushed)
|
||||
pushed.Port += tc.portShift
|
||||
pushed.Enable = true
|
||||
_, needRestart, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed)
|
||||
if err != nil {
|
||||
t.Fatalf("UpdateInbound: %v", err)
|
||||
}
|
||||
if got := probe.recorded(); !slices.Equal(got, tc.wantOps) {
|
||||
t.Fatalf("core operations = %q, want %q", got, tc.wantOps)
|
||||
}
|
||||
if needRestart {
|
||||
t.Fatal("needRestart = true, want the push applied without restarting the core")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// AmneziaWG has its own in-place reconfigure; deleting and re-adding the inbound
|
||||
// rebuilt the device and dropped every peer's session on each pushed client change.
|
||||
func TestNodeAmneziaWGPushReconfiguresTheDeviceInPlace(t *testing.T) {
|
||||
setupConflictDB(t)
|
||||
probe := &inboundOpsProbe{}
|
||||
useTestRuntimeManager(t).SetLocalRuntimeOverride(probe)
|
||||
|
||||
settings := awgRelayWindowSettings(t, "hot-awg")
|
||||
seedInboundConflict(t, "hot-awg", "0.0.0.0", 51830, model.AmneziaWG, "", settings)
|
||||
stored := loadInboundByTag(t, "hot-awg")
|
||||
clients, err := (&InboundService{}).GetClients(stored)
|
||||
if err != nil {
|
||||
t.Fatalf("GetClients: %v", err)
|
||||
}
|
||||
if err := (&ClientService{}).SyncInbound(nil, stored.Id, clients); err != nil {
|
||||
t.Fatalf("SyncInbound: %v", err)
|
||||
}
|
||||
|
||||
pushed := *stored
|
||||
pushed.Settings = strings.Replace(settings, `"enable":true`, `"enable":false`, 1)
|
||||
if _, _, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed); err != nil {
|
||||
t.Fatalf("UpdateInbound: %v", err)
|
||||
}
|
||||
if got, want := probe.recorded(), []string{"update-inbound"}; !slices.Equal(got, want) {
|
||||
t.Fatalf("runtime operations = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// A master re-sends every inbound after it restarts; an unchanged one must not
|
||||
// touch the core, even for a protocol whose users have no per-user API path.
|
||||
func TestNodeInboundUnchangedPushLeavesTheCoreAlone(t *testing.T) {
|
||||
setupConflictDB(t)
|
||||
probe := &inboundOpsProbe{}
|
||||
useTestRuntimeManager(t).SetLocalRuntimeOverride(probe)
|
||||
|
||||
clients := []model.Client{{Email: "ss-a", Password: "a-pass", Enable: true, CreatedAt: 1, UpdatedAt: 1}}
|
||||
settings := `{"method":"chacha20-ietf-poly1305","network":"tcp,udp","clients":[{"method":"chacha20-ietf-poly1305","password":"a-pass","email":"ss-a","enable":true,"created_at":1,"updated_at":1}]}`
|
||||
seedInboundConflict(t, "hot-ss", "0.0.0.0", 50082, model.Shadowsocks, `{"network":"tcp"}`, settings)
|
||||
stored := loadInboundByTag(t, "hot-ss")
|
||||
if err := (&ClientService{}).SyncInbound(nil, stored.Id, clients); err != nil {
|
||||
t.Fatalf("SyncInbound: %v", err)
|
||||
}
|
||||
|
||||
pushed := *stored
|
||||
if _, _, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed); err != nil {
|
||||
t.Fatalf("UpdateInbound: %v", err)
|
||||
}
|
||||
if got := probe.recorded(); len(got) != 0 {
|
||||
t.Fatalf("core operations = %q, want none for an unchanged inbound", got)
|
||||
}
|
||||
}
|
||||
@@ -443,6 +443,16 @@ func IsMissingHandlerErr(err error) bool {
|
||||
strings.Contains(msg, "not enough information")
|
||||
}
|
||||
|
||||
// IsMissingUserErr reports whether err is xray removing a user the inbound does
|
||||
// not hold; a missing handler fails the same call and is not one.
|
||||
func IsMissingUserErr(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "not found") && !strings.Contains(msg, "failed to get handler")
|
||||
}
|
||||
|
||||
// IsExistingTagErr reports whether err is xray's response to adding a handler
|
||||
// whose tag is already taken by a running handler.
|
||||
func IsExistingTagErr(err error) bool {
|
||||
|
||||
@@ -145,7 +145,7 @@ func panelUser(email string, fields map[string]any) map[string]any {
|
||||
|
||||
// TestXrayAPI_E2E_Users runs the panel's add/remove user surface against a live
|
||||
// core for every protocol it builds accounts for, pinning the error texts
|
||||
// IsUserExistsErr and IsMissingHandlerErr match on.
|
||||
// IsUserExistsErr, IsMissingHandlerErr and IsMissingUserErr match on.
|
||||
func TestXrayAPI_E2E_Users(t *testing.T) {
|
||||
c := startE2ECore(t, []any{
|
||||
map[string]any{
|
||||
@@ -248,6 +248,9 @@ func TestXrayAPI_E2E_Users(t *testing.T) {
|
||||
if !IsMissingHandlerErr(err) {
|
||||
t.Fatalf("missing-user error not matched by IsMissingHandlerErr: %q", err)
|
||||
}
|
||||
if !IsMissingUserErr(err) {
|
||||
t.Fatalf("missing-user error not matched by IsMissingUserErr: %q", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -259,6 +262,13 @@ func TestXrayAPI_E2E_Users(t *testing.T) {
|
||||
if !IsMissingHandlerErr(err) {
|
||||
t.Fatalf("unknown-tag error not matched by IsMissingHandlerErr: %q", err)
|
||||
}
|
||||
err = c.api.RemoveUser("no-such-tag", "e2e-ghost")
|
||||
if err == nil {
|
||||
t.Fatal("RemoveUser on an unknown tag must fail")
|
||||
}
|
||||
if IsMissingUserErr(err) {
|
||||
t.Fatalf("a missing handler must not read as a missing user: %q", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -221,6 +221,19 @@ func droppedClients(oldIb, newIb *InboundConfig) []UserOp {
|
||||
return dropped
|
||||
}
|
||||
|
||||
// InboundUserOps returns the per-user ops that bring a running inbound from oldIb
|
||||
// to newIb; ok is false when only rebuilding its handler can apply the change.
|
||||
func InboundUserOps(oldIb, newIb *InboundConfig) (*HotDiff, bool) {
|
||||
diff := &HotDiff{}
|
||||
if inboundEqualNormalized(oldIb, newIb) {
|
||||
return diff, true
|
||||
}
|
||||
if !diffInboundUsers(oldIb, newIb, diff) {
|
||||
return nil, false
|
||||
}
|
||||
return diff, true
|
||||
}
|
||||
|
||||
var userDiffableProtocols = map[string]struct{}{"vless": {}, "vmess": {}, "trojan": {}, "hysteria": {}, "masque": {}}
|
||||
|
||||
// diffInboundUsers emits per-user AlterInbound ops when two same-tag inbounds
|
||||
|
||||
Reference in New Issue
Block a user