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
2026-10-16 02:44:59 +03:00 · 2026-10-14 00:37:00 +02:00
parent 6f1dd0d757
commit f8b72de995
5 changed files with 389 additions and 83 deletions
@@ -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
 			}