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) {