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 <alireza7@gmail.com>
This commit is contained in:
25f2002654
2026-10-23 22:34:46 +05:30
committed by GitHub
parent e2184777a0
commit f172bc3f10
3 changed files with 72 additions and 32 deletions
+8 -8
View File
@@ -37,23 +37,23 @@ func GreatestExpr(a, b string) string {
return fmt.Sprintf("MAX(%s, %s)", a, b) return fmt.Sprintf("MAX(%s, %s)", a, b)
} }
// ClientTrafficEnableMergeExpr: placeholders nodeEnable, nodeExpiry, nodeTotal, // ClientTrafficEnableMergeExpr: placeholders nodeEnable, now, deltaUp, deltaDown,
// now, deltaUp, deltaDown. Mirrors nodeDisableIsStale (#6228 / #4917). // nodeUp, nodeDown. Mirrors nodeDisableIsStale (#6228 / #4917 / #6870).
func ClientTrafficEnableMergeExpr() string { func ClientTrafficEnableMergeExpr() string {
if IsPostgres() { if IsPostgres() {
return `CASE return `CASE
WHEN ?::boolean THEN enable::boolean WHEN ?::boolean THEN enable::boolean
WHEN (expiry_time <> CAST(? AS BIGINT) OR total <> CAST(? AS BIGINT)) WHEN (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT))
AND (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) AND (total <= 0 OR up + ? + down + ? < total)
AND (total <= 0 OR up + ? + down + ? < total) THEN enable::boolean AND (total <= 0 OR CAST(? AS BIGINT) + CAST(? AS BIGINT) < total) THEN enable::boolean
ELSE false ELSE false
END` END`
} }
return `CASE return `CASE
WHEN ? THEN enable WHEN ? THEN enable
WHEN (expiry_time <> CAST(? AS BIGINT) OR total <> CAST(? AS BIGINT)) WHEN (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT))
AND (expiry_time <= 0 OR expiry_time > CAST(? AS BIGINT)) AND (total <= 0 OR up + ? + down + ? < total)
AND (total <= 0 OR up + ? + down + ? < total) THEN enable AND (total <= 0 OR CAST(? AS BIGINT) + CAST(? AS BIGINT) < total) THEN enable
ELSE 0 ELSE 0
END` END`
} }
+10 -11
View File
@@ -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. // mergeNodeClientTraffic folds one node-reported client into its central row.
// While frozen (a push pending or just landed) only the counters may move. // 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 { if frozen {
return tx.Exec( return tx.Exec(
fmt.Sprintf( fmt.Sprintf(
@@ -254,7 +254,7 @@ func (s *InboundService) mergeNodeClientTraffic(tx *gorm.DB, cs xray.ClientTraff
database.GreatestExpr("last_online", "?"), database.GreatestExpr("last_online", "?"),
), ),
deltaUp, deltaDown, deltaUp, deltaDown,
cs.Enable, cs.ExpiryTime, cs.Total, now, deltaUp, deltaDown, cs.Enable, now, deltaUp, deltaDown, canon.Up, canon.Down,
cs.Total, cs.Total,
cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetWeekday, cs.ExpiryTime, cs.Reset, cs.ResetDay, cs.ResetWeekday,
cs.LastOnline, cs.Email, cs.LastOnline, cs.Email,
@@ -285,15 +285,14 @@ func masterLimitsAllowClient(master *xray.ClientTraffic, now, deltaUp, deltaDown
return true return true
} }
// nodeDisableIsStale reports an enable=false the node decided against limits the // nodeDisableIsStale reports a node-reported enable=false the master's limits no
// master has since changed, so it must not latch back (#6228 / #4917). // longer justify, so it must not latch back (#6228 / #4917 / #6870).
func nodeDisableIsStale(master *xray.ClientTraffic, node xray.ClientTraffic, now, deltaUp, deltaDown int64) bool { func nodeDisableIsStale(master *xray.ClientTraffic, nodeUsed, now, deltaUp, deltaDown int64) bool {
if master == nil { if master == nil {
return false return false
} }
// Matching limits mean the node judged the client on the master's own terms: // A master seeded at zero on adoption undercounts; the node's own usage still binds (#4917).
// that verdict is genuine and still latches, as #4917 requires. if master.Total > 0 && nodeUsed >= master.Total {
if node.ExpiryTime == master.ExpiryTime && node.Total == master.Total {
return false return false
} }
return masterLimitsAllowClient(master, now, deltaUp, deltaDown) 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 // Only a real latch to disabled is structural; one-way merge never
// re-enables from the node. // re-enables from the node.
enableChanged := !clientFrozen && existing.Enable && !cs.Enable && 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) metaChanged := !clientFrozen && (existing.Total != cs.Total || existing.Reset != cs.Reset || existing.ResetWeekday != cs.ResetWeekday)
if enableChanged || metaChanged || expiryChanged { if enableChanged || metaChanged || expiryChanged {
structuralChange = true structuralChange = true
@@ -1175,7 +1174,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
next.Down = clampTrafficCounter(row.Down + deltaDown) next.Down = clampTrafficCounter(row.Down + deltaDown)
next.LastOnline = max(row.LastOnline, cs.LastOnline) next.LastOnline = max(row.LastOnline, cs.LastOnline)
if !clientFrozen { 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.Enable = false
} }
next.ExpiryTime = mergeActivationExpiry(row.ExpiryTime, cs.ExpiryTime) 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 // A multi-attached client is reported once per inbound and an idle one
// every tick: rewriting an unchanged row only holds its lock longer. // every tick: rewriting an unchanged row only holds its lock longer.
if next != *row { 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 return false, err
} }
*row = next *row = next
@@ -54,12 +54,6 @@ func TestNodeDisableIsStale(t *testing.T) {
wantStale bool wantStale bool
}{ }{
{name: "nil master", node: xray.ClientTraffic{ExpiryTime: earlyAbs}}, {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", name: "node still holds the pre-extension deadline",
master: &xray.ClientTraffic{ExpiryTime: lateAbs}, master: &xray.ClientTraffic{ExpiryTime: lateAbs},
@@ -69,7 +63,7 @@ func TestNodeDisableIsStale(t *testing.T) {
{ {
name: "node still holds the pre-top-up quota", name: "node still holds the pre-top-up quota",
master: &xray.ClientTraffic{ExpiryTime: lateAbs, Total: 2 * quota, Up: 60, Down: 50}, 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, wantStale: true,
}, },
{ {
@@ -102,10 +96,21 @@ func TestNodeDisableIsStale(t *testing.T) {
node: xray.ClientTraffic{ExpiryTime: -2592000000}, node: xray.ClientTraffic{ExpiryTime: -2592000000},
wantStale: true, 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 { for _, c := range cases {
t.Run(c.name, func(t *testing.T) { 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) 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 // A node that counted past the quota latches enable=false even while the master,
// one-way-merge enable=false onto the master (#4917). // seeded at zero on adoption, still counts below it (#4917).
func TestNodeQuotaDisable_SameExpiryStillLatches(t *testing.T) { func TestNodeQuotaDisable_SameExpiryStillLatches(t *testing.T) {
db := initTrafficTestDB(t) db := initTrafficTestDB(t)
createNodeInbound(t, db, 1, "n1-in", 41001) 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 // 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. // node's limits lag — otherwise genuine quota cuts would be skipped.
func TestNodeQuotaDisable_OlderExpiryStillLatchesWhenOverQuota(t *testing.T) { func TestNodeQuotaDisable_OlderExpiryStillLatchesWhenOverQuota(t *testing.T) {
@@ -870,6 +896,7 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) {
masterUp, masterDown, tot int64 masterUp, masterDown, tot int64
deltaUp, deltaDown int64 deltaUp, deltaDown int64
nodeExpiry, nodeTotal int64 nodeExpiry, nodeTotal int64
nodeUp, nodeDown int64
nodeEnable bool nodeEnable bool
wantExpiry int64 wantExpiry int64
wantEnable bool wantEnable bool
@@ -920,6 +947,20 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) {
nodeExpiry: earlyAbs, nodeTotal: 100, nodeEnable: false, nodeExpiry: earlyAbs, nodeTotal: 100, nodeEnable: false,
wantExpiry: lateAbs, wantEnable: 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() enableExpr := database.ClientTrafficEnableMergeExpr()
@@ -939,8 +980,8 @@ func TestClientTrafficMergeSQLMatchesHelpers(t *testing.T) {
} }
wantExpiry := mergeActivationExpiry(c.masterExpiry, c.nodeExpiry) wantExpiry := mergeActivationExpiry(c.masterExpiry, c.nodeExpiry)
wantEnable := c.masterEnable wantEnable := c.masterEnable
node := xray.ClientTraffic{ExpiryTime: c.nodeExpiry, Total: c.nodeTotal} node := xray.ClientTraffic{ExpiryTime: c.nodeExpiry, Total: c.nodeTotal, Up: c.nodeUp, Down: c.nodeDown}
if !c.nodeEnable && !nodeDisableIsStale(master, node, now, c.deltaUp, c.deltaDown) { if !c.nodeEnable && !nodeDisableIsStale(master, node.Up+node.Down, now, c.deltaUp, c.deltaDown) {
wantEnable = false wantEnable = false
} }
if wantExpiry != c.wantExpiry || wantEnable != c.wantEnable { 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 = ?`, `UPDATE client_traffics SET enable = %s, expiry_time = %s WHERE email = ?`,
enableExpr, expiryExpr, 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, c.nodeExpiry,
rowEmail, rowEmail,
).Error; err != nil { ).Error; err != nil {