From d005d89a8bd686460411e3e363eb2cc0495c94ac Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 31 Oct 2026 14:01:45 +0100 Subject: [PATCH] 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 | 74 +++++++++++++------ .../service/node_reset_undelivered_test.go | 66 +++++++++++++++++ 2 files changed, 117 insertions(+), 23 deletions(-) diff --git a/internal/web/service/node_reset_queue.go b/internal/web/service/node_reset_queue.go index 5772a23db..1b0d41a75 100644 --- a/internal/web/service/node_reset_queue.go +++ b/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) - } else { - for _, email := range emails { - if err = rt.ResetClientTraffic(ctx, nil, email); err != nil { - break + if err := bulk.ResetClientTraffics(ctx, emails); err != nil { + return err + } + for i := range batch { + if err := settleNodeReset(db, &batch[i]); err != nil { + return err } } + continue } - if err != nil { - return err - } + // Each reset settles on its own: one the node never accepts must neither hold back + // 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). - Delete(&model.NodePendingReset{}).Error; err != nil { + if err := rt.ResetClientTraffic(ctx, nil, batch[i].Email); 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{} { - rt, err := mgr.RuntimeFor(&nodeIDs[i]) - if err != nil { + done := make(chan struct{}) + common.GoRecover("node-reset-delivery", func() { + 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 diff --git a/internal/web/service/node_reset_undelivered_test.go b/internal/web/service/node_reset_undelivered_test.go index 535157b37..f2f71ed60 100644 --- a/internal/web/service/node_reset_undelivered_test.go +++ b/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) + } +}