fix(node): settle each queued node reset on its own

A per-client reset the node never accepted stopped the delivery loop:
resets queued before it were re-sent on every sync, each time wiping the
usage made since, and those after it were never sent. Each reset now
settles by itself, a refused one moves to the back of the queue, and the
loop goes on until the context is spent.

A node also delivered its master's reset to its own sub-nodes inline, so
one silent sub-node held the master's request until it timed out and was
replayed; the inline attempt now waits at most a second and the rest
finishes, or stays queued, in the background.

internal/web/service/node_reset_queue.go

@@ -8,6 +8,7 @@ import (
 	"github.com/mhsanaei/3x-ui/v3/internal/database"
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
 	"github.com/mhsanaei/3x-ui/v3/internal/logger"
+	"github.com/mhsanaei/3x-ui/v3/internal/util/common"
 	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
 
 	"gorm.io/gorm"
@@ -94,12 +95,13 @@ func (s *InboundService) DeliverNodeResets(ctx context.Context, nodeID int, rt r
 	defer lock.(*sync.Mutex).Unlock()
 	db := database.GetDB()
 	var rows []model.NodePendingReset
-	if err := db.Where("node_id = ?", nodeID).Order("id").Find(&rows).Error; err != nil {
+	if err := db.Where("node_id = ?", nodeID).Order("queued_at, id").Find(&rows).Error; err != nil {
 		return err
 	}
 	if len(rows) == 0 {
 		return nil
 	}
+	var failed error
 	bulk, canBulk := rt.(nodeBulkResetter)
 	for start := 0; start < len(rows); start += sqlInChunk {
 		batch := rows[start:min(start+sqlInChunk, len(rows))]
@@ -107,27 +109,41 @@ func (s *InboundService) DeliverNodeResets(ctx context.Context, nodeID int, rt r
 		for i := range batch {
 			emails[i] = batch[i].Email
 		}
-		var err error
 		if canBulk && len(batch) > nodeBulkPushThreshold {
-			err = bulk.ResetClientTraffics(ctx, emails)
+			if err := bulk.ResetClientTraffics(ctx, emails); err != nil {
-		} else {
+				return err
-			for _, email := range emails {
+			}
-				if err = rt.ResetClientTraffic(ctx, nil, email); err != nil {
+			for i := range batch {
-					break
+				if err := settleNodeReset(db, &batch[i]); err != nil {
+					return err
 				}
 			}
+			continue
 		}
-		if err != nil {
+		// Each reset settles on its own: one the node never accepts must neither hold back
-			return err
+		// the rest nor have them re-sent, which would wipe the usage made in between.
-		}
 		for i := range batch {
-			if err := db.Where("id = ? AND queued_at = ?", batch[i].Id, batch[i].QueuedAt).
+			if err := rt.ResetClientTraffic(ctx, nil, batch[i].Email); err != nil {
-				Delete(&model.NodePendingReset{}).Error; err != nil {
+				failed = err
+				// To the back of the queue, so it cannot spend the next attempt's budget first.
+				db.Model(&model.NodePendingReset{}).Where("id = ? AND queued_at = ?", batch[i].Id, batch[i].QueuedAt).
+					Update("queued_at", time.Now().UnixNano())
+				if ctx.Err() != nil {
+					return failed
+				}
+				continue
+			}
+			if err := settleNodeReset(db, &batch[i]); err != nil {
 				return err
 			}
 		}
 	}
-	return nil
+	return failed
+}
+
+// settleNodeReset drops a delivered row unless it was re-queued while it was being sent.
+func settleNodeReset(db *gorm.DB, row *model.NodePendingReset) error {
+	return db.Where("id = ? AND queued_at = ?", row.Id, row.QueuedAt).Delete(&model.NodePendingReset{}).Error
 }
 
 // deliverNodeResetsNow tries each node once right after a reset commits; what
@@ -137,16 +153,28 @@ func (s *InboundService) deliverNodeResetsNow(nodeIDs []int) {
 	if mgr == nil || len(nodeIDs) == 0 {
 		return
 	}
-	fanoutInboundResults(nodeIDs, nodeFanoutConcurrency, func(i int) struct{} {
+	done := make(chan struct{})
-		rt, err := mgr.RuntimeFor(&nodeIDs[i])
+	common.GoRecover("node-reset-delivery", func() {
-		if err != nil {
+		defer close(done)
+		fanoutInboundResults(nodeIDs, nodeFanoutConcurrency, func(i int) struct{} {
+			rt, err := mgr.RuntimeFor(&nodeIDs[i])
+			if err != nil {
+				return struct{}{}
+			}
+			ctx, cancel := nodePushContext()
+			defer cancel()
+			if err := s.DeliverNodeResets(ctx, nodeIDs[i], rt); err != nil {
+				logger.Warning("reset delivery to", rt.Name(), "deferred to the next sync:", err)
+			}
 			return struct{}{}
-		}
+		})
-		ctx, cancel := nodePushContext()
-		defer cancel()
-		if err := s.DeliverNodeResets(ctx, nodeIDs[i], rt); err != nil {
-			logger.Warning("reset delivery to", rt.Name(), "deferred to the next sync:", err)
-		}
-		return struct{}{}
 	})
+	// A silent node must not hold the request that queued the reset (a node relaying its
+	// master's reset): the master would time out and replay it, wiping usage made since.
+	select {
+	case <-done:
+	case <-time.After(nodeResetInlineWait):
+	}
 }
+
+const nodeResetInlineWait = time.Second

internal/web/service/node_reset_undelivered_test.go

@@ -275,3 +275,69 @@ func TestConcurrentNodeResetDeliveriesSendOnce(t *testing.T) {
 		t.Fatalf("node received resets %v, want exactly [once]", got)
 	}
 }
+
+// selectiveResetRuntime refuses the resets of the emails in refuse.
+type selectiveResetRuntime struct {
+	resetRecordingRuntime
+	refuse map[string]bool
+}
+
+func (r *selectiveResetRuntime) ResetClientTraffic(ctx context.Context, ib *model.Inbound, email string) error {
+	if r.refuse[email] {
+		return errors.New("refused")
+	}
+	return r.resetRecordingRuntime.ResetClientTraffic(ctx, ib, email)
+}
+
+// One reset the node never accepts must not hold back the ones queued after it, nor
+// have the ones before it re-sent (each re-send wipes the usage made in between).
+func TestNodeResetsSettleEachClientOnItsOwn(t *testing.T) {
+	nodeID, _, _ := setupRecordingNode(t, false)
+	now := time.Now().UnixNano()
+	for i, email := range []string{"a", "stuck", "c"} {
+		if err := database.GetDB().Create(&model.NodePendingReset{NodeId: nodeID, Email: email, QueuedAt: now + int64(i)}).Error; err != nil {
+			t.Fatalf("queue %s: %v", email, err)
+		}
+	}
+	rt := &selectiveResetRuntime{refuse: map[string]bool{"stuck": true}}
+	for range 2 {
+		_ = (&InboundService{}).DeliverNodeResets(context.Background(), nodeID, rt)
+	}
+	if got := rt.delivered(); !slices.Equal(got, []string{"a", "c"}) {
+		t.Fatalf("node received %v, want a and c once each", got)
+	}
+	if left := pendingResetEmails(t, nodeID); !slices.Equal(left, []string{"stuck"}) {
+		t.Fatalf("still queued: %v, want only the refused reset", left)
+	}
+}
+
+// stallingResetRuntime never answers a reset until its context ends.
+type stallingResetRuntime struct{ fakeNodeRuntime }
+
+func (*stallingResetRuntime) ResetClientTraffic(ctx context.Context, _ *model.Inbound, _ string) error {
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// A node relays its master's reset to its own sub-nodes inline; one silent sub-node must
+// not hold that request until the master times out and sends the reset again.
+func TestNodeResetDeliveryDoesNotHoldTheRequestBehindASilentNode(t *testing.T) {
+	setupBulkDB(t)
+	mgr := useTestRuntimeManager(t)
+	node := &model.Node{Name: "silent", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Status: "online"}
+	if err := database.GetDB().Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	mgr.SetRuntimeOverride(node.Id, &stallingResetRuntime{})
+	ib := nodeInbound(t, node.Id, 41912, []model.Client{{Email: "held", ID: "11111111-1111-1111-1111-1111111111ab", Enable: true}})
+	if err := (&InboundService{}).AddClientStat(database.GetDB(), ib.Id, &model.Client{Email: "held", Enable: true}); err != nil {
+		t.Fatalf("AddClientStat: %v", err)
+	}
+	start := time.Now()
+	if _, err := (&InboundService{}).ResetClientTraffic(ib.Id, "held"); err != nil {
+		t.Fatalf("ResetClientTraffic: %v", err)
+	}
+	if took := time.Since(start); took > 2*time.Second {
+		t.Fatalf("the reset request took %v behind a silent node", took)
+	}
+}