fix(node): remove a client a pushed inbound disables from the running core

UpdateInbound applies a master's push to the node's running core as the
difference between the stored and the pushed user sets, but both sets
dropped clients whose traffic row is off, and the same transaction had
already stored the pushed flags on those rows. A client the push
disabled therefore vanished from both sets, no RemoveUser was sent and
the node's Xray kept serving it.

The running set is now taken from the stored row's own enable flags,
so the client the push switches off is removed from every handler,
including a second inbound sharing the same traffic row.

internal/web/service/inbound.go

@@ -2281,7 +2281,7 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
 				}
 				if runtimeInbound != nil && running.Enable {
 					// Unbuildable means unknown running state: the rebuild below covers it.
-					runningInbound, _ = s.buildInboundForLocalRuntime(tx, &running)
+					runningInbound, _ = s.buildRunningInbound(tx, &running)
 				}
 				postCommitApply = func() {
 					if runningInbound != nil && applyInboundUserOps(rt, runningInbound, runtimeInbound) {
@@ -2438,6 +2438,16 @@ func (s *InboundService) buildInboundForNodePush(tx *gorm.DB, inbound *model.Inb
 // Strips disabled clients on top of the node payload. Safe only because the
 // target here is an in-memory Xray/mtg config, not another panel's database.
 func (s *InboundService) buildInboundForLocalRuntime(tx *gorm.DB, inbound *model.Inbound) (*model.Inbound, error) {
+	return s.stripDisabledClients(tx, inbound, true)
+}
+
+// The stored row as its handler holds it: its own enable flags only. A stats row
+// is shared by every inbound of the client and already holds what this push stored.
+func (s *InboundService) buildRunningInbound(tx *gorm.DB, stored *model.Inbound) (*model.Inbound, error) {
+	return s.stripDisabledClients(tx, stored, false)
+}
+
+func (s *InboundService) stripDisabledClients(tx *gorm.DB, inbound *model.Inbound, byStatsRow bool) (*model.Inbound, error) {
 	built, err := s.buildInboundForNodePush(tx, inbound)
 	if err != nil {
 		return nil, err
@@ -2459,9 +2469,11 @@ func (s *InboundService) buildInboundForLocalRuntime(tx *gorm.DB, inbound *model
 			emails = append(emails, email)
 		}
 	}
-	disabled, err := trafficDisabledEmails(tx, emails)
-	if err != nil {
-		return nil, err
+	var disabled map[string]struct{}
+	if byStatsRow {
+		if disabled, err = trafficDisabledEmails(tx, emails); err != nil {
+			return nil, err
+		}
 	}
 
 	trafficIDs := make(map[string]int)

internal/web/service/inbound_update_hot_users_test.go

@@ -133,6 +133,10 @@ func TestNodeInboundPushChangingOnlyClientsKeepsTheHandler(t *testing.T) {
 			if err := (&ClientService{}).SyncInbound(nil, stored.Id, tc.stored); err != nil {
 				t.Fatalf("SyncInbound: %v", err)
 			}
+			// Every client the API creates has a stats row, and the push rewrites its flag.
+			for _, c := range tc.stored {
+				mkTraffic(t, stored.Id, c.Email, 0, 0, 0, 0, c.Enable)
+			}
 			if tc.enabling {
 				if err := database.GetDB().Model(stored).Update("enable", false).Error; err != nil {
 					t.Fatalf("disable stored inbound: %v", err)
@@ -157,6 +161,43 @@ func TestNodeInboundPushChangingOnlyClientsKeepsTheHandler(t *testing.T) {
 	}
 }
 
+// A client's stats row is shared by all its inbounds, so the push for the first one
+// has switched it off by the time the second arrives: that handler holds the user still.
+func TestNodeInboundPushDisablingASharedClientRemovesItFromEveryHandler(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)}
+
+	setupConflictDB(t)
+	probe := &inboundOpsProbe{}
+	useTestRuntimeManager(t).SetLocalRuntimeOverride(probe)
+
+	tags := []string{"hot-users-1", "hot-users-2"}
+	for i, tag := range tags {
+		seedInboundConflict(t, tag, "0.0.0.0", 50081+i, model.VLESS, `{"network":"tcp"}`, clientsSettings(t, active))
+		if err := (&ClientService{}).SyncInbound(nil, loadInboundByTag(t, tag).Id, active); err != nil {
+			t.Fatalf("SyncInbound %s: %v", tag, err)
+		}
+	}
+	for _, c := range active {
+		mkTraffic(t, loadInboundByTag(t, tags[0]).Id, c.Email, 0, 0, 0, 0, true)
+	}
+
+	for _, tag := range tags {
+		pushed := *loadInboundByTag(t, tag)
+		pushed.Settings = clientsSettings(t, bDisabled)
+		if _, _, err := (&InboundService{FromNodeSync: true}).UpdateInbound(&pushed); err != nil {
+			t.Fatalf("UpdateInbound %s: %v", tag, err)
+		}
+	}
+	if got, want := probe.recorded(), []string{"remove-user b", "remove-user b"}; !slices.Equal(got, want) {
+		t.Fatalf("core operations = %q, want %q", got, want)
+	}
+}
+
 // 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) {