From a0989e0f4d0b57f61c10254c141e80188d33ef88 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sun, 5 Jul 2026 02:06:58 +0200 Subject: [PATCH] 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 --- internal/web/runtime/reconcile_skip_test.go | 254 ++++++++++++++++++ internal/web/runtime/remote.go | 44 ++- internal/web/service/bulk_clients_test.go | 10 +- internal/web/service/client_bulk.go | 18 ++ internal/web/service/client_inbound_apply.go | 64 ++++- internal/web/service/inbound_node.go | 51 ++++ .../web/service/node_bulk_dispatch_test.go | 75 ++++++ 7 files changed, 508 insertions(+), 8 deletions(-) diff --git a/internal/web/runtime/reconcile_skip_test.go b/internal/web/runtime/reconcile_skip_test.go index 74da152b0..5be854c6b 100644 --- a/internal/web/runtime/reconcile_skip_test.go +++ b/internal/web/runtime/reconcile_skip_test.go @@ -63,3 +63,257 @@ func TestReconcileInbound_SkipsUnchanged(t *testing.T) { t.Fatalf("absent-on-node reconcile pushes=%d, want 3", got) } } + +type nodeCallCounts struct { + adds atomic.Int32 + inboundUpdates atomic.Int32 + clientMutations atomic.Int32 +} + +func newCountingNodeServer(t *testing.T, clientsResp string) (*httptest.Server, *nodeCallCounts) { + t.Helper() + counts := &nodeCallCounts{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.Contains(r.URL.Path, "/panel/api/inbounds/add"): + counts.adds.Add(1) + case strings.Contains(r.URL.Path, "/panel/api/inbounds/update/"): + counts.inboundUpdates.Add(1) + case strings.Contains(r.URL.Path, "/panel/api/clients/"): + counts.clientMutations.Add(1) + _, _ = w.Write([]byte(clientsResp)) + return + } + _, _ = w.Write([]byte(`{"success":true}`)) + })) + t.Cleanup(srv.Close) + return srv, counts +} + +func perClientMutationCases(client model.Client) []struct { + name string + run func(*Remote, *model.Inbound) error +} { + return []struct { + name string + run func(*Remote, *model.Inbound) error + }{ + {"add", func(r *Remote, ib *model.Inbound) error { + return r.AddClient(context.Background(), ib, client) + }}, + {"delete", func(r *Remote, ib *model.Inbound) error { + return r.DeleteUser(context.Background(), ib, client.Email) + }}, + {"update", func(r *Remote, ib *model.Inbound) error { + return r.UpdateUser(context.Background(), ib, client.Email, client) + }}, + } +} + +// TestPerClientMutationsAloneDoNotSeedReconcileFingerprint: a per-client RPC +// only proves one client's slice converged, so without an explicit advance the +// dirty-reconcile backup must still send the full inbound. +func TestPerClientMutationsAloneDoNotSeedReconcileFingerprint(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + client := model.Client{ID: "11111111-1111-1111-1111-111111111111", Email: "a@x", SubID: "s", Enable: true} + for _, tt := range perClientMutationCases(client) { + t.Run(tt.name, func(t *testing.T) { + counts.inboundUpdates.Store(0) + counts.clientMutations.Store(0) + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + ib := &model.Inbound{Tag: "in-" + tt.name, Protocol: model.VLESS, Port: 443, Settings: `{"clients":[{"id":"11111111-1111-1111-1111-111111111111","email":"a@x","subId":"s","enable":true}]}`} + r.cacheSet(ib.Tag, 7) + + if err := tt.run(r, ib); err != nil { + t.Fatalf("%s client mutation: %v", tt.name, err) + } + if got := counts.clientMutations.Load(); got != 1 { + t.Fatalf("%s client mutation requests=%d, want 1", tt.name, got) + } + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || !pushed { + t.Fatalf("%s reconcile after unadvanced mutation: pushed=%v err=%v, want full push", tt.name, pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 1 { + t.Fatalf("%s reconcile sent %d full inbound updates, want 1", tt.name, got) + } + }) + } +} + +// TestAdvancePushedInboundEnablesReconcileSkip: when the node provably held the +// pre-edit payload and every per-client push succeeded, advancing the +// fingerprint lets the next reconcile skip the redundant full push. +func TestAdvancePushedInboundEnablesReconcileSkip(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + client := model.Client{ID: "11111111-1111-1111-1111-111111111111", Email: "a@x", SubID: "s", Enable: true} + prevByOp := map[string]string{ + "add": `{"clients":[]}`, + "delete": `{"clients":[{"email":"a@x","enable":true}]}`, + "update": `{"clients":[{"email":"a@x","enable":false}]}`, + } + newByOp := map[string]string{ + "add": `{"clients":[{"email":"a@x","enable":true}]}`, + "delete": `{"clients":[]}`, + "update": `{"clients":[{"email":"a@x","enable":true}]}`, + } + for _, tt := range perClientMutationCases(client) { + t.Run(tt.name, func(t *testing.T) { + counts.inboundUpdates.Store(0) + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + prevIb := &model.Inbound{Tag: "in-adv-" + tt.name, Protocol: model.VLESS, Port: 443, Settings: prevByOp[tt.name]} + ib := &model.Inbound{Tag: prevIb.Tag, Protocol: model.VLESS, Port: 443, Settings: newByOp[tt.name]} + r.cacheSet(ib.Tag, 7) + r.recordPushedInbound(prevIb) + + if err := tt.run(r, ib); err != nil { + t.Fatalf("%s client mutation: %v", tt.name, err) + } + r.AdvancePushedInbound(prevIb, ib) + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || pushed { + t.Fatalf("%s reconcile after advance: pushed=%v err=%v, want skip", tt.name, pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 0 { + t.Fatalf("%s reconcile sent %d full inbound updates, want 0", tt.name, got) + } + }) + } +} + +// TestAdvancePushedInboundRequiresMatchingPreviousFingerprint: if changes were +// folded to dirty (or an earlier push failed), the recorded fingerprint no +// longer matches the pre-edit payload; a later successful client push must not +// mask the pending reconcile. +func TestAdvancePushedInboundRequiresMatchingPreviousFingerprint(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + staleIb := &model.Inbound{Tag: "in-stale", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[{"email":"folded@x"}]}`} + prevIb := &model.Inbound{Tag: "in-stale", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[]}`} + ib := &model.Inbound{Tag: "in-stale", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[{"email":"a@x"}]}`} + r.cacheSet(ib.Tag, 7) + r.recordPushedInbound(staleIb) + + if err := r.UpdateUser(context.Background(), ib, "a@x", model.Client{Email: "a@x"}); err != nil { + t.Fatalf("client mutation: %v", err) + } + r.AdvancePushedInbound(prevIb, ib) + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || !pushed { + t.Fatalf("reconcile with unproven pre-state: pushed=%v err=%v, want full push", pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 1 { + t.Fatalf("reconcile sent %d full inbound updates, want 1", got) + } +} + +// TestAdoptedSerializationChainKeepsReconcileSkip: after a push the node +// re-serializes settings its own way and the master adopts that form back into +// its DB; stamping the adopted payload keeps edit->advance->skip alive instead +// of degrading every edit to a full reconcile push. +func TestAdoptedSerializationChainKeepsReconcileSkip(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + pushForm := &model.Inbound{Tag: "in-adopt", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[{"email":"a@x"}]}`} + adoptedForm := &model.Inbound{Tag: "in-adopt", Protocol: model.VLESS, Port: 443, Settings: "{\n \"clients\": [{\"email\": \"a@x\"}]\n}"} + edited := &model.Inbound{Tag: "in-adopt", Protocol: model.VLESS, Port: 443, Settings: "{\n \"clients\": [{\"comment\": \"c\", \"email\": \"a@x\"}]\n}"} + r.cacheSet(pushForm.Tag, 7) + + r.recordPushedInbound(pushForm) + r.RecordAdoptedInbound(adoptedForm) + if err := r.UpdateUser(context.Background(), edited, "a@x", model.Client{Email: "a@x"}); err != nil { + t.Fatalf("client mutation: %v", err) + } + r.AdvancePushedInbound(adoptedForm, edited) + if pushed, err := r.ReconcileInbound(context.Background(), edited, true); err != nil || pushed { + t.Fatalf("reconcile after adopted-form advance: pushed=%v err=%v, want skip", pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 0 { + t.Fatalf("full inbound updates=%d, want 0", got) + } +} + +func TestDeleteUserNotFoundHandling(t *testing.T) { + t.Run("envelope not-found counts as already deleted", func(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":false,"msg":"client not found"}`) + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + ib := &model.Inbound{Tag: "in-delete-missing", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[]}`} + r.cacheSet(ib.Tag, 7) + + if err := r.DeleteUser(context.Background(), ib, "missing@x"); err != nil { + t.Fatalf("DeleteUser missing client: %v", err) + } + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || !pushed { + t.Fatalf("reconcile after missing-client delete: pushed=%v err=%v, want full push", pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 1 { + t.Fatalf("reconcile sent %d full inbound updates, want 1", got) + } + }) + t.Run("http 404 from an old node stays an error", func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/panel/api/clients/") { + http.Error(w, "404 page not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true}`)) + })) + defer srv.Close() + + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + ib := &model.Inbound{Tag: "in-delete-404", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[]}`} + r.cacheSet(ib.Tag, 7) + + err := r.DeleteUser(context.Background(), ib, "missing@x") + if err == nil || !strings.Contains(err.Error(), "HTTP 404") { + t.Fatalf("DeleteUser against old node = %v, want HTTP 404 error", err) + } + }) +} + +// TestDelInboundDropsReconcileFingerprint: deleting an inbound must forget its +// fingerprint so a later same-tag inbound with identical content is re-pushed. +func TestDelInboundDropsReconcileFingerprint(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + ib := &model.Inbound{Tag: "in-del", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[]}`} + r.cacheSet(ib.Tag, 7) + + if pushed, err := r.ReconcileInbound(context.Background(), ib, false); err != nil || !pushed { + t.Fatalf("initial reconcile: pushed=%v err=%v, want push", pushed, err) + } + if err := r.DelInbound(context.Background(), ib); err != nil { + t.Fatalf("DelInbound: %v", err) + } + r.cacheSet(ib.Tag, 7) + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || !pushed { + t.Fatalf("reconcile after DelInbound: pushed=%v err=%v, want full push", pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 2 { + t.Fatalf("full inbound updates=%d, want 2", got) + } +} + +func TestUpdateInboundFallbackAddSeedsReconcileFingerprint(t *testing.T) { + srv, counts := newCountingNodeServer(t, `{"success":true}`) + + r := NewRemote(nodeForPlainServer(t, srv, "verify", "tok"), nil) + ib := &model.Inbound{Tag: "in-add-fallback", Protocol: model.VLESS, Port: 443, Settings: `{"clients":[]}`} + + if err := r.UpdateInbound(context.Background(), ib, ib); err != nil { + t.Fatalf("UpdateInbound fallback add: %v", err) + } + if got := counts.adds.Load(); got != 1 { + t.Fatalf("fallback add requests=%d, want 1", got) + } + if pushed, err := r.ReconcileInbound(context.Background(), ib, true); err != nil || pushed { + t.Fatalf("reconcile after fallback add: pushed=%v err=%v, want skip", pushed, err) + } + if got := counts.inboundUpdates.Load(); got != 0 { + t.Fatalf("reconcile sent %d full inbound updates, want 0", got) + } +} diff --git a/internal/web/runtime/remote.go b/internal/web/runtime/remote.go index 4a255840f..d6d0faf1b 100644 --- a/internal/web/runtime/remote.go +++ b/internal/web/runtime/remote.go @@ -68,6 +68,12 @@ type envelope struct { Obj json.RawMessage `json:"obj"` } +// remoteAPIError is a node-panel envelope failure (HTTP 200, success=false), +// distinct from transport/HTTP-status errors so callers can trust its message. +type remoteAPIError struct{ msg string } + +func (e *remoteAPIError) Error() string { return "remote: " + e.msg } + type Remote struct { node *model.Node @@ -275,7 +281,7 @@ func (r *Remote) do(ctx context.Context, method, path string, body any) (*envelo return nil, fmt.Errorf("decode envelope: %w", err) } if !env.Success { - return &env, fmt.Errorf("remote: %s", env.Msg) + return &env, &remoteAPIError{msg: env.Msg} } return &env, nil } @@ -340,6 +346,7 @@ func (r *Remote) cacheDel(tag string) { r.mu.Lock() defer r.mu.Unlock() delete(r.remoteIDByTag, tag) + delete(r.pushedFP, tag) } func (r *Remote) ListRemoteTags(ctx context.Context) ([]string, error) { @@ -407,6 +414,7 @@ func (r *Remote) AddInbound(ctx context.Context, ib *model.Inbound) error { r.cacheSet(created.Tag, created.Id) } } + r.recordPushedInbound(ib) return nil } @@ -436,6 +444,7 @@ func (r *Remote) UpdateInbound(ctx context.Context, oldIb, newIb *model.Inbound) r.cacheDel(oldIb.Tag) } r.cacheSet(newIb.Tag, id) + r.recordPushedInbound(newIb) return nil } @@ -457,10 +466,38 @@ func (r *Remote) ReconcileInbound(ctx context.Context, ib *model.Inbound, exists if err := r.UpdateInbound(ctx, ib, ib); err != nil { return false, err } + return true, nil +} + +// recordPushedInbound stamps the fingerprint after a full-payload push — the +// only operation that proves the node holds the entire wire payload. +func (r *Remote) recordPushedInbound(ib *model.Inbound) { + fp := wireFingerprint(wireInbound(ib, r.node.Id)) r.mu.Lock() r.pushedFP[ib.Tag] = fp r.mu.Unlock() - return true, nil +} + +// 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. +func (r *Remote) RecordAdoptedInbound(ib *model.Inbound) { + r.recordPushedInbound(ib) +} + +// 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 +// node held the exact pre-edit state; otherwise the stale fingerprint stays and +// the next reconcile re-sends the full inbound. +func (r *Remote) AdvancePushedInbound(prevIb, ib *model.Inbound) { + prevFP := wireFingerprint(wireInbound(prevIb, r.node.Id)) + nextFP := wireFingerprint(wireInbound(ib, r.node.Id)) + r.mu.Lock() + if r.pushedFP[ib.Tag] == prevFP { + r.pushedFP[ib.Tag] = nextFP + } + r.mu.Unlock() } // wireFingerprint hashes a wire payload so an unchanged inbound is cheap to detect. @@ -509,7 +546,8 @@ func (r *Remote) DeleteUser(ctx context.Context, ib *model.Inbound, email string if err == nil { return nil } - if strings.Contains(strings.ToLower(err.Error()), "not found") { + var apiErr *remoteAPIError + if errors.As(err, &apiErr) && strings.Contains(strings.ToLower(apiErr.msg), "not found") { return nil } return err diff --git a/internal/web/service/bulk_clients_test.go b/internal/web/service/bulk_clients_test.go index fb7708b8f..334ab5121 100644 --- a/internal/web/service/bulk_clients_test.go +++ b/internal/web/service/bulk_clients_test.go @@ -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 { diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index 7db659123..97512c041 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -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 diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index f1c0856be..7303c4364 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -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) } } } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index f3881b039..4bdbe6a17 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -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 { diff --git a/internal/web/service/node_bulk_dispatch_test.go b/internal/web/service/node_bulk_dispatch_test.go index c67f8c612..f52509fa2 100644 --- a/internal/web/service/node_bulk_dispatch_test.go +++ b/internal/web/service/node_bulk_dispatch_test.go @@ -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) {