fix(node): stop stale expiry sync from undoing client extensions (#6228) (#6231)

* fix(node): stop stale expiry sync from undoing client extensions (#6228) After an expired client is extended on the master, a lagging node could overwrite client_traffics with an older absolute expiry and latch enable=false. Reject older absolute expiries on merge, ignore expiry-stale disables when the master is not over quota, lift stale lifecycle fields out of adopted settings, stamp reconcile fingerprints from the pre-lift node blob, and mark the node dirty so the next tick re-pushes. * fix(node): lockstep client_traffics expiry/enable SQL with review fixes (#6228) Make expiry merge keep any master absolute (node only activates when master is unset/duration). Include this tick's up/down deltas in the enable stale-disable quota check so a crossing-tick disable is not dropped. * fix(node): add settings absolute helper for renew/lift guards (#6228) Expose settingsClientAbsoluteExpiry so traffic merge can tell a real node auto-renew (settings+stats later) from lagging ClientStats after a master shorten. Trim lift godoc to the invariant. * fix(node): authority-aware lifecycle merge for multi-node sync (#6228) While config_dirty, accumulate traffic only — do not adopt node expiry/enable/total/reset (and preserve dipped baselines so a false renew cannot fire after clear). On clean ticks, master absolute expiry wins; node auto-renew still goes through nodeClientRenewed when settings also show the later deadline. Defer settings lifecycle lift until after traffic deltas land, align SyncInbound via applyMasterClientLifecycle, and avoid re-MarkNodeDirty when already dirty. * fix(node): clear config_dirty only after the post-reconcile traffic merge (#6228) After a successful ReconcileNode, keep the node dirty through the same tick's SetRemoteTraffic so lagging ClientStats cannot clobber the just-pushed master lifecycle, then ClearNodeDirty. * test(node): cover dirty-gate, master-absolute, and renew false-positives (#6228) Add regressions for extend/shorten while dirty, clean-sibling shorten, settings vs lagging disable, renew recovery after dirty, renew with matching settings, and shorten+Reset lagging stats not treated as renew. * fix(node): address the review findings on the lifecycle merge (#6228) The automated review on #6231 flagged a blocking regression and six smaller issues. All of them are fixed here. Blocking: making the master's absolute expiry always win left nodeClientRenewed as the only channel for a node-side auto-renew, and that required a counter dip. A client that used no traffic in the period never dips, so its renewal was dropped, the master kept the expired deadline and disableInvalidClients removed it with no way back (master-side autoRenewClients skips node inbounds). The node bumps reset_count on every renewal, so that counter is now an independent renewal signal and is persisted with the renewal so it keeps converging. The deferred ClearNodeDirty made every reconcile-success tick merge in dirty mode, which suppressed inbound adoption, new client_traffics rows, the orphan sweeps and the whole SyncInbound record loop -- and left the node dirty forever whenever SetRemoteTraffic errored. The clear goes back to where it was; a separate justPushed flag now freezes only the client lifecycle merge for the tick whose push just landed. staleNodeDisable only recognised a lagging disable by an older expiry, so a quota top-up (raise totalGB, leave the expiry alone) was re-latched to disabled by the next lagging snapshot -- the #6228 symptom on a second axis. The reviewer's suggestion of dropping the expiry precondition outright fails TestNodeQuotaDisable_SameExpiryStillLatches, because the master's own counters legitimately sit below a node's after a seeded-at-zero adoption. nodeDisableIsStale instead compares the limits the node judged the client against with the master's own: matching limits mean a genuine verdict that still latches (#4917), differing limits mean the node has not seen the master's change yet. It also now measures the master deadline against wall-clock now, so an expired master row no longer looks "extended" merely because the node's copy is older still. Also: the settings lift now writes enable in both directions, so a blob fetched before a master disable cannot carry enable=true back into central settings and on to the node; the renewal guard parses the inbound settings once per inbound instead of once per renewing client; the adoption loop only writes settings when they actually changed; and two comments that described mechanisms the code does not use were corrected. The test deadlines are now relative to the run: the merge compares against now, so fixed timestamps would have rotted into the wrong side of it. --------- Co-authored-by: mrchatam <287639636+mrchatam@users.noreply.github.com> Co-authored-by: Sanaei <ho3ein.sanaei@gmail.com>
2026-08-25 12:27:13 +00:00 · 2026-08-24 00:47:56 +03:30
parent da01b7637d
commit 6f7a305239
14 changed files with 1193 additions and 83 deletions
@@ -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 {