From f172bc3f10f5190929ace2351c9b6a85423c4dd4 Mon Sep 17 00:00:00 2001 From: 25f2002654 Date: Fri, 23 Oct 2026 22:34:46 +0530 Subject: [PATCH] fix(node-sync): do not latch node disable when master allows (#6870) (#6876) * fix(node-sync): do not latch node disable when master allows (#6870) nodeDisableIsStale trusted any node-reported enable=false whose limits matched the master's, treating it as a genuine verdict and latching it on the master. But matching limits no longer prove the verdict: a single stray false (a depleted snapshot the master later re-enabled, a stale disableInvalidClients hit, or a push that lagged a quota reset) leaves the client stuck disabled on the node, and every master tick flips the client's state again. This is the loop reported in #6870. Make nodeDisableIsStale a single rule: stale iff the master's own deadline and quota still permit the client. A genuine combined-quota or expiry disable (masterLimitsAllowClient == false) still latches, preserving #4917. Mirror the same rule in ClientTrafficEnableMergeExpr so the SQL UPDATE and the Go in-memory replay cannot drift. * fix(node-sync): keep node usage over quota latching a node disable Dropping the matching-limits latch (#6870) let the master re-enable a client the node had genuinely counted past its quota whenever the master's own counters lagged, e.g. after a seeded-at-zero adoption, reopening #4917. A node's enable=false now stays latched when the node's own usage already spends the master's quota; otherwise the master's limits decide, so a stray disable no longer loops. Restores TestNodeQuotaDisable_SameExpiryStillLatches as it was on main. --------- Co-authored-by: Alireza Ahmadi --- internal/database/dialect.go | 16 ++--- internal/web/service/inbound_node.go | 21 +++--- .../service/node_client_expiry_sync_test.go | 67 +++++++++++++++---- 3 files changed, 72 insertions(+), 32 deletions(-) diff --git a/internal/database/dialect.go b/internal/database/dialect.go index 6aa009026..53ffe77f4 100644 --- a/internal/database/dialect.go +++ b/internal/database/dialect.go @@ -37,23 +37,23 @@ 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). +// ClientTrafficEnableMergeExpr: placeholders nodeEnable, now, deltaUp, deltaDown, +// nodeUp, nodeDown. Mirrors nodeDisableIsStale (#6228 / #4917 / #6870). func ClientTrafficEnableMergeExpr() string { if IsPostgres() { 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 + WHEN (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) + AND (total <= 0 OR up + ? + down + ? < total) + AND (total <= 0 OR CAST(? AS BIGINT) + CAST(? AS BIGINT) < total) THEN enable::boolean ELSE false 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 + WHEN (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) + AND (total <= 0 OR up + ? + down + ? < total) + AND (total <= 0 OR CAST(? AS BIGINT) + CAST(? AS BIGINT) < total) THEN enable ELSE 0 END` } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 2b05f8d15..d2ba6bbe8 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -226,7 +226,7 @@ func (s *InboundService) upsertNodeBaseline(tx *gorm.DB, nodeID int, email strin // 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 { +func (s *InboundService) mergeNodeClientTraffic(tx *gorm.DB, cs xray.ClientTraffic, canon nodeTrafficCounter, frozen bool, deltaUp, deltaDown, now int64) error { if frozen { return tx.Exec( fmt.Sprintf( @@ -254,7 +254,7 @@ func (s *InboundService) mergeNodeClientTraffic(tx *gorm.DB, cs xray.ClientTraff database.GreatestExpr("last_online", "?"), ), deltaUp, deltaDown, - cs.Enable, cs.ExpiryTime, cs.Total, now, deltaUp, deltaDown, + cs.Enable, now, deltaUp, deltaDown, canon.Up, canon.Down, cs.Total, cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetWeekday, cs.LastOnline, cs.Email, @@ -285,15 +285,14 @@ func masterLimitsAllowClient(master *xray.ClientTraffic, now, deltaUp, deltaDown 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 { +// nodeDisableIsStale reports a node-reported enable=false the master's limits no +// longer justify, so it must not latch back (#6228 / #4917 / #6870). +func nodeDisableIsStale(master *xray.ClientTraffic, nodeUsed, 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 { + // A master seeded at zero on adoption undercounts; the node's own usage still binds (#4917). + if master.Total > 0 && nodeUsed >= master.Total { return false } return masterLimitsAllowClient(master, now, deltaUp, deltaDown) @@ -1117,7 +1116,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // Only a real latch to disabled is structural; one-way merge never // re-enables from the node. enableChanged := !clientFrozen && existing.Enable && !cs.Enable && - !nodeDisableIsStale(existing, cs, now, deltaUp, deltaDown) + !nodeDisableIsStale(existing, canon.Up+canon.Down, now, deltaUp, deltaDown) metaChanged := !clientFrozen && (existing.Total != cs.Total || existing.Reset != cs.Reset || existing.ResetWeekday != cs.ResetWeekday) if enableChanged || metaChanged || expiryChanged { structuralChange = true @@ -1175,7 +1174,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi next.Down = clampTrafficCounter(row.Down + deltaDown) next.LastOnline = max(row.LastOnline, cs.LastOnline) if !clientFrozen { - if !cs.Enable && !nodeDisableIsStale(row, cs, now, deltaUp, deltaDown) { + if !cs.Enable && !nodeDisableIsStale(row, canon.Up+canon.Down, now, deltaUp, deltaDown) { next.Enable = false } next.ExpiryTime = mergeActivationExpiry(row.ExpiryTime, cs.ExpiryTime) @@ -1185,7 +1184,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // 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 { + if err := s.mergeNodeClientTraffic(tx, cs, canon, clientFrozen, deltaUp, deltaDown, now); err != nil { return false, err } *row = next diff --git a/internal/web/service/node_client_expiry_sync_test.go b/internal/web/service/node_client_expiry_sync_test.go index bdd498c06..bf88b14fc 100644 --- a/internal/web/service/node_client_expiry_sync_test.go +++ b/internal/web/service/node_client_expiry_sync_test.go @@ -54,12 +54,6 @@ func TestNodeDisableIsStale(t *testing.T) { 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}, @@ -69,7 +63,7 @@ func TestNodeDisableIsStale(t *testing.T) { { 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}, + node: xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 60, Down: 50}, wantStale: true, }, { @@ -102,10 +96,21 @@ func TestNodeDisableIsStale(t *testing.T) { node: xray.ClientTraffic{ExpiryTime: -2592000000}, wantStale: true, }, + { + name: "matching limits, neither master nor node usage spends the quota (#6870)", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 10, Down: 10}, + node: xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 10, Down: 10}, + wantStale: true, + }, + { + name: "node usage spends the quota the master undercounts (#4917)", + master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 40, Down: 50}, + node: xray.ClientTraffic{ExpiryTime: lateAbs, Total: quota, Up: 60, Down: 50}, + }, } 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 { + if got := nodeDisableIsStale(c.master, c.node.Up+c.node.Down, now, c.deltaUp, c.deltaDown); got != c.wantStale { t.Fatalf("nodeDisableIsStale(...) = %v, want %v", got, c.wantStale) } }) @@ -389,8 +394,8 @@ func TestNodeStaleLift_MarksNodeDirty(t *testing.T) { } } -// A node disable decided on the master's own limits is genuine and must still -// one-way-merge enable=false onto the master (#4917). +// A node that counted past the quota latches enable=false even while the master, +// seeded at zero on adoption, still counts below it (#4917). func TestNodeQuotaDisable_SameExpiryStillLatches(t *testing.T) { db := initTrafficTestDB(t) createNodeInbound(t, db, 1, "n1-in", 41001) @@ -410,6 +415,27 @@ func TestNodeQuotaDisable_SameExpiryStillLatches(t *testing.T) { } } +// A stray node enable=false that neither the node's usage nor the master's limits +// justify must not latch onto the master and loop back to the node (#6870). +func TestNodeStrayDisable_DoesNotLatch(t *testing.T) { + db := initTrafficTestDB(t) + createNodeInbound(t, db, 1, "n1-in", 41001) + svc := &InboundService{} + + const email = "stray" + 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: 20, Down: 20, Total: 100, ExpiryTime: expiry, Enable: false, + }) + if got := readTraffic(t, db, email); !got.Enable { + t.Fatal("stray node disable within quota and expiry latched master enable off (#6870)") + } +} + // 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) { @@ -870,6 +896,7 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) { masterUp, masterDown, tot int64 deltaUp, deltaDown int64 nodeExpiry, nodeTotal int64 + nodeUp, nodeDown int64 nodeEnable bool wantExpiry int64 wantEnable bool @@ -920,6 +947,20 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) { nodeExpiry: earlyAbs, nodeTotal: 100, nodeEnable: false, wantExpiry: lateAbs, wantEnable: false, }, + { + name: "matching limits, master and node under quota (#6870)", + masterExpiry: lateAbs, masterEnable: true, + masterUp: 10, masterDown: 10, tot: 100, + nodeExpiry: lateAbs, nodeTotal: 100, nodeUp: 10, nodeDown: 10, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: true, + }, + { + name: "node usage spends the quota the master undercounts (#4917)", + masterExpiry: lateAbs, masterEnable: true, + masterUp: 40, masterDown: 50, tot: 100, + nodeExpiry: lateAbs, nodeTotal: 100, nodeUp: 60, nodeDown: 50, nodeEnable: false, + wantExpiry: lateAbs, wantEnable: false, + }, } enableExpr := database.ClientTrafficEnableMergeExpr() @@ -939,8 +980,8 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) { } 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) { + node := xray.ClientTraffic{ExpiryTime: c.nodeExpiry, Total: c.nodeTotal, Up: c.nodeUp, Down: c.nodeDown} + if !c.nodeEnable && !nodeDisableIsStale(master, node.Up+node.Down, now, c.deltaUp, c.deltaDown) { wantEnable = false } if wantExpiry != c.wantExpiry || wantEnable != c.wantEnable { @@ -953,7 +994,7 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) { `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.nodeEnable, now, c.deltaUp, c.deltaDown, node.Up, node.Down, c.nodeExpiry, rowEmail, ).Error; err != nil {