From f8b72de9958c08031871c3a56e4a76cd32f82aa2 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Wed, 14 Oct 2026 00:37:00 +0200 Subject: [PATCH] fix(node-sync): write only changed client rows in the node traffic merge Every 5s node merge rewrote one client_traffics row and one node_client_traffics baseline per (inbound, client) pair the node reported, even when nothing had moved: a client is reported once per inbound it is on, and idle clients every tick. On top of that each inbound's SyncInbound wrote its own per-inbound values onto the shared clients record, so a client on a flow and a non-flow inbound (or on a WireGuard tunnel plus any other inbound) had its record flipped back and forth twice per merge. With 7 nodes and ~20k client links that is tens of thousands of statements per tick, all inside transactions on the single serialized writer, holding row locks for seconds. A renew of a client on 38 inbounds takes one writer turn per inbound, queued behind those merges each time, and outlived the 30s HTTP write timeout even though it committed; last_sub_fetch updates waited on the same locks. The merge now computes the row its UPDATE would leave, mirroring the SQL merge expressions, and skips the write when nothing changes; the baseline upsert is skipped when the node counter has not moved. The record's per-inbound fields get one owner each, so repeated syncs agree: flow is settled to the effective flow (lowest-id link carrying one) for clients whose links changed, and WireGuard/AmneziaWG peer fields come from the client's lowest-id tunnel. A repeated unchanged snapshot now writes no client row; one active client writes two. Verified beyond make verify: the CI Postgres lane, the merge and flow tests run against PostgreSQL 16, and make node-e2e. Closes #6815 --- .../web/service/client_flow_isolation_test.go | 32 ++++ internal/web/service/client_link.go | 130 ++++++++++++-- internal/web/service/inbound_node.go | 139 ++++++++------- .../service/node_client_traffic_sum_test.go | 162 ++++++++++++++++++ .../web/service/xray_wireguard_config_test.go | 9 +- 5 files changed, 389 insertions(+), 83 deletions(-) diff --git a/internal/web/service/client_flow_isolation_test.go b/internal/web/service/client_flow_isolation_test.go index 4ef25afeb..57a60f66d 100644 --- a/internal/web/service/client_flow_isolation_test.go +++ b/internal/web/service/client_flow_isolation_test.go @@ -324,3 +324,35 @@ func TestBulkAttach_PreservesVisionFlowWhenCanonicalColumnZeroed(t *testing.T) { t.Errorf("bulk-attached non-flow inbound must not receive Vision flow, got %#v", wsList) } } + +// A client detached from its last inbound has no effective flow left to settle +// its record to, so the record keeps the flow a later attach falls back on. +func TestRecordFlow_DetachedFromLastInboundKeepsFlow(t *testing.T) { + dbDir := t.TempDir() + t.Setenv("XUI_DB_FOLDER", dbDir) + dbtest.InitDB(t, filepath.Join(dbDir, "x-ui.db")) + + reality := &model.Inbound{Tag: "vless-reality", Enable: true, Port: 43001, Protocol: model.VLESS, StreamSettings: `{"network":"tcp","security":"reality"}`} + if err := database.GetDB().Create(reality).Error; err != nil { + t.Fatalf("create reality inbound: %v", err) + } + + svc := ClientService{} + const email = "detached@example.com" + const vision = "xtls-rprx-vision" + client := model.Client{Email: email, ID: "ce8d33df-3a64-4f10-8f9b-91c3a8e0c333", Enable: true, Flow: vision} + if err := svc.SyncInbound(nil, reality.Id, []model.Client{client}); err != nil { + t.Fatalf("SyncInbound(attach): %v", err) + } + if err := svc.SyncInbound(nil, reality.Id, nil); err != nil { + t.Fatalf("SyncInbound(detach): %v", err) + } + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + if rec.Flow != vision { + t.Errorf("record flow after detaching its last inbound = %q, want %q", rec.Flow, vision) + } +} diff --git a/internal/web/service/client_link.go b/internal/web/service/client_link.go index 0250127a8..3af9b354e 100644 --- a/internal/web/service/client_link.go +++ b/internal/web/service/client_link.go @@ -105,7 +105,11 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ tx = database.GetDB() } - if err := s.validateProtocolClients(tx, inboundId, clients, detachEmails, prune); err != nil { + var inbound model.Inbound + if err := tx.Select("protocol").Where("id = ?", inboundId).Take(&inbound).Error; err != nil { + return err + } + if err := s.validateProtocolClients(tx, inbound.Protocol, inboundId, clients, detachEmails, prune); err != nil { return err } @@ -137,6 +141,19 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ } } + isTunnel := inbound.Protocol == model.WireGuard || inbound.Protocol == model.AmneziaWG + var onLowerTunnel map[int]struct{} + if isTunnel && len(existing) > 0 { + ids := make([]int, 0, len(existing)) + for _, row := range existing { + ids = append(ids, row.Id) + } + var err error + if onLowerTunnel, err = clientsOnLowerTunnel(tx, inboundId, ids); err != nil { + return err + } + } + idByEmail := make(map[string]int, len(emails)) pending := make(map[string]*model.ClientRecord, len(emails)) toCreate := make([]*model.ClientRecord, 0, len(emails)) @@ -161,6 +178,8 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ before := *row applyClientRecordMerge(row, incoming) + _, outranked := onLowerTunnel[row.Id] + keepLinkOwnedFields(row, &before, isTunnel && !outranked) preservedUpdatedAt := max(incoming.UpdatedAt, row.UpdatedAt) row.UpdatedAt = preservedUpdatedAt @@ -222,7 +241,11 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ wantedIds = append(wantedIds, id) } - if err := s.reconcileInboundLinks(tx, inboundId, wantedFlow, wantedIds, detachEmails, prune); err != nil { + relinked, err := s.reconcileInboundLinks(tx, inboundId, wantedFlow, wantedIds, detachEmails, prune) + if err != nil { + return err + } + if err := settleRecordFlows(tx, relinked); err != nil { return err } @@ -260,19 +283,19 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ return nil } -// reconcileInboundLinks writes only the client_inbounds rows that differ. prune -// also removes links absent from wantedFlow, which only a full sync may do. -func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wantedFlow map[int]string, wantedIds []int, detachEmails []string, prune bool) error { +// reconcileInboundLinks writes only the client_inbounds rows that differ and returns +// their clients. prune also removes links absent from wantedFlow, as only a full sync may. +func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wantedFlow map[int]string, wantedIds []int, detachEmails []string, prune bool) ([]int, error) { var current []model.ClientInbound if prune { if err := tx.Where("inbound_id = ?", inboundId).Find(¤t).Error; err != nil { - return err + return nil, err } } else { for _, batch := range chunkInts(wantedIds, sqlInChunk) { var rows []model.ClientInbound if err := tx.Where("inbound_id = ? AND client_id IN ?", inboundId, batch).Find(&rows).Error; err != nil { - return err + return nil, err } current = append(current, rows...) } @@ -300,7 +323,7 @@ func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wanted for _, batch := range chunkStrings(detachEmails, sqlInChunk) { var ids []int if err := tx.Model(&model.ClientRecord{}).Where("email IN ?", batch).Pluck("id", &ids).Error; err != nil { - return err + return nil, err } for _, id := range ids { if _, keep := wantedFlow[id]; !keep { @@ -325,7 +348,7 @@ func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wanted for _, batch := range chunkInts(toDelete, sqlInChunk) { if err := tx.Where("inbound_id = ? AND client_id IN ?", inboundId, batch). Delete(&model.ClientInbound{}).Error; err != nil { - return err + return nil, err } } for flow, ids := range toUpdate { @@ -333,7 +356,7 @@ func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wanted if err := tx.Model(&model.ClientInbound{}). Where("inbound_id = ? AND client_id IN ?", inboundId, batch). Update("flow_override", flow).Error; err != nil { - return err + return nil, err } } } @@ -344,8 +367,87 @@ func (s *ClientService) reconcileInboundLinks(tx *gorm.DB, inboundId int, wanted Columns: []clause.Column{{Name: "client_id"}, {Name: "inbound_id"}}, DoUpdates: clause.AssignmentColumns([]string{"flow_override"}), }).CreateInBatches(toInsert, 200).Error; err != nil { + return nil, err + } + } + + relinked := append([]int(nil), toDelete...) + for _, ids := range toUpdate { + relinked = append(relinked, ids...) + } + for _, link := range toInsert { + relinked = append(relinked, link.ClientId) + } + return relinked, nil +} + +// keepLinkOwnedFields undoes the merge of the record fields one inbound cannot decide +// alone, or syncing a client's inbounds in turn rewrites the record back and forth. +func keepLinkOwnedFields(row, before *model.ClientRecord, ownsTunnelPeer bool) { + row.Flow = before.Flow + if ownsTunnelPeer { + return + } + row.PrivateKey = before.PrivateKey + row.PublicKey = before.PublicKey + row.AllowedIPs = before.AllowedIPs + row.PreSharedKey = before.PreSharedKey + row.KeepAlive = before.KeepAlive +} + +// clientsOnLowerTunnel returns the clients also attached to a WireGuard/AmneziaWG +// inbound with a lower id than inboundId: that tunnel owns their record's peer fields. +func clientsOnLowerTunnel(tx *gorm.DB, inboundId int, clientIds []int) (map[int]struct{}, error) { + out := make(map[int]struct{}) + for _, batch := range chunkInts(clientIds, sqlInChunk) { + var ids []int + if err := tx.Table("client_inbounds"). + Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). + Where("client_inbounds.client_id IN ? AND client_inbounds.inbound_id < ? AND inbounds.protocol IN ?", + batch, inboundId, []model.Protocol{model.WireGuard, model.AmneziaWG}). + Distinct().Pluck("client_inbounds.client_id", &ids).Error; err != nil { + return nil, err + } + for _, id := range ids { + out[id] = struct{}{} + } + } + return out, nil +} + +// settleRecordFlows sets the record flow of each still-linked client to its +// effective flow (EffectiveFlow's rule); an unlinked record keeps its own. +func settleRecordFlows(tx *gorm.DB, clientIds []int) error { + for _, batch := range chunkInts(clientIds, sqlInChunk) { + var links []model.ClientInbound + if err := tx.Select("client_id", "flow_override"). + Where("client_id IN ?", batch). + Order("inbound_id ASC"). + Find(&links).Error; err != nil { return err } + effective := make(map[int]string, len(batch)) + for _, link := range links { + if flow, seen := effective[link.ClientId]; !seen || (flow == "" && link.FlowOverride != "") { + effective[link.ClientId] = link.FlowOverride + } + } + var records []model.ClientRecord + if err := tx.Select("id", "flow").Where("id IN ?", batch).Find(&records).Error; err != nil { + return err + } + idsByFlow := make(map[string][]int) + for _, rec := range records { + if flow, linked := effective[rec.Id]; linked && flow != rec.Flow { + idsByFlow[flow] = append(idsByFlow[flow], rec.Id) + } + } + for flow, ids := range idsByFlow { + if err := tx.Model(&model.ClientRecord{}).Where("id IN ?", ids). + UpdateColumn("flow", flow).Error; err != nil { + return err + } + } } return nil } @@ -418,12 +520,8 @@ func (s *ClientService) ListForInboundBySubId(tx *gorm.DB, inboundId int, subId // validateProtocolClients runs inside the write every path linking clients to an inbound // shares, so a client the protocol's core config would refuse never reaches it. -func (s *ClientService) validateProtocolClients(tx *gorm.DB, inboundID int, changed []model.Client, detached []string, prune bool) error { - var inbound model.Inbound - if err := tx.Select("protocol").Where("id = ?", inboundID).Take(&inbound).Error; err != nil { - return err - } - switch inbound.Protocol { +func (s *ClientService) validateProtocolClients(tx *gorm.DB, protocol model.Protocol, inboundID int, changed []model.Client, detached []string, prune bool) error { + switch protocol { case model.TUIC: return s.validateTuicIdentities(tx, inboundID, changed, detached, prune) case model.MASQUE: diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index f481d2c05..2b05f8d15 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -224,6 +224,43 @@ func (s *InboundService) upsertNodeBaseline(tx *gorm.DB, nodeID int, email strin }).Create(&model.NodeClientTraffic{NodeId: nodeID, Email: email, Up: up, Down: down}).Error } +// mergeNodeClientTraffic folds one node-reported client into its central row. +// While frozen (a push pending or just landed) only the counters may move. +func (s *InboundService) mergeNodeClientTraffic(tx *gorm.DB, cs xray.ClientTraffic, frozen bool, deltaUp, deltaDown, now int64) error { + if frozen { + return tx.Exec( + fmt.Sprintf( + `UPDATE client_traffics + SET up = %s, down = %s, last_online = %s + WHERE email = ?`, + database.ClampedAddExpr("up"), + database.ClampedAddExpr("down"), + database.GreatestExpr("last_online", "?"), + ), + deltaUp, deltaDown, cs.LastOnline, cs.Email, + ).Error + } + return tx.Exec( + fmt.Sprintf( + `UPDATE client_traffics + SET up = %s, down = %s, enable = %s, total = ?, + expiry_time = %s, + reset = ?, reset_day = ?, reset_weekday = ?, last_online = %s + WHERE email = ?`, + database.ClampedAddExpr("up"), + database.ClampedAddExpr("down"), + database.ClientTrafficEnableMergeExpr(), + database.ClientTrafficExpiryMergeExpr(), + database.GreatestExpr("last_online", "?"), + ), + deltaUp, deltaDown, + cs.Enable, cs.ExpiryTime, cs.Total, now, deltaUp, deltaDown, + cs.Total, + cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetWeekday, + cs.LastOnline, cs.Email, + ).Error +} + // mergeActivationExpiry: master absolute wins; node may only activate when // master is unset/duration. Node auto-renew goes through nodeClientRenewed. func mergeActivationExpiry(existing, node int64) int64 { @@ -634,22 +671,25 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } - // Membership set for the rowExists checks below. Only the snapshot's emails - // are ever probed, so scope the lookup to those instead of plucking the whole - // client_traffics table (50k+ rows) on every node poll. - existingEmails := make(map[string]struct{}, len(snapEmailsAll)) + // The stored row of every reported email, wherever it is homed, so the merge + // can skip a write that would leave it unchanged. Homed rows share centralCS's pointer. + statByEmail := make(map[string]*xray.ClientTraffic, len(snapEmailsAll)) if len(snapEmailsAll) > 0 { snapEmailList := make([]string, 0, len(snapEmailsAll)) for email := range snapEmailsAll { snapEmailList = append(snapEmailList, email) } for _, batch := range chunkStrings(snapEmailList, sqliteMaxVars) { - var found []string - if err := db.Model(xray.ClientTraffic{}).Where("email IN ?", batch).Pluck("email", &found).Error; err != nil { + var found []xray.ClientTraffic + if err := db.Model(xray.ClientTraffic{}).Where("email IN ?", batch).Find(&found).Error; err != nil { return false, err } - for _, e := range found { - existingEmails[e] = struct{}{} + for i := range found { + if homed, ok := centralCSByEmail[found[i].Email]; ok { + statByEmail[found[i].Email] = homed + continue + } + statByEmail[found[i].Email] = &found[i] } } } @@ -1024,7 +1064,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } - if _, rowExists := existingEmails[cs.Email]; !rowExists { + if _, rowExists := statByEmail[cs.Email]; !rowExists { if dirty { continue } @@ -1062,7 +1102,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } centralCS[csKey{c.Id, cs.Email}] = row centralCSByEmail[cs.Email] = row - existingEmails[cs.Email] = struct{}{} + statByEmail[cs.Email] = row structuralChange = true if err := s.upsertNodeBaseline(tx, nodeID, cs.Email, canon.Up, canon.Down); err != nil { return false, err @@ -1121,64 +1161,34 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi existing.Total = cs.Total existing.ExpiryTime = cs.ExpiryTime existing.Reset = cs.Reset + existing.ResetDay = cs.ResetDay existing.ResetWeekday = cs.ResetWeekday existing.ResetCount = cs.ResetCount + existing.LastOnline = max(existing.LastOnline, cs.LastOnline) structuralChange = true - } else if clientFrozen { - // Push pending or just landed: only counters may move, the master - // keeps expiry/enable/total/reset. - if err := tx.Exec( - fmt.Sprintf( - `UPDATE client_traffics - SET up = %s, down = %s, last_online = %s - WHERE email = ?`, - database.ClampedAddExpr("up"), - database.ClampedAddExpr("down"), - database.GreatestExpr("last_online", "?"), - ), - deltaUp, deltaDown, cs.LastOnline, cs.Email, - ).Error; err != nil { - return false, err - } - if existing != nil { - existing.Up = clampTrafficCounter(existing.Up + deltaUp) - existing.Down = clampTrafficCounter(existing.Down + deltaDown) - } } else { - enableExpr := database.ClientTrafficEnableMergeExpr() - expiryExpr := database.ClientTrafficExpiryMergeExpr() - if err := tx.Exec( - fmt.Sprintf( - `UPDATE client_traffics - SET up = %s, down = %s, enable = %s, total = ?, - expiry_time = %s, - reset = ?, reset_day = ?, reset_weekday = ?, last_online = %s - WHERE email = ?`, - database.ClampedAddExpr("up"), - database.ClampedAddExpr("down"), - enableExpr, - expiryExpr, - database.GreatestExpr("last_online", "?"), - ), - deltaUp, deltaDown, - cs.Enable, cs.ExpiryTime, cs.Total, now, deltaUp, deltaDown, - cs.Total, - cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetWeekday, - cs.LastOnline, cs.Email, - ).Error; err != nil { - return false, err - } - if existing != nil { - priorExpiry := existing.ExpiryTime - if !cs.Enable && !nodeDisableIsStale(existing, cs, now, deltaUp, deltaDown) { - existing.Enable = false + // The row as the UPDATE below would leave it, computed from its + // pre-update values exactly as the SQL reads them. + row := statByEmail[cs.Email] + next := *row + next.Up = clampTrafficCounter(row.Up + deltaUp) + next.Down = clampTrafficCounter(row.Down + deltaDown) + next.LastOnline = max(row.LastOnline, cs.LastOnline) + if !clientFrozen { + if !cs.Enable && !nodeDisableIsStale(row, cs, now, deltaUp, deltaDown) { + next.Enable = false } - existing.ExpiryTime = mergeActivationExpiry(priorExpiry, cs.ExpiryTime) - existing.Up = clampTrafficCounter(existing.Up + deltaUp) - existing.Down = clampTrafficCounter(existing.Down + deltaDown) - existing.Total = cs.Total - existing.Reset = cs.Reset - existing.ResetWeekday = cs.ResetWeekday + next.ExpiryTime = mergeActivationExpiry(row.ExpiryTime, cs.ExpiryTime) + next.Total = cs.Total + next.Reset, next.ResetDay, next.ResetWeekday = cs.Reset, cs.ResetDay, cs.ResetWeekday + } + // A multi-attached client is reported once per inbound and an idle one + // every tick: rewriting an unchanged row only holds its lock longer. + if next != *row { + if err := s.mergeNodeClientTraffic(tx, cs, clientFrozen, deltaUp, deltaDown, now); err != nil { + return false, err + } + *row = next } } // A dip plus a lagging longer expiry mimics nodeClientRenewed and would @@ -1186,6 +1196,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if clientFrozen && seen && (canon.Up < base.Up || canon.Down < base.Down) { continue } + if seen && base == canon { + continue + } if err := s.upsertNodeBaseline(tx, nodeID, cs.Email, canon.Up, canon.Down); err != nil { return false, err } diff --git a/internal/web/service/node_client_traffic_sum_test.go b/internal/web/service/node_client_traffic_sum_test.go index ff804525f..dece8e23b 100644 --- a/internal/web/service/node_client_traffic_sum_test.go +++ b/internal/web/service/node_client_traffic_sum_test.go @@ -1,9 +1,14 @@ package service import ( + "encoding/json" "fmt" + "maps" "path/filepath" + "regexp" + "slices" "strings" + "sync" "testing" "gorm.io/gorm" @@ -471,6 +476,163 @@ func TestMultiAttach_SameNode_DivergentSiblings(t *testing.T) { assertUpDown(t, readTraffic(t, db, email), 100, 100, "equal siblings accrue the single increment") } +var clientWriteSQL = regexp.MustCompile(`(?i)^\s*(?:INSERT\s+INTO|UPDATE|DELETE\s+FROM)\s+["` + "`" + `]?(clients|client_inbounds|client_traffics|node_client_traffics)\b`) + +// countClientRowWrites returns, per client table, how many rows fn's +// INSERT/UPDATE/DELETE statements touched. +func countClientRowWrites(t *testing.T, fn func()) map[string]int64 { + t.Helper() + db := database.GetDB() + var mu sync.Mutex + rows := map[string]int64{} + record := func(tx *gorm.DB) { + if tx.Error != nil || tx.RowsAffected == 0 || tx.Statement == nil { + return + } + if m := clientWriteSQL.FindStringSubmatch(tx.Statement.SQL.String()); m != nil { + mu.Lock() + rows[m[1]] += tx.RowsAffected + mu.Unlock() + } + } + const name = "test:count_client_row_writes" + cb := db.Callback() + for _, err := range []error{ + cb.Create().After("gorm:create").Register(name, record), + cb.Update().After("gorm:update").Register(name, record), + cb.Delete().After("gorm:delete").Register(name, record), + cb.Raw().After("gorm:raw").Register(name, record), + } { + if err != nil { + t.Fatalf("register write callback: %v", err) + } + } + defer func() { + for _, err := range []error{ + cb.Create().Remove(name), cb.Update().Remove(name), + cb.Delete().Remove(name), cb.Raw().Remove(name), + } { + if err != nil { + t.Errorf("remove write callback: %v", err) + } + } + }() + fn() + return rows +} + +func TestNodeMerge_WritesOnlyChangedClients(t *testing.T) { + const clientCount = 20 + type snapInbound struct { + tag string + protocol model.Protocol + entry func(i int, email string) map[string]any + } + vlessEntry := func(flow string) func(int, string) map[string]any { + return func(i int, email string) map[string]any { + e := map[string]any{"email": email, "id": fmt.Sprintf("%08d-0000-4000-8000-000000000000", i), "subId": email, "enable": true} + if flow != "" { + e["flow"] = flow + } + return e + } + } + wgEntry := func(tunnel int) func(int, string) map[string]any { + return func(i int, email string) map[string]any { + return map[string]any{ + "email": email, "subId": email, "enable": true, + "privateKey": fmt.Sprintf("priv-%d-%d", tunnel, i), + "publicKey": fmt.Sprintf("pub-%d-%d", tunnel, i), + "allowedIPs": []string{fmt.Sprintf("10.%d.0.%d/32", tunnel, i+2)}, + "preSharedKey": fmt.Sprintf("psk-%d-%d", tunnel, i), + "keepAlive": 25 * tunnel, + } + } + } + + cases := []struct { + name string + inbounds []snapInbound + }{ + {"vless with and without flow", []snapInbound{ + {"n1-reality", model.VLESS, vlessEntry("xtls-rprx-vision")}, + {"n1-ws", model.VLESS, vlessEntry("")}, + }}, + {"wireguard and vless", []snapInbound{ + {"n1-wg", model.WireGuard, wgEntry(1)}, + {"n1-vless", model.VLESS, vlessEntry("")}, + }}, + {"two wireguard tunnels", []snapInbound{ + {"n1-wg-a", model.WireGuard, wgEntry(1)}, + {"n1-wg-b", model.WireGuard, wgEntry(2)}, + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + + emails := make([]string, clientCount) + stats := make([]xray.ClientTraffic, clientCount) + for i := range emails { + emails[i] = fmt.Sprintf("multi-%d", i) + stats[i] = xray.ClientTraffic{Email: emails[i], Up: 1000, Down: 2000, Enable: true, LastOnline: 5000} + } + settings := make([]string, len(tc.inbounds)) + for n, ib := range tc.inbounds { + entries := make([]map[string]any, clientCount) + for i, email := range emails { + entries[i] = ib.entry(i, email) + } + raw, err := json.Marshal(map[string]any{"clients": entries}) + if err != nil { + t.Fatalf("marshal settings: %v", err) + } + settings[n] = string(raw) + nodeID := 1 + if err := db.Create(&model.Inbound{ + UserId: 1, NodeID: &nodeID, Tag: ib.tag, Enable: true, + Port: 42000 + n, Protocol: ib.protocol, Settings: settings[n], + }).Error; err != nil { + t.Fatalf("create inbound %q: %v", ib.tag, err) + } + } + merge := func(stats []xray.ClientTraffic) { + t.Helper() + // A node copies a multi-attached client's one counter onto every inbound it is on. + snap := &runtime.TrafficSnapshot{} + for n, ib := range tc.inbounds { + snap.Inbounds = append(snap.Inbounds, &model.Inbound{ + Tag: ib.tag, Enable: true, Port: 42000 + n, Protocol: ib.protocol, + Settings: settings[n], ClientStats: slices.Clone(stats), + }) + } + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + } + + merge(stats) + if got := countClientRowWrites(t, func() { merge(stats) }); len(got) != 0 { + t.Errorf("repeating an unchanged snapshot wrote client rows %v, want none", got) + } + + active := slices.Clone(stats) + active[0].Up, active[0].Down, active[0].LastOnline = 1300, 2600, 6000 + got := countClientRowWrites(t, func() { merge(active) }) + want := map[string]int64{"client_traffics": 1, "node_client_traffics": 1} + if !maps.Equal(got, want) { + t.Errorf("one client's traffic tick wrote client rows %v, want %v", got, want) + } + ct := readTraffic(t, db, emails[0]) + assertUpDown(t, ct, 300, 600, "active client after its tick") + if ct.LastOnline != 6000 { + t.Errorf("last_online = %d, want 6000", ct.LastOnline) + } + }) + } +} + func TestDelClientStat_CleansNodeBaselines(t *testing.T) { db := initTrafficTestDB(t) svc := &InboundService{} diff --git a/internal/web/service/xray_wireguard_config_test.go b/internal/web/service/xray_wireguard_config_test.go index f38aa2367..61aed076c 100644 --- a/internal/web/service/xray_wireguard_config_test.go +++ b/internal/web/service/xray_wireguard_config_test.go @@ -94,12 +94,13 @@ func seedDualTunnelClient(t *testing.T, enabled bool, wgKeepAlive *int) string { wgInbound := &model.Inbound{Tag: "wg-dual", Enable: true, Port: 51823, Protocol: model.WireGuard, Settings: string(wgSettings)} awgInbound := &model.Inbound{Tag: "awg-dual", Enable: true, Port: 51824, Protocol: model.AmneziaWG, Settings: string(awgSettings)} - if err := db.Create(wgInbound).Error; err != nil { - t.Fatalf("create wg inbound: %v", err) - } + // The lower-id tunnel owns the shared record's peer fields: AmneziaWG first. if err := db.Create(awgInbound).Error; err != nil { t.Fatalf("create awg inbound: %v", err) } + if err := db.Create(wgInbound).Error; err != nil { + t.Fatalf("create wg inbound: %v", err) + } svc := ClientService{} if err := svc.SyncInbound(nil, wgInbound.Id, []model.Client{wgClient}); err != nil { @@ -202,7 +203,7 @@ func TestGetXrayConfigWireGuardUsesInboundLocalTunnelFields(t *testing.T) { t.Fatalf("read shared client: %v", err) } if shared.AllowedIPs != "10.8.1.5/32" || shared.PreSharedKey != "awg-psk" || shared.PublicKey != "pub-awg" || shared.KeepAlive != 25 { - t.Fatalf("test setup did not persist AmneziaWG last: allowedIPs=%q preSharedKey=%q publicKey=%q keepAlive=%d", shared.AllowedIPs, shared.PreSharedKey, shared.PublicKey, shared.KeepAlive) + t.Fatalf("test setup did not persist AmneziaWG identity: allowedIPs=%q preSharedKey=%q publicKey=%q keepAlive=%d", shared.AllowedIPs, shared.PreSharedKey, shared.PublicKey, shared.KeepAlive) } peers := wgPeerList(t, wgInboundEmittedSettings(t, "wg-dual"))