diff --git a/internal/database/dialect.go b/internal/database/dialect.go index d142eb209..6aa009026 100644 --- a/internal/database/dialect.go +++ b/internal/database/dialect.go @@ -37,9 +37,32 @@ func GreatestExpr(a, b string) string { return fmt.Sprintf("MAX(%s, %s)", a, b) } +// ClientTrafficEnableMergeExpr: placeholders nodeEnable, nodeExpiry, nodeTotal, +// now, deltaUp, deltaDown. Mirrors nodeDisableIsStale (#6228 / #4917). func ClientTrafficEnableMergeExpr() string { if IsPostgres() { - return "CASE WHEN ?::boolean THEN enable::boolean ELSE false END" + return `CASE + WHEN ?::boolean THEN enable::boolean + WHEN (expiry_time <> CAST(? AS BIGINT) OR total <> CAST(? AS BIGINT)) + AND (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) + AND (total <= 0 OR up + ? + down + ? < total) THEN enable::boolean + ELSE false + END` } - return "CASE WHEN ? THEN enable ELSE 0 END" + return `CASE + WHEN ? THEN enable + WHEN (expiry_time <> CAST(? AS BIGINT) OR total <> CAST(? AS BIGINT)) + AND (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) + AND (total <= 0 OR up + ? + down + ? < total) THEN enable + ELSE 0 + END` +} + +// ClientTrafficExpiryMergeExpr: placeholder nodeExpiry once. Master absolute is +// kept; CAST avoids Postgres int4 inference on ms timestamps. +func ClientTrafficExpiryMergeExpr() string { + return `CASE + WHEN expiry_time > 0 THEN expiry_time + ELSE CAST(? AS BIGINT) + END` } diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index c38368693..f130b1283 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -363,6 +363,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy return nil } + justPushed := false if n.ConfigDirty { reconcileCtx, reconcileCancel := context.WithTimeout(context.Background(), nodeReconcileTimeout) reconcileErr := j.inboundService.ReconcileNode(reconcileCtx, rt, n) @@ -377,6 +378,9 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy logger.Warningf("node traffic sync: clear dirty for %s failed: %v", n.Name, clearErr) } j.structural.set() + // The snapshot below may still predate the push we just made, so its + // lagging lifecycle values must not merge back this tick (#6228). + justPushed = true } } @@ -407,7 +411,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy } } } - changed, err := j.inboundService.SetRemoteTraffic(n.Id, snap, dirty) + changed, err := j.inboundService.SetRemoteTraffic(n.Id, snap, dirty, justPushed) if err != nil { logger.Warningf("node traffic sync: merge for %s failed: %v", n.Name, err) return nil diff --git a/internal/web/service/client_group_node_sync_test.go b/internal/web/service/client_group_node_sync_test.go index 23e4874ea..9a37f3f10 100644 --- a/internal/web/service/client_group_node_sync_test.go +++ b/internal/web/service/client_group_node_sync_test.go @@ -63,7 +63,7 @@ func TestSetRemoteTraffic_PreservesPanelLocalGroupAndComment(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } diff --git a/internal/web/service/client_locks.go b/internal/web/service/client_locks.go index b9d7d10ee..3788f4be1 100644 --- a/internal/web/service/client_locks.go +++ b/internal/web/service/client_locks.go @@ -8,6 +8,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/xray" "gorm.io/gorm" ) @@ -232,3 +233,96 @@ func stripTombstonedClients(settings string) (string, bool) { } return string(b), true } + +// liftClientLifecycleInSettings rewrites adopted settings from master traffic +// so a lagging node blob cannot store pre-extension expiry/enable (#6228). +func liftClientLifecycleInSettings(settings string, trafficByEmail map[string]*xray.ClientTraffic) (string, bool) { + if settings == "" || len(trafficByEmail) == 0 { + return settings, false + } + var parsed map[string]any + if err := json.Unmarshal([]byte(settings), &parsed); err != nil { + return settings, false + } + clients, _ := parsed["clients"].([]any) + if len(clients) == 0 { + return settings, false + } + changed := false + for i := range clients { + cm, ok := clients[i].(map[string]any) + if !ok { + continue + } + email, _ := cm["email"].(string) + if email == "" { + continue + } + tr := trafficByEmail[email] + if tr == nil { + continue + } + nodeExpiry, hasExpiry := jsonClientInt64(cm["expiryTime"]) + if !hasExpiry { + continue + } + merged := mergeActivationExpiry(tr.ExpiryTime, nodeExpiry) + if merged != nodeExpiry { + cm["expiryTime"] = merged + changed = true + } + // tr is the already-merged master row, authoritative in both directions: + // a lagging blob must not re-enable a disabled client either (#4917). + if nodeEnable, _ := cm["enable"].(bool); nodeEnable != tr.Enable { + cm["enable"] = tr.Enable + changed = true + } + clients[i] = cm + } + if !changed { + return settings, false + } + parsed["clients"] = clients + b, err := json.MarshalIndent(parsed, "", " ") + if err != nil { + return settings, false + } + return string(b), true +} + +// settingsClientAbsoluteExpiries indexes the absolute (>0) expiryTime of every +// client in a settings blob. Never nil, so callers can cache it per inbound. +func settingsClientAbsoluteExpiries(settings string) map[string]int64 { + out := map[string]int64{} + if settings == "" { + return out + } + var parsed map[string]any + if err := json.Unmarshal([]byte(settings), &parsed); err != nil { + return out + } + clients, _ := parsed["clients"].([]any) + for _, c := range clients { + cm, ok := c.(map[string]any) + if !ok { + continue + } + email, _ := cm["email"].(string) + if email == "" { + continue + } + if exp, has := jsonClientInt64(cm["expiryTime"]); has && exp > 0 { + out[email] = exp + } + } + return out +} + +func jsonClientInt64(v any) (int64, bool) { + // json.Unmarshal into map[string]any yields float64 for numbers. + n, ok := v.(float64) + if !ok { + return 0, false + } + return int64(n), true +} diff --git a/internal/web/service/client_sync_orphan_test.go b/internal/web/service/client_sync_orphan_test.go index 57d06de35..bb057e974 100644 --- a/internal/web/service/client_sync_orphan_test.go +++ b/internal/web/service/client_sync_orphan_test.go @@ -49,7 +49,7 @@ func TestSyncOrphanSurvivesMergeUntilGraceElapses(t *testing.T) { t.Fatalf("setup: clients=%d client_traffics=%d, want 1/1", rec, traf) } - if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false, false); err != nil { t.Fatalf("orphaning merge: %v", err) } if rec, traf := countClientRows(t, db, email); rec != 1 || traf != 1 { @@ -99,7 +99,7 @@ func TestSyncOrphanMarkClearedOnReattach(t *testing.T) { syncNodeWithSettings(t, svc, 1, "n1-in", settings, xray.ClientTraffic{Email: email, Up: 5, Down: 5, Enable: true}) - if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false, false); err != nil { t.Fatalf("orphaning merge: %v", err) } if readOrphanMark(t, db, email) <= 0 { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 44909ff4f..0fc458b27 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -215,27 +215,71 @@ func (s *InboundService) upsertNodeBaseline(tx *gorm.DB, nodeID int, email strin }).Create(&model.NodeClientTraffic{NodeId: nodeID, Email: email, Up: up, Down: down}).Error } -// mergeActivationExpiry reconciles a node-reported client expiry with the value -// already stored on the master. "Start after first connect" persists a negative -// duration that each node converts to an absolute deadline (now+duration) the -// first time the client connects there. The per-email client_traffics row is -// shared across every node, so a node that has not yet seen a first connection -// keeps reporting the negative duration — which must never reset a deadline -// another node already activated. -// -// A node may legitimately move an already-activated deadline forward (traffic -// reset / auto-renew extends it), so any positive node value is still adopted — -// only an un-activated (<= 0) value is rejected once an absolute deadline -// exists. Kept in lockstep with the SQL CASE in setRemoteTrafficLocked. +// 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 { - if existing > 0 && node <= 0 { + if existing > 0 { return existing } return node } +// masterLimitsAllowClient reports whether the master's own deadline and quota +// (including this tick's deltas) still permit the client. +func masterLimitsAllowClient(master *xray.ClientTraffic, now, deltaUp, deltaDown int64) bool { + if master == nil { + return false + } + if master.ExpiryTime > 0 && master.ExpiryTime <= now { + return false + } + if master.Total > 0 && master.Up+deltaUp+master.Down+deltaDown >= master.Total { + return false + } + return true +} + +// nodeDisableIsStale reports an enable=false the node decided against limits the +// master has since changed, so it must not latch back (#6228 / #4917). +func nodeDisableIsStale(master *xray.ClientTraffic, node xray.ClientTraffic, now, deltaUp, deltaDown int64) bool { + if master == nil { + return false + } + // Matching limits mean the node judged the client on the master's own terms: + // that verdict is genuine and still latches, as #4917 requires. + if node.ExpiryTime == master.ExpiryTime && node.Total == master.Total { + return false + } + return masterLimitsAllowClient(master, now, deltaUp, deltaDown) +} + +func clampTrafficCounter(v int64) int64 { + if v > database.TrafficMax { + return database.TrafficMax + } + if v < 0 { + return 0 + } + return v +} + +// applyMasterClientLifecycle overlays the already-merged master row onto a +// node-reported client for SyncInbound (#6228). +func applyMasterClientLifecycle(c *model.Client, master *xray.ClientTraffic, cs *xray.ClientTraffic) { + if master == nil { + // No central row to speak for the client: the node's own latch is all + // there is, and it may only disable. + if cs != nil && !cs.Enable { + c.Enable = false + } + return + } + c.ExpiryTime = mergeActivationExpiry(master.ExpiryTime, c.ExpiryTime) + c.Enable = master.Enable +} + // nodeClientRenewed reports a node-side auto-renew: an absolute deadline moved -// forward while the node's cumulative counter fell below the stored baseline. +// forward, evidenced by a renewal-count bump or a drop below the stored baseline. func nodeClientRenewed(existing *xray.ClientTraffic, cs xray.ClientTraffic, canon, base nodeTrafficCounter) bool { if (cs.Reset <= 0 && cs.ResetDay <= 0) || cs.ExpiryTime <= 0 || existing.ExpiryTime <= 0 { return false @@ -243,6 +287,11 @@ func nodeClientRenewed(existing *xray.ClientTraffic, cs xray.ClientTraffic, cano if cs.ExpiryTime <= existing.ExpiryTime { return false } + // A client that used no traffic in the period never dips, so the renewal + // counter is the only evidence autoRenewClients leaves behind (#6228). + if cs.ResetCount > existing.ResetCount { + return true + } return canon.Up < base.Up || canon.Down < base.Down } @@ -292,11 +341,13 @@ func (s *InboundService) SnapshotHasUnadoptedInbounds(nodeID int, snap *runtime. return false, nil } -func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnapshot, dirty bool) (bool, error) { +// SetRemoteTraffic merges a node snapshot. justPushed marks the tick whose +// config push just landed, whose snapshot may still predate it (#6228). +func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnapshot, dirty, justPushed bool) (bool, error) { var structuralChange bool err := submitTrafficWrite(func() error { var inner error - structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty) + structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty, justPushed) return inner }) return structuralChange, err @@ -361,7 +412,7 @@ func adoptedWireInbound(c, snapIb *model.Inbound, adoptedSettings string) *model return &a } -func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty bool) (bool, error) { +func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty, justPushed bool) (bool, error) { if snap == nil || nodeID <= 0 { return false, nil } @@ -382,6 +433,9 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // Re-read inside the serialized writer: a client added while this snapshot // was in flight marks the node dirty after the caller sampled the flag. dirty = dirty || nodeRow.ConfigDirty + // Adoption, record sync and sweeps still run on a just-pushed tick; only the + // client lifecycle merge waits for a snapshot that reflects the push. + lifecycleFrozen := dirty || justPushed nodeRow.Id = nodeID unmanagedTag := unmanagedTagPredicate(&nodeRow) selfKey := effectiveNodeKey(&model.Node{Id: nodeID, Guid: nodeRow.Guid}) @@ -521,8 +575,15 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi }() structuralChange := false + lifecycleLifted := false var adoptedInbounds []*model.Inbound + type pendingAdopt struct { + central *model.Inbound + snapIb *model.Inbound + wireSettings string + } + var pendingAdopts []pendingAdopt newInboundIDs := make(map[int]struct{}) @@ -659,9 +720,13 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if deduped, changed := dedupeSettingsClients(adoptedSettings); changed { adoptedSettings = deduped } - updates := map[string]any{} if !dirty { + // Defer lifecycle lift until after client_traffics absorbs this tick's + // deltas so quota stale-disable matches SQL (#6228). + pendingAdopts = append(pendingAdopts, pendingAdopt{ + central: c, snapIb: snapIb, wireSettings: adoptedSettings, + }) updates["enable"] = snapIb.Enable updates["remark"] = snapIb.Remark updates["sub_sort_index"] = normalizeSubSortIndex(snapIb.SubSortIndex) @@ -670,15 +735,11 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi updates["protocol"] = snapIb.Protocol updates["total"] = snapIb.Total updates["expiry_time"] = snapIb.ExpiryTime - updates["settings"] = adoptedSettings updates["stream_settings"] = snapIb.StreamSettings updates["sniffing"] = snapIb.Sniffing updates["traffic_reset"] = snapIb.TrafficReset updates["traffic_reset_day"] = normalizeTrafficResetDay(snapIb.TrafficResetDay) 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 @@ -691,8 +752,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi updates["origin_node_guid"] = og } - if !dirty && (c.Settings != adoptedSettings || - c.Remark != snapIb.Remark || + if !dirty && (c.Remark != snapIb.Remark || c.Listen != snapIb.Listen || c.Port != snapIb.Port || c.Total != snapIb.Total || @@ -802,6 +862,8 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi continue } snapEmails := make(map[string]struct{}, len(snapIb.ClientStats)) + // Parsed once per inbound on the first renewal candidate, not per client. + var snapExpiries map[string]int64 for _, cs := range snapIb.ClientStats { snapEmails[cs.Email] = struct{}{} @@ -864,27 +926,42 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } existing := centralCSByEmail[cs.Email] - if existing != nil && - (existing.Enable != cs.Enable || - existing.Total != cs.Total || - existing.ExpiryTime != mergeActivationExpiry(existing.ExpiryTime, cs.ExpiryTime) || - existing.Reset != cs.Reset) { - structuralChange = true + if existing != nil { + expiryChanged := !lifecycleFrozen && existing.ExpiryTime != mergeActivationExpiry(existing.ExpiryTime, cs.ExpiryTime) + // Only a real latch to disabled is structural; one-way merge never + // re-enables from the node. + enableChanged := !lifecycleFrozen && existing.Enable && !cs.Enable && + !nodeDisableIsStale(existing, cs, now, deltaUp, deltaDown) + metaChanged := !lifecycleFrozen && (existing.Total != cs.Total || existing.Reset != cs.Reset) + if enableChanged || metaChanged || expiryChanged { + structuralChange = true + } } - if seen && existing != nil && nodeClientRenewed(existing, cs, canon, base) { + renewed := !lifecycleFrozen && seen && existing != nil && nodeClientRenewed(existing, cs, canon, base) + if renewed { + // Reject when the node's own settings still carry the old absolute: + // lagging ClientStats after a master shorten mimic a renew (#6228). + if snapExpiries == nil { + snapExpiries = settingsClientAbsoluteExpiries(snapIb.Settings) + } + if se, ok := snapExpiries[cs.Email]; ok && se <= existing.ExpiryTime { + renewed = false + } + } + if renewed { // A renewal starts a fresh quota window: adopt the node's counters // and enable state, drop stale pushes (mirrors autoRenewClients). if err := tx.Exec( fmt.Sprintf( `UPDATE client_traffics SET up = ?, down = ?, enable = ?, total = ?, - expiry_time = ?, reset = ?, reset_day = ?, last_online = %s + expiry_time = ?, reset = ?, reset_day = ?, reset_count = ?, last_online = %s WHERE email = ?`, database.GreatestExpr("last_online", "?"), ), canon.Up, canon.Down, cs.Enable, cs.Total, - cs.ExpiryTime, cs.Reset, cs.ResetDay, + cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetCount, cs.LastOnline, cs.Email, ).Error; err != nil { return false, err @@ -892,33 +969,74 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if err := clearGlobalTraffic(tx, cs.Email); err != nil { return false, err } + existing.Up = canon.Up + existing.Down = canon.Down + existing.Enable = cs.Enable + existing.Total = cs.Total + existing.ExpiryTime = cs.ExpiryTime + existing.Reset = cs.Reset + existing.ResetCount = cs.ResetCount + structuralChange = true + } else if lifecycleFrozen { + // 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() - // expiry_time merge mirrors mergeActivationExpiry: a node that has not - // yet seen the client's first connection keeps reporting the negative - // "start after first connect" duration, which must never reset the - // absolute deadline another node already activated. A positive node - // value is still adopted (e.g. auto-renew moves the deadline forward). - // CAST(? AS BIGINT): in the `<= 0` comparison Postgres would otherwise - // infer int4 from the literal and overflow on real expiry values. + expiryExpr := database.ClientTrafficExpiryMergeExpr() if err := tx.Exec( fmt.Sprintf( `UPDATE client_traffics SET up = %s, down = %s, enable = %s, total = ?, - expiry_time = CASE WHEN expiry_time > 0 AND CAST(? AS BIGINT) <= 0 THEN expiry_time ELSE CAST(? AS BIGINT) END, + expiry_time = %s, reset = ?, reset_day = ?, last_online = %s WHERE email = ?`, database.ClampedAddExpr("up"), database.ClampedAddExpr("down"), enableExpr, + expiryExpr, database.GreatestExpr("last_online", "?"), ), - deltaUp, deltaDown, cs.Enable, cs.Total, - cs.ExpiryTime, cs.ExpiryTime, cs.Reset, cs.ResetDay, + deltaUp, deltaDown, + cs.Enable, cs.ExpiryTime, cs.Total, now, deltaUp, deltaDown, + cs.Total, + cs.ExpiryTime, cs.Reset, cs.ResetDay, 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 + } + 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 + } + } + // A dip plus a lagging longer expiry mimics nodeClientRenewed and would + // undo a master shorten once the freeze lifts (#6228). + if lifecycleFrozen && seen && (canon.Up < base.Up || canon.Down < base.Down) { + continue } if err := s.upsertNodeBaseline(tx, nodeID, cs.Email, canon.Up, canon.Down); err != nil { return false, err @@ -971,6 +1089,27 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } var perInboundOld []oldSet syncFailedInbounds := map[int]struct{}{} + for _, p := range pendingAdopts { + lifted, liftChanged := liftClientLifecycleInSettings(p.wireSettings, centralCSByEmail) + adoptedSettings := p.wireSettings + if liftChanged { + adoptedSettings = lifted + lifecycleLifted = true + } + if p.central.Settings != adoptedSettings { + if err := tx.Model(model.Inbound{}). + Where("id = ?", p.central.Id). + Update("settings", adoptedSettings).Error; err != nil { + return false, err + } + structuralChange = true + } + // The fingerprint stamps the un-lifted wire blob on purpose: a lift must + // leave reconcile a mismatch to re-push against. + if liftChanged || adoptedWireChanged(p.central, p.snapIb, p.wireSettings) { + adoptedInbounds = append(adoptedInbounds, adoptedWireInbound(p.central, p.snapIb, p.wireSettings)) + } + } for _, snapIb := range snap.Inbounds { if snapIb == nil { continue @@ -1001,18 +1140,22 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi logger.Warningf("setRemoteTraffic: parse clients for tag %q failed: %v", snapIb.Tag, gcErr) continue } - csEnableByEmail := make(map[string]bool, len(snapIb.ClientStats)) + csByEmail := make(map[string]xray.ClientTraffic, len(snapIb.ClientStats)) for _, cs := range snapIb.ClientStats { - csEnableByEmail[cs.Email] = cs.Enable + csByEmail[cs.Email] = cs } filtered := clients[:0] for i := range clients { if isClientEmailTombstoned(clients[i].Email) { continue } - if cse, hit := csEnableByEmail[clients[i].Email]; hit && !cse { - clients[i].Enable = false + existing := centralCSByEmail[clients[i].Email] + var csPtr *xray.ClientTraffic + if cs, hit := csByEmail[clients[i].Email]; hit { + csCopy := cs + csPtr = &csCopy } + applyMasterClientLifecycle(&clients[i], existing, csPtr) filtered = append(filtered, clients[i]) } localEmails := make([]string, 0, len(filtered)) @@ -1101,6 +1244,18 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } committed = true + if lifecycleLifted && !dirty { + var already model.Node + if err := database.GetDB().Select("config_dirty").Where("id = ?", nodeID).First(&already).Error; err == nil && already.ConfigDirty { + logger.Debugf("setRemoteTraffic: node %d lifecycle lift; already dirty", nodeID) + } else { + logger.Infof("setRemoteTraffic: node %d lifecycle lift; marking dirty for re-push", nodeID) + if err := (&NodeService{}).MarkNodeDirty(nodeID); err != nil { + logger.Warningf("setRemoteTraffic: mark node %d dirty after lifecycle lift failed: %v", nodeID, err) + } + } + } + if len(adoptedInbounds) > 0 { if mgr := runtime.GetManager(); mgr != nil { if rt, rtErr := mgr.RuntimeFor(&nodeID); rtErr == nil { diff --git a/internal/web/service/inbound_node_sweep_log_test.go b/internal/web/service/inbound_node_sweep_log_test.go index b8c0192c1..d6a1cd536 100644 --- a/internal/web/service/inbound_node_sweep_log_test.go +++ b/internal/web/service/inbound_node_sweep_log_test.go @@ -28,7 +28,7 @@ func TestNodeSnapshotSweepLogsRemovedInbound(t *testing.T) { Tag: survivor.Tag, Port: survivor.Port, Protocol: model.VLESS, Enable: true, Settings: survivor.Settings, }}} - if _, err := (&InboundService{}).setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := (&InboundService{}).setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } diff --git a/internal/web/service/node_client_expiry_sync_test.go b/internal/web/service/node_client_expiry_sync_test.go index 17eba9626..bdd498c06 100644 --- a/internal/web/service/node_client_expiry_sync_test.go +++ b/internal/web/service/node_client_expiry_sync_test.go @@ -1,9 +1,13 @@ package service import ( + "fmt" "testing" + "time" + "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/xray" ) @@ -24,8 +28,9 @@ func TestMergeActivationExpiry(t *testing.T) { {"activation adopted over stored duration", dur, early, early}, {"node still un-activated does not reset deadline", early, dur, early}, {"node un-activated zero does not reset deadline", early, 0, early}, - {"node renewal extends the deadline forward", early, late, late}, - {"node positive adopted even if earlier", late, early, early}, + {"master absolute ignores later node absolute", early, late, early}, + {"node value equal to master is a no-op", early, early, early}, + {"stale earlier absolute does not clobber later", late, early, late}, {"both un-activated keep node value", dur, dur, dur}, } for _, c := range cases { @@ -37,6 +42,76 @@ func TestMergeActivationExpiry(t *testing.T) { } } +func TestNodeDisableIsStale(t *testing.T) { + now := time.Now().UnixMilli() + const quota = int64(100) + cases := []struct { + name string + master *xray.ClientTraffic + node xray.ClientTraffic + deltaUp int64 + deltaDown int64 + wantStale bool + }{ + {name: "nil master", node: xray.ClientTraffic{ExpiryTime: earlyAbs}}, + { + name: "matching limits are a genuine verdict", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota}, + node: xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota}, + wantStale: false, + }, + { + name: "node still holds the pre-extension deadline", + master: &xray.ClientTraffic{ExpiryTime: lateAbs}, + node: xray.ClientTraffic{ExpiryTime: earlyAbs}, + wantStale: true, + }, + { + name: "node still holds the pre-top-up quota", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: 2 * quota, Up: 60, Down: 50}, + node: xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota}, + wantStale: true, + }, + { + name: "master itself expired", + master: &xray.ClientTraffic{ExpiryTime: earlyAbs}, + node: xray.ClientTraffic{ExpiryTime: earlyAbs - 1000}, + }, + { + name: "master itself over quota", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 60, Down: 50}, + node: xray.ClientTraffic{ExpiryTime: earlyAbs, Total: quota}, + }, + { + name: "this tick's deltas cross the master quota", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 40, Down: 50}, + node: xray.ClientTraffic{ExpiryTime: earlyAbs, Total: quota}, + deltaUp: 10, + deltaDown: 10, + }, + { + // Same row without the deltas: the crossing above is the deltas' doing. + name: "under the master quota before this tick's deltas", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 40, Down: 50}, + node: xray.ClientTraffic{ExpiryTime: earlyAbs, Total: quota}, + wantStale: true, + }, + { + name: "un-activated node duration against an absolute master", + master: &xray.ClientTraffic{ExpiryTime: lateAbs}, + node: xray.ClientTraffic{ExpiryTime: -2592000000}, + wantStale: true, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := nodeDisableIsStale(c.master, c.node, now, c.deltaUp, c.deltaDown); got != c.wantStale { + t.Fatalf("nodeDisableIsStale(...) = %v, want %v", got, c.wantStale) + } + }) + } +} + // TestNodeFirstConnectExpiry_NotClobbered reproduces the multi-node bug: a // client is attached to inbounds on two nodes with a "start after first connect" // expiry. The client connects only on node 1, which activates an absolute @@ -117,10 +192,8 @@ func TestNodeFirstConnectExpiry_NotClobbered_WithSettings(t *testing.T) { } } -// TestNodeRenewExtendsExpiry guards against over-correcting: a node that renews -// a client (traffic reset / auto-renew) legitimately moves the deadline FORWARD -// to a later absolute timestamp, and that must still propagate to the master. -// The guard only rejects un-activated (<= 0) values, never a positive one. +// TestNodeRenewExtendsExpiry: node auto-renew (reset + later expiry + counter +// drop) must still move master expiry forward via nodeClientRenewed. func TestNodeRenewExtendsExpiry(t *testing.T) { db := initTrafficTestDB(t) createNodeInbound(t, db, 1, "n1-in", 41001) @@ -130,17 +203,778 @@ func TestNodeRenewExtendsExpiry(t *testing.T) { const first = int64(1893456000000) const renewed = first + int64(2592000000) // +30 days after auto-renew - syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 10, Down: 10, ExpiryTime: first, Enable: true}) + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 0, Down: 0, ExpiryTime: first, Reset: 30, Enable: true, + }) + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 100, Down: 100, ExpiryTime: first, Reset: 30, Enable: true, + }) if got := readTraffic(t, db, email).ExpiryTime; got != first { t.Fatalf("after activation: expiry = %d, want %d", got, first) } - syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 20, Down: 20, ExpiryTime: renewed, Enable: true}) + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 5, Down: 5, ExpiryTime: renewed, Reset: 30, Enable: true, + }) if got := readTraffic(t, db, email).ExpiryTime; got != renewed { t.Fatalf("node renewal did not propagate: expiry = %d, want %d", got, renewed) } } +// TestNodeRenew_WithMatchingSettings: renew still applies when settings JSON +// also carries the later absolute (guard must not block real renewals). +func TestNodeRenew_WithMatchingSettings(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "renew-settings" + firstSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, renewFirstExpiry) + renewSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, renewSecondExpiry) + + syncNodeWithSettings(t, svc, 1, "n1-in", firstSettings, xray.ClientTraffic{ + Email: email, Up: 0, Down: 0, ExpiryTime: renewFirstExpiry, Reset: renewPeriodDays, Enable: true, + }) + syncNodeWithSettings(t, svc, 1, "n1-in", firstSettings, xray.ClientTraffic{ + Email: email, Up: 100, Down: 100, ExpiryTime: renewFirstExpiry, Reset: renewPeriodDays, Enable: true, + }) + syncNodeWithSettings(t, svc, 1, "n1-in", renewSettings, xray.ClientTraffic{ + Email: email, Up: 5, Down: 5, ExpiryTime: renewSecondExpiry, Reset: renewPeriodDays, Enable: true, + }) + if got := readTraffic(t, db, email).ExpiryTime; got != renewSecondExpiry { + t.Fatalf("renewal with matching settings: got %d want %d", got, renewSecondExpiry) + } +} + +// A node still holding the pre-extension deadline must not undo the extension on +// traffics, client records or the adopted settings JSON (#6228). +func TestNodeStaleExpiryAfterExtend_NotClobbered(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "extended") + svc := &InboundService{} + + const email = "extended" + expired, extended := earlyAbs, lateAbs + + staleSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":false,"expiryTime":%d}]}`, email, expired) + + syncNodeWithSettings(t, svc, 1, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + if got := readTraffic(t, db, email); got.ExpiryTime != expired || got.Enable { + t.Fatalf("after expiry: expiry=%d enable=%v, want expiry=%d enable=false", + got.ExpiryTime, got.Enable, expired) + } + + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": extended, "enable": true}).Error; err != nil { + t.Fatalf("master extend traffic: %v", err) + } + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": extended, "enable": true}).Error; err != nil { + t.Fatalf("master extend record: %v", err) + } + + syncNodeWithSettings(t, svc, 1, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + + got := readTraffic(t, db, email) + if got.ExpiryTime != extended { + t.Fatalf("stale node expiry clobbered traffics: expiry=%d, want %d", got.ExpiryTime, extended) + } + if !got.Enable { + t.Fatal("stale node disable latched traffics.enable off after extension") + } + + var rec model.ClientRecord + if err := db.Where("email = ?", email).First(&rec).Error; err != nil { + t.Fatalf("read client record: %v", err) + } + if rec.ExpiryTime != extended { + t.Fatalf("stale SyncInbound clobbered record expiry: %d, want %d", rec.ExpiryTime, extended) + } + if !rec.Enable { + t.Fatal("stale SyncInbound latched clients.enable off after extension") + } + + var ib model.Inbound + if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil { + t.Fatalf("read inbound: %v", err) + } + clients, err := svc.GetClients(&ib) + if err != nil { + t.Fatalf("GetClients: %v", err) + } + var found bool + for _, c := range clients { + if c.Email != email { + continue + } + found = true + if c.ExpiryTime != extended { + t.Fatalf("adopted settings kept stale expiry: %d, want %d", c.ExpiryTime, extended) + } + if !c.Enable { + t.Fatal("adopted settings kept enable=false after extension") + } + } + if !found { + t.Fatal("client missing from adopted inbound settings after stale sync") + } +} + +// TestNodeStaleLift_MarksNodeDirty: a lifecycle lift must mark the node dirty +// and store lifted settings centrally so reconcile can re-push (#6228). +func TestNodeStaleLift_MarksNodeDirty(t *testing.T) { + db := initTrafficTestDB(t) + node := &model.Node{Name: "lift-n", Address: "127.0.0.1", Port: 2097, ApiToken: "tok", Enable: true, Status: "online"} + if err := db.Create(node).Error; err != nil { + t.Fatalf("create node: %v", err) + } + createNodeInboundWithClient(t, db, node.Id, "n1-in", 41001, "extended") + svc := &InboundService{} + + const email = "extended" + expired, extended := earlyAbs, lateAbs + staleSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":false,"expiryTime":%d}]}`, email, expired) + + syncNodeWithSettings(t, svc, node.Id, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": extended, "enable": true}).Error; err != nil { + t.Fatalf("master extend: %v", err) + } + if err := db.Model(model.Node{}).Where("id = ?", node.Id). + Updates(map[string]any{"config_dirty": false, "config_dirty_at": int64(0)}).Error; err != nil { + t.Fatalf("clear dirty: %v", err) + } + + syncNodeWithSettings(t, svc, node.Id, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + + var n model.Node + if err := db.Select("config_dirty").Where("id = ?", node.Id).First(&n).Error; err != nil { + t.Fatalf("read node: %v", err) + } + if !n.ConfigDirty { + t.Fatal("lifecycle lift must mark the node dirty so reconcile re-pushes") + } + + var ib model.Inbound + if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil { + t.Fatalf("read inbound: %v", err) + } + clients, err := svc.GetClients(&ib) + if err != nil { + t.Fatalf("GetClients: %v", err) + } + var found bool + for _, c := range clients { + if c.Email != email { + continue + } + found = true + if c.ExpiryTime != extended || !c.Enable { + t.Fatalf("lifted settings not stored: expiry=%d enable=%v", c.ExpiryTime, c.Enable) + } + } + if !found { + t.Fatal("client missing from lifted inbound settings") + } + if ib.Settings == staleSettings { + t.Fatal("central settings must differ from pre-lift wire blob (FP basis)") + } +} + +// A node disable decided on the master's own limits is genuine and must still +// one-way-merge enable=false onto the master (#4917). +func TestNodeQuotaDisable_SameExpiryStillLatches(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "quota" + const expiry = int64(1893456000000) + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 10, Down: 10, Total: 100, ExpiryTime: expiry, Enable: true, + }) + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 60, Down: 50, Total: 100, ExpiryTime: expiry, Enable: false, + }) + if got := readTraffic(t, db, email); got.Enable { + t.Fatal("same-expiry node disable must still latch master enable off (#4917)") + } +} + +// Once the master row is itself depleted, a node disable latches even though the +// node's limits lag — otherwise genuine quota cuts would be skipped. +func TestNodeQuotaDisable_OlderExpiryStillLatchesWhenOverQuota(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "quota-lag" + early, late := earlyAbs, lateAbs + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 10, Down: 10, Total: 100, ExpiryTime: late, Enable: true, + }) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{ + "expiry_time": late, "enable": true, "up": int64(60), "down": int64(50), "total": int64(100), + }).Error; err != nil { + t.Fatalf("seed over-quota master: %v", err) + } + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 60, Down: 50, Total: 100, ExpiryTime: early, Enable: false, + }) + if got := readTraffic(t, db, email); got.Enable { + t.Fatal("over-quota master must still adopt node disable despite older node expiry") + } + if got := readTraffic(t, db, email); got.ExpiryTime != late { + t.Fatalf("expiry should stay at master extension: got %d want %d", got.ExpiryTime, late) + } +} + +// TestNodeMasterShorten_NotClobberedWhileDirty: master shortened expiry while +// config_dirty; a lagging longer node snapshot must not raise client_traffics. +func TestNodeMasterShorten_NotClobberedWhileDirty(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "shortened") + svc := &InboundService{} + + const email = "shortened" + longExp, shortExp := lateAbs, earlyAbs + longSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, longExp) + + syncNodeWithSettings(t, svc, 1, "n1-in", longSettings, + xray.ClientTraffic{Email: email, ExpiryTime: longExp, Enable: true}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": shortExp, "enable": true}).Error; err != nil { + t.Fatalf("master shorten: %v", err) + } + if err := db.Model(model.Node{}).Where("id = ?", 1). + Updates(map[string]any{"config_dirty": true, "config_dirty_at": int64(1)}).Error; err != nil { + t.Fatalf("mark dirty: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", Settings: longSettings, + ClientStats: []xray.ClientTraffic{{Email: email, ExpiryTime: longExp, Enable: true, Up: 5, Down: 5}}, + }}, + } + before := readTraffic(t, db, email) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Update("total", int64(999)).Error; err != nil { + t.Fatalf("master total: %v", err) + } + if _, err := svc.setRemoteTrafficLocked(1, snap, true, false); err != nil { + t.Fatalf("dirty sync: %v", err) + } + got := readTraffic(t, db, email) + if got.ExpiryTime != shortExp { + t.Fatalf("dirty sync raised expiry: got %d want %d", got.ExpiryTime, shortExp) + } + if got.Total != 999 { + t.Fatalf("dirty sync adopted node total: got %d want 999", got.Total) + } + if got.Up < before.Up+5 || got.Down < before.Down+5 { + t.Fatalf("dirty sync must still accumulate traffic: before=(%d,%d) after=(%d,%d)", + before.Up, before.Down, got.Up, got.Down) + } +} + +// The master's shortened absolute survives a snapshot whose ClientStats still +// report the longer deadline (#6228). +func TestNodeMasterShorten_LaggingClientStatsIgnored(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "short-settings") + svc := &InboundService{} + + const email = "short-settings" + longExp, shortExp := lateAbs, earlyAbs + longSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, longExp) + shortSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, shortExp) + + syncNodeWithSettings(t, svc, 1, "n1-in", longSettings, + xray.ClientTraffic{Email: email, ExpiryTime: longExp, Enable: true}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": shortExp, "enable": true}).Error; err != nil { + t.Fatalf("master shorten: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", Settings: shortSettings, + ClientStats: []xray.ClientTraffic{{Email: email, ExpiryTime: longExp, Enable: true}}, + }}, + } + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("clean sync: %v", err) + } + if got := readTraffic(t, db, email); got.ExpiryTime != shortExp { + t.Fatalf("lagging ClientStats raised expiry: got %d want %d", got.ExpiryTime, shortExp) + } +} + +// TestNodeMasterShorten_CleanSiblingCannotRaise: a clean sibling still holding +// the longer deadline in settings+ClientStats must not undo a master shorten. +func TestNodeMasterShorten_CleanSiblingCannotRaise(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "sib-short") + createNodeInboundWithClient(t, db, 2, "n2-in", 41002, "sib-short") + svc := &InboundService{} + + const email = "sib-short" + longExp, shortExp := lateAbs, earlyAbs + longSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, longExp) + + syncNodeWithSettings(t, svc, 1, "n1-in", longSettings, + xray.ClientTraffic{Email: email, ExpiryTime: longExp, Enable: true}) + syncNodeWithSettings(t, svc, 2, "n2-in", longSettings, + xray.ClientTraffic{Email: email, ExpiryTime: longExp, Enable: true}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": shortExp, "enable": true}).Error; err != nil { + t.Fatalf("master shorten: %v", err) + } + + // Sibling 2 is clean and still reports the old longer deadline. + syncNodeWithSettings(t, svc, 2, "n2-in", longSettings, + xray.ClientTraffic{Email: email, ExpiryTime: longExp, Enable: true}) + if got := readTraffic(t, db, email); got.ExpiryTime != shortExp { + t.Fatalf("clean sibling raised shortened expiry: got %d want %d", got.ExpiryTime, shortExp) + } +} + +// TestNodeMasterShorten_LaggingStatsNotTreatedAsRenew: after shorten, ClientStats +// may still show the longer deadline with Reset+dip — must not call renewal. +func TestNodeMasterShorten_LaggingStatsNotTreatedAsRenew(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "short-renew-trap" + longExp, shortExp := lateAbs, earlyAbs + shortSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, shortExp) + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 0, Down: 0, ExpiryTime: longExp, Reset: 30, Enable: true, + }) + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 100, Down: 100, ExpiryTime: longExp, Reset: 30, Enable: true, + }) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": shortExp, "enable": true}).Error; err != nil { + t.Fatalf("master shorten: %v", err) + } + + syncNodeWithSettings(t, svc, 1, "n1-in", shortSettings, + xray.ClientTraffic{Email: email, Up: 5, Down: 5, ExpiryTime: longExp, Reset: 30, Enable: true}) + if got := readTraffic(t, db, email); got.ExpiryTime != shortExp { + t.Fatalf("lagging stats treated as renew: got %d want %d", got.ExpiryTime, shortExp) + } +} + +// TestNodeStaleExpiryAfterExtend_WhileDirty keeps master extend while dirty. +func TestNodeStaleExpiryAfterExtend_WhileDirty(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "extended-dirty") + svc := &InboundService{} + + const email = "extended-dirty" + expired, extended := earlyAbs, lateAbs + staleSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":false,"expiryTime":%d}]}`, email, expired) + + syncNodeWithSettings(t, svc, 1, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": extended, "enable": true}).Error; err != nil { + t.Fatalf("master extend: %v", err) + } + if err := db.Model(model.Node{}).Where("id = ?", 1). + Updates(map[string]any{"config_dirty": true, "config_dirty_at": int64(1)}).Error; err != nil { + t.Fatalf("mark dirty: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", Settings: staleSettings, + ClientStats: []xray.ClientTraffic{{Email: email, ExpiryTime: expired, Enable: false}}, + }}, + } + if _, err := svc.setRemoteTrafficLocked(1, snap, true, false); err != nil { + t.Fatalf("dirty sync: %v", err) + } + got := readTraffic(t, db, email) + if got.ExpiryTime != extended || !got.Enable { + t.Fatalf("dirty sync clobbered extend: expiry=%d enable=%v", got.ExpiryTime, got.Enable) + } +} + +// TestNodeRenewal_SkippedWhileDirty: renewal-shaped stats must not adopt +// expiry/enable over a pending master push. +func TestNodeRenewal_SkippedWhileDirty(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "renew-dirty" + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 100, Down: 100, ExpiryTime: earlyAbs, Enable: true, Reset: 30, + }) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": earlyAbs, "enable": true, "up": int64(100), "down": int64(100)}).Error; err != nil { + t.Fatalf("seed master: %v", err) + } + if err := db.Model(model.Node{}).Where("id = ?", 1). + Updates(map[string]any{"config_dirty": true, "config_dirty_at": int64(1)}).Error; err != nil { + t.Fatalf("mark dirty: %v", err) + } + // Renewal shape: later expiry + counters below baseline. + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", + ClientStats: []xray.ClientTraffic{{ + Email: email, Up: 10, Down: 10, ExpiryTime: lateAbs, Enable: true, Reset: 30, + }}, + }}, + } + if _, err := svc.setRemoteTrafficLocked(1, snap, true, false); err != nil { + t.Fatalf("dirty sync: %v", err) + } + got := readTraffic(t, db, email) + if got.ExpiryTime != earlyAbs { + t.Fatalf("renewal while dirty raised expiry: got %d want %d", got.ExpiryTime, earlyAbs) + } + + // Same renewal-shaped stats on a clean tick must still advance expiry — + // dirty must not have burned the dipped baseline. + if err := db.Model(model.Node{}).Where("id = ?", 1). + Updates(map[string]any{"config_dirty": false, "config_dirty_at": int64(0)}).Error; err != nil { + t.Fatalf("clear dirty: %v", err) + } + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("clean sync: %v", err) + } + if got := readTraffic(t, db, email); got.ExpiryTime != lateAbs { + t.Fatalf("renewal after dirty clear did not apply: got %d want %d", got.ExpiryTime, lateAbs) + } +} + +// TestNodeExtend_FreshSettingsLaggingDisable: after extend, settings may already +// show the new absolute while ClientStats still report enable=false + old expiry. +func TestNodeExtend_FreshSettingsLaggingDisable(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "ext-lag") + svc := &InboundService{} + + const email = "ext-lag" + expired, extended := earlyAbs, lateAbs + staleSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":false,"expiryTime":%d}]}`, email, expired) + freshSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, extended) + + syncNodeWithSettings(t, svc, 1, "n1-in", staleSettings, + xray.ClientTraffic{Email: email, ExpiryTime: expired, Enable: false}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{"expiry_time": extended, "enable": true}).Error; err != nil { + t.Fatalf("master extend: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", Settings: freshSettings, + ClientStats: []xray.ClientTraffic{{Email: email, ExpiryTime: expired, Enable: false}}, + }}, + } + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("clean sync: %v", err) + } + got := readTraffic(t, db, email) + if got.ExpiryTime != extended { + t.Fatalf("expiry clobbered: got %d want %d", got.ExpiryTime, extended) + } + if !got.Enable { + t.Fatal("lagging ClientStats enable=false latched master off after extend") + } +} + +// A client that used no traffic never dips below its baseline, so the renewal +// counter is the only evidence the node auto-renewed (#6228). +func TestNodeRenew_ZeroTrafficUsesResetCount(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "idle-renew" + firstSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, earlyAbs) + renewSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, lateAbs) + + syncNodeWithSettings(t, svc, 1, "n1-in", firstSettings, xray.ClientTraffic{ + Email: email, ExpiryTime: earlyAbs, Reset: 30, Enable: true, + }) + syncNodeWithSettings(t, svc, 1, "n1-in", renewSettings, xray.ClientTraffic{ + Email: email, ExpiryTime: lateAbs, Reset: 30, ResetCount: 1, Enable: true, + }) + + got := readTraffic(t, db, email) + if got.ExpiryTime != lateAbs { + t.Fatalf("zero-traffic renewal dropped: expiry=%d want %d", got.ExpiryTime, lateAbs) + } + if got.ResetCount != 1 { + t.Fatalf("renewal count not persisted, so the next renewal cannot be seen: got %d want 1", got.ResetCount) + } +} + +// A quota top-up leaves the expiry alone, so the node's lagging disable must be +// recognised by the stale quota it was decided against (#6228). +func TestNodeQuotaTopUp_LaggingDisableIgnored(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "topped-up" + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 10, Down: 10, Total: 100, ExpiryTime: lateAbs, Enable: true, + }) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Updates(map[string]any{ + "total": int64(500), "enable": true, "up": int64(60), "down": int64(50), + }).Error; err != nil { + t.Fatalf("master top-up: %v", err) + } + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: email, Up: 60, Down: 50, Total: 100, ExpiryTime: lateAbs, Enable: false, + }) + if got := readTraffic(t, db, email); !got.Enable { + t.Fatal("node disable decided on the pre-top-up quota latched over the raised one") + } +} + +// The settings lift is authoritative in both directions: a blob predating a +// master disable must not carry enable=true back into central settings (#4917). +func TestNodeStaleEnable_LiftedOffInSettings(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "cut-off") + svc := &InboundService{} + + const email = "cut-off" + liveSettings := fmt.Sprintf( + `{"clients":[{"email":%q,"enable":true,"expiryTime":%d}]}`, email, lateAbs) + + syncNodeWithSettings(t, svc, 1, "n1-in", liveSettings, + xray.ClientTraffic{Email: email, ExpiryTime: lateAbs, Enable: true}) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email). + Update("enable", false).Error; err != nil { + t.Fatalf("master disable: %v", err) + } + + syncNodeWithSettings(t, svc, 1, "n1-in", liveSettings, + xray.ClientTraffic{Email: email, ExpiryTime: lateAbs, Enable: true}) + + var ib model.Inbound + if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil { + t.Fatalf("read inbound: %v", err) + } + clients, err := svc.GetClients(&ib) + if err != nil { + t.Fatalf("GetClients: %v", err) + } + var found bool + for _, c := range clients { + if c.Email != email { + continue + } + found = true + if c.Enable { + t.Fatal("adopted settings re-enabled a client the master disabled") + } + } + if !found { + t.Fatal("client missing from adopted inbound settings") + } +} + +// The tick whose push just landed freezes only the lifecycle merge: adoption, +// new client rows and traffic accumulation must keep working (#6228). +func TestNodeJustPushed_FreezesLifecycleOnly(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const kept = "kept" + const fresh = "fresh" + extended := lateAbs + 86400000 + + syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{ + Email: kept, Up: 10, Down: 10, Total: 100, ExpiryTime: lateAbs, Enable: true, + }) + if err := db.Model(xray.ClientTraffic{}).Where("email = ?", kept). + Updates(map[string]any{"expiry_time": extended, "total": int64(500)}).Error; err != nil { + t.Fatalf("master edit: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "n1-in", + ClientStats: []xray.ClientTraffic{ + {Email: kept, Up: 20, Down: 20, Total: 100, ExpiryTime: lateAbs, Enable: false}, + {Email: fresh, Up: 5, Down: 5, ExpiryTime: lateAbs, Enable: true}, + }, + }}, + } + if _, err := svc.setRemoteTrafficLocked(1, snap, false, true); err != nil { + t.Fatalf("just-pushed sync: %v", err) + } + + got := readTraffic(t, db, kept) + if got.ExpiryTime != extended || got.Total != 500 || !got.Enable { + t.Fatalf("just-pushed tick adopted lagging lifecycle: expiry=%d total=%d enable=%v", + got.ExpiryTime, got.Total, got.Enable) + } + if got.Up != 10 || got.Down != 10 { + t.Fatalf("just-pushed tick dropped this tick's traffic: up=%d down=%d, want 10/10", got.Up, got.Down) + } + if row := readTraffic(t, db, fresh); row.Email != fresh { + t.Fatalf("just-pushed tick skipped adoption of a new client: %+v", row) + } +} + +// TestClientTrafficMergeSQLMatchesHelpers pins the dialect SQL expressions +// against the Go helpers so the in-memory replay after UPDATE cannot drift. +func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) { + db := initTrafficTestDB(t) + + const email = "sql-merge" + now := time.Now().UnixMilli() + cases := []struct { + name string + masterExpiry int64 + masterEnable bool + masterUp, masterDown, tot int64 + deltaUp, deltaDown int64 + nodeExpiry, nodeTotal int64 + nodeEnable bool + wantExpiry int64 + wantEnable bool + }{ + { + name: "stale expiry+disable after extend", + masterExpiry: lateAbs, masterEnable: true, + nodeExpiry: earlyAbs, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: true, + }, + { + name: "same expiry quota disable", + masterExpiry: lateAbs, masterEnable: true, + masterUp: 60, masterDown: 50, tot: 100, + nodeExpiry: lateAbs, nodeTotal: 100, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: false, + }, + { + name: "older expiry but master over quota", + masterExpiry: lateAbs, masterEnable: true, + masterUp: 60, masterDown: 50, tot: 100, + nodeExpiry: earlyAbs, nodeTotal: 100, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: false, + }, + { + name: "master absolute ignores later node", + masterExpiry: earlyAbs, masterEnable: true, + nodeExpiry: lateAbs, nodeEnable: true, + wantExpiry: earlyAbs, wantEnable: true, + }, + { + name: "negative node keeps absolute", + masterExpiry: lateAbs, masterEnable: true, + nodeExpiry: -2592000000, nodeEnable: true, + wantExpiry: lateAbs, wantEnable: true, + }, + { + name: "expired master latches node disable", + masterExpiry: earlyAbs, masterEnable: true, + nodeExpiry: earlyAbs - 1000, nodeEnable: false, + wantExpiry: earlyAbs, wantEnable: false, + }, + { + name: "older expiry crosses quota via deltas", + masterExpiry: lateAbs, masterEnable: true, + masterUp: 40, masterDown: 50, tot: 100, + deltaUp: 10, deltaDown: 10, + nodeExpiry: earlyAbs, nodeTotal: 100, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: false, + }, + } + + enableExpr := database.ClientTrafficEnableMergeExpr() + expiryExpr := database.ClientTrafficExpiryMergeExpr() + for i, c := range cases { + t.Run(c.name, func(t *testing.T) { + rowEmail := fmt.Sprintf("%s-%d", email, i) + if err := db.Create(&xray.ClientTraffic{ + InboundId: 1, Email: rowEmail, Enable: c.masterEnable, + ExpiryTime: c.masterExpiry, Up: c.masterUp, Down: c.masterDown, Total: c.tot, + }).Error; err != nil { + t.Fatalf("seed: %v", err) + } + master := &xray.ClientTraffic{ + ExpiryTime: c.masterExpiry, Enable: c.masterEnable, + Up: c.masterUp, Down: c.masterDown, Total: c.tot, + } + wantExpiry := mergeActivationExpiry(c.masterExpiry, c.nodeExpiry) + wantEnable := c.masterEnable + node := xray.ClientTraffic{ExpiryTime: c.nodeExpiry, Total: c.nodeTotal} + if !c.nodeEnable && !nodeDisableIsStale(master, node, now, c.deltaUp, c.deltaDown) { + wantEnable = false + } + if wantExpiry != c.wantExpiry || wantEnable != c.wantEnable { + t.Fatalf("helper expectation drift: helpers=(%d,%v) fixture=(%d,%v)", + wantExpiry, wantEnable, c.wantExpiry, c.wantEnable) + } + + if err := db.Exec( + fmt.Sprintf( + `UPDATE client_traffics SET enable = %s, expiry_time = %s WHERE email = ?`, + enableExpr, expiryExpr, + ), + c.nodeEnable, c.nodeExpiry, c.nodeTotal, now, c.deltaUp, c.deltaDown, + c.nodeExpiry, + rowEmail, + ).Error; err != nil { + t.Fatalf("SQL merge: %v", err) + } + got := readTraffic(t, db, rowEmail) + if got.ExpiryTime != c.wantExpiry || got.Enable != c.wantEnable { + t.Fatalf("SQL merge got expiry=%d enable=%v, want expiry=%d enable=%v", + got.ExpiryTime, got.Enable, c.wantExpiry, c.wantEnable) + } + }) + } +} + +// Relative to the run: the merge rules now compare the master deadline against +// wall-clock now, so fixed timestamps would rot into the wrong side of it. +var ( + earlyAbs = time.Now().UnixMilli() - 30*86400000 + lateAbs = time.Now().UnixMilli() + 30*86400000 +) + // TestNodeActivationLiftsClientRecordExpiry reproduces #5714: the node activates // the deadline (positive ClientStats) while its settings JSON still carries the // negative duration, so SyncInbound keeps writing the stale value into the diff --git a/internal/web/service/node_client_traffic_sum_test.go b/internal/web/service/node_client_traffic_sum_test.go index dbb09e35c..f13c030a8 100644 --- a/internal/web/service/node_client_traffic_sum_test.go +++ b/internal/web/service/node_client_traffic_sum_test.go @@ -51,7 +51,7 @@ func syncNode(t *testing.T, svc *InboundService, nodeID int, tag string, stats . snap := &runtime.TrafficSnapshot{ Inbounds: []*model.Inbound{{Tag: tag, ClientStats: stats}}, } - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked node %d: %v", nodeID, err) } } @@ -65,7 +65,7 @@ func syncNodeWithSettings(t *testing.T, svc *InboundService, nodeID int, tag, se snap := &runtime.TrafficSnapshot{ Inbounds: []*model.Inbound{{Tag: tag, Settings: settings, ClientStats: stats}}, } - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked node %d: %v", nodeID, err) } } @@ -344,7 +344,7 @@ func TestInboundRemoval_KeepsSharedEmailRow(t *testing.T) { // vanishes from the snapshot. The shared accumulator must survive — losing // it would let the next node sync re-seed the row with that node's counter // alone, showing only the last panel's number instead of the sum. - if _, err := svc.setRemoteTrafficLocked(1, &runtime.TrafficSnapshot{}, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, &runtime.TrafficSnapshot{}, false, false); err != nil { t.Fatalf("sync node 1 with empty snapshot: %v", err) } assertUpDown(t, readTraffic(t, db, email), 110, 110, "after node 1 inbound removal") @@ -406,7 +406,7 @@ func TestStatsUnderSiblingInbound_KeepsNodeBaseline(t *testing.T) { {Tag: "n1-a", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: up, Down: down, Enable: true}}}, {Tag: "n1-b", Settings: `{"clients": []}`}, }} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("sync: %v", err) } } @@ -451,7 +451,7 @@ func TestMultiAttach_SameNode_DivergentSiblings(t *testing.T) { {Tag: "n1-b", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: b, Down: b, Enable: true}}}, {Tag: "n1-c", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: c, Down: c, Enable: true}}}, }} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("sync: %v", err) } } diff --git a/internal/web/service/node_dirty_test.go b/internal/web/service/node_dirty_test.go index de4872e1d..111cba574 100644 --- a/internal/web/service/node_dirty_test.go +++ b/internal/web/service/node_dirty_test.go @@ -52,7 +52,7 @@ func TestSetRemoteTraffic_DirtyPreservesConfig(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(id, snap, true); err != nil { + if _, err := svc.setRemoteTrafficLocked(id, snap, true, false); err != nil { t.Fatalf("setRemoteTrafficLocked dirty: %v", err) } @@ -93,7 +93,7 @@ func TestSetRemoteTraffic_MissingDisabledInboundIsNotSwept(t *testing.T) { Tag: reported.Tag, Enable: true, Port: reported.Port, Protocol: reported.Protocol, Settings: reported.Settings, }}} - if _, err := (&InboundService{}).setRemoteTrafficLocked(node.Id, snap, false); err != nil { + if _, err := (&InboundService{}).setRemoteTrafficLocked(node.Id, snap, false, false); err != nil { t.Fatal(err) } var count int64 diff --git a/internal/web/service/node_host_adopt_test.go b/internal/web/service/node_host_adopt_test.go index 1edb98b31..7ab57fc17 100644 --- a/internal/web/service/node_host_adopt_test.go +++ b/internal/web/service/node_host_adopt_test.go @@ -46,7 +46,7 @@ func TestSetRemoteTraffic_AdoptsNodeHostRows(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } diff --git a/internal/web/service/node_origin_guid_test.go b/internal/web/service/node_origin_guid_test.go index 2748cb1b5..dc3a8d79a 100644 --- a/internal/web/service/node_origin_guid_test.go +++ b/internal/web/service/node_origin_guid_test.go @@ -63,7 +63,7 @@ func TestSetRemoteTraffic_AttributesOriginNodeGuid(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -124,7 +124,7 @@ func TestSetRemoteTraffic_RemapsClonedNodeOwnGuidOrigin(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -191,7 +191,7 @@ func TestSetRemoteTraffic_RemapsActiveInboundTreeAndCentralTags(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -254,7 +254,7 @@ func TestSetRemoteTraffic_NormalizesForwardedActiveInboundSubtreeTags(t *testing } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -301,7 +301,7 @@ func TestSetRemoteTraffic_DropsForeignActiveInboundGuid(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -336,7 +336,7 @@ func TestSetRemoteTraffic_EmptySnapshotKeepsCentralInbounds(t *testing.T) { // Empty snapshot — the node reported no inbounds this cycle. svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, &runtime.TrafficSnapshot{}, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, &runtime.TrafficSnapshot{}, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -391,7 +391,7 @@ func TestSetRemoteTraffic_PreservesLocalShareAddressStrategy(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -436,7 +436,7 @@ func TestSetRemoteTraffic_DefaultsShareAddressFieldsForNewCentralInbound(t *test } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } diff --git a/internal/web/service/node_sweep_guard_test.go b/internal/web/service/node_sweep_guard_test.go index 15c66fac1..76d1cace4 100644 --- a/internal/web/service/node_sweep_guard_test.go +++ b/internal/web/service/node_sweep_guard_test.go @@ -74,7 +74,7 @@ func TestSetRemoteTrafficRereadsConfigDirty(t *testing.T) { t.Fatalf("mark node dirty: %v", err) } - if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithoutClients(t, "n1-in"), false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -101,7 +101,7 @@ func TestDeselectedTagIsNotSwept(t *testing.T) { keepSettings := `{"clients":[{"email":"kept@x","enable":true}]}` dropSettings := `{"clients":[{"email":"dropped@x","enable":true}]}` snap := snapshotWithTwoInbounds(t, "keep", keepSettings, "kept@x", "drop", dropSettings, "dropped@x") - if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { t.Fatalf("seed sync: %v", err) } if rec, traf := countClientRows(t, db, "dropped@x"); rec != 1 || traf != 1 { @@ -109,7 +109,7 @@ func TestDeselectedTagIsNotSwept(t *testing.T) { } keepOnly := snapshotWithClients(t, "keep", keepSettings, xray.ClientTraffic{Email: "kept@x", Enable: true}) - if _, err := svc.setRemoteTrafficLocked(1, keepOnly, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(1, keepOnly, false, false); err != nil { t.Fatalf("post-deselect sync: %v", err) } diff --git a/internal/web/service/node_tag_sync_test.go b/internal/web/service/node_tag_sync_test.go index f97a12e9a..8dde58aa2 100644 --- a/internal/web/service/node_tag_sync_test.go +++ b/internal/web/service/node_tag_sync_test.go @@ -46,7 +46,7 @@ func TestSetRemoteTraffic_KeepsInboundOnPrefixMismatch(t *testing.T) { } svc := InboundService{} - if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) } @@ -83,7 +83,7 @@ func TestSetRemoteTraffic_AdoptsCompatibleOriginAliasWithoutDuplicate(t *testing Tag: "already-deployed", Enable: true, Port: 8443, Protocol: model.VLESS, Settings: `{"clients":[]}`, Up: 11, Down: 22, }}} - if _, err := (&InboundService{}).setRemoteTrafficLocked(nodeID, snap, false); err != nil { + if _, err := (&InboundService{}).setRemoteTrafficLocked(nodeID, snap, false, false); err != nil { t.Fatalf("setRemoteTrafficLocked: %v", err) }