fix(traffic): keep node reset propagation out of the serial traffic writer

ResetAllTraffics and ResetInboundTraffic performed their remote-node reset HTTP
calls inside submitTrafficWrite. Each call can block up to the remote timeout,
and Reset All Traffics loops every node serially, so the single traffic-writer
goroutine was held for seconds — long enough that the concurrent 5s traffic poll
timed out submitting its own write and dropped the deltas it had already drained
from xray. Do the DB reset inside the writer, then propagate to the nodes after
it returns, matching how the mtproto quota reset is already sequenced.

internal/web/service/inbound_traffic.go

@@ -657,6 +657,7 @@ func (s *InboundService) ResetAllTraffics() error {
 		return s.resetAllTrafficsLocked()
 	})
 	if err == nil {
+		s.propagateResetAllTrafficsToNodes()
 		s.resetAllMtprotoQuotas()
 	}
 	return err
@@ -666,52 +667,53 @@ func (s *InboundService) resetAllTrafficsLocked() error {
 	db := database.GetDB()
 	now := time.Now().UnixMilli()
 
-	if err := db.Model(model.Inbound{}).
+	return db.Model(model.Inbound{}).
 		Where("user_id > ?", 0).
 		Updates(map[string]any{
 			"up":                      0,
 			"down":                    0,
 			"last_traffic_reset_time": now,
-		}).Error; err != nil {
-		return err
-	}
+		}).Error
+}
 
+// propagateResetAllTrafficsToNodes tells every node to zero its own counters.
+// Kept OUT of the traffic-writer transaction: each remote call can block up to
+// remoteHTTPTimeout, and holding the single serial writer across N such calls
+// stalls traffic accounting and drops the deltas of every concurrent poll.
+func (s *InboundService) propagateResetAllTrafficsToNodes() {
 	nodes, err := (&NodeService{}).GetAll()
-	if err == nil {
-		for _, node := range nodes {
-			if rt, err := runtime.GetManager().RuntimeFor(&node.Id); err == nil {
-				if e := rt.ResetAllTraffics(context.Background()); e != nil {
-					logger.Warning("ResetAllTraffics: remote propagation to", rt.Name(), "failed:", e)
-				}
+	if err != nil {
+		return
+	}
+	for _, node := range nodes {
+		if rt, err := runtime.GetManager().RuntimeFor(&node.Id); err == nil {
+			if e := rt.ResetAllTraffics(context.Background()); e != nil {
+				logger.Warning("ResetAllTraffics: remote propagation to", rt.Name(), "failed:", e)
 			}
 		}
 	}
-
-	return nil
 }
 
 func (s *InboundService) ResetInboundTraffic(id int) error {
-	return submitTrafficWrite(func() error {
-		db := database.GetDB()
-		if err := db.Model(model.Inbound{}).
+	if err := submitTrafficWrite(func() error {
+		return database.GetDB().Model(model.Inbound{}).
 			Where("id = ?", id).
-			Updates(map[string]any{"up": 0, "down": 0}).Error; err != nil {
-			return err
-		}
+			Updates(map[string]any{"up": 0, "down": 0}).Error
+	}); err != nil {
+		return err
+	}
 
-		inbound, err := s.GetInbound(id)
-		if err == nil && inbound != nil && inbound.NodeID != nil {
-			if rt, rterr := s.runtimeFor(inbound); rterr == nil {
-				if e := rt.ResetInboundTraffic(context.Background(), inbound); e != nil {
-					logger.Warning("ResetInboundTraffic: remote propagation to", rt.Name(), "failed:", e)
-				}
-			} else {
-				logger.Warning("ResetInboundTraffic: runtime lookup failed:", rterr)
+	inbound, err := s.GetInbound(id)
+	if err == nil && inbound != nil && inbound.NodeID != nil {
+		if rt, rterr := s.runtimeFor(inbound); rterr == nil {
+			if e := rt.ResetInboundTraffic(context.Background(), inbound); e != nil {
+				logger.Warning("ResetInboundTraffic: remote propagation to", rt.Name(), "failed:", e)
 			}
+		} else {
+			logger.Warning("ResetInboundTraffic: runtime lookup failed:", rterr)
 		}
-
-		return nil
-	})
+	}
+	return nil
 }
 
 func (s *InboundService) DelDepletedClients(id int) (err error) {

internal/web/service/reset_writer_test.go

@@ -0,0 +1,71 @@
+package service
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
+)
+
+type blockingResetRuntime struct {
+	fakeNodeRuntime
+	reached chan struct{}
+	release chan struct{}
+}
+
+func (b *blockingResetRuntime) ResetAllTraffics(context.Context) error {
+	close(b.reached)
+	<-b.release
+	return nil
+}
+
+// A "Reset All Traffics" that propagates to nodes must not hold the single serial
+// traffic writer across the remote HTTP calls: each can block for seconds, and
+// stalling the writer drops every concurrent poll's traffic deltas.
+func TestResetAllTrafficsDoesNotBlockWriterOnNodeCall(t *testing.T) {
+	db := initTrafficTestDB(t)
+	resetTrafficWriterForTest(t)
+	StartTrafficWriter()
+
+	mgr := runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }})
+	runtime.SetManager(mgr)
+	t.Cleanup(func() { runtime.SetManager(nil) })
+
+	node := &model.Node{Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Status: "online"}
+	if err := db.Create(node).Error; err != nil {
+		t.Fatalf("create node: %v", err)
+	}
+	fake := &blockingResetRuntime{reached: make(chan struct{}), release: make(chan struct{})}
+	mgr.SetRuntimeOverride(node.Id, fake)
+
+	done := make(chan error, 1)
+	go func() { done <- (&InboundService{}).ResetAllTraffics() }()
+
+	select {
+	case <-fake.reached:
+	case <-time.After(3 * time.Second):
+		close(fake.release)
+		t.Fatal("node ResetAllTraffics was never reached")
+	}
+
+	writerFree := make(chan error, 1)
+	go func() { writerFree <- submitTrafficWrite(func() error { return nil }) }()
+	select {
+	case err := <-writerFree:
+		if err != nil {
+			close(fake.release)
+			t.Fatalf("concurrent writer submit failed: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		close(fake.release)
+		<-done
+		t.Fatal("the serial traffic writer was blocked by a node reset HTTP call")
+	}
+
+	close(fake.release)
+	if err := <-done; err != nil {
+		t.Fatalf("ResetAllTraffics: %v", err)
+	}
+}