diff --git a/internal/web/job/xray_traffic_inform_test.go b/internal/web/job/xray_traffic_inform_test.go new file mode 100644 index 000000000..9941d3d9d --- /dev/null +++ b/internal/web/job/xray_traffic_inform_test.go @@ -0,0 +1,97 @@ +package job + +import ( + "errors" + "io" + "net" + "testing" + "time" + + "github.com/valyala/fasthttp" +) + +// stallingListener accepts connections, reads whatever is sent and then never +// answers and never closes — the receiver shape that used to hold the traffic +// job open indefinitely (#6115). +func stallingListener(t *testing.T) (addr string, release func()) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + done := make(chan struct{}) + go func() { + defer close(done) + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + <-done + _ = c.Close() + }(conn) + go func(c net.Conn) { _, _ = io.Copy(io.Discard, c) }(conn) + } + }() + return ln.Addr().String(), func() { + _ = ln.Close() + <-done + } +} + +func TestExternalInformClient_StalledReceiverTimesOut(t *testing.T) { + addr, release := stallingListener(t) + defer release() + + request := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(request) + request.Header.SetMethod("POST") + request.Header.SetContentType("application/json; charset=UTF-8") + request.SetBody([]byte(`{"clientTraffics":[],"inboundTraffics":[]}`)) + request.SetRequestURI("http://" + addr + "/inform") + request.Header.SetConnectionClose() + + response := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(response) + + start := time.Now() + err := externalInformClient.DoTimeout(request, response, externalInformTimeout) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("a receiver that never answers must produce an error, not a completed request") + } + if !errors.Is(err, fasthttp.ErrTimeout) { + t.Errorf("want a timeout error, got %v", err) + } + // The whole point is that the call cannot outlast the 5s poll cadence. + if elapsed > externalInformTimeout+2*time.Second { + t.Errorf("call took %v, must be bounded by %v", elapsed, externalInformTimeout) + } +} + +func TestExternalInformClient_HasDeadlines(t *testing.T) { + if externalInformClient.ReadTimeout <= 0 || externalInformClient.WriteTimeout <= 0 { + t.Fatal("the traffic-notify client must carry read and write deadlines") + } + if externalInformTimeout >= 5*time.Second { + t.Errorf("the call budget (%v) must stay under the 5s traffic poll cadence", externalInformTimeout) + } +} + +// An idle panel posts nothing, which keeps a misbehaving receiver out of the +// job's path entirely for most ticks. +func TestInformSkippedWhenNothingToReport(t *testing.T) { + j := &XrayTrafficJob{} + done := make(chan struct{}) + go func() { + defer close(done) + j.informTrafficToExternalAPI(nil, nil) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("an empty payload must return before touching settings or the network") + } +} diff --git a/internal/web/job/xray_traffic_job.go b/internal/web/job/xray_traffic_job.go index 4d277519b..b071d541d 100644 --- a/internal/web/job/xray_traffic_job.go +++ b/internal/web/job/xray_traffic_job.go @@ -2,6 +2,7 @@ package job import ( "encoding/json" + "time" "github.com/mhsanaei/3x-ui/v3/internal/logger" "github.com/mhsanaei/3x-ui/v3/internal/web/service" @@ -28,6 +29,26 @@ type XrayTrafficJob struct { // refetch for the rest. const clientStatsSnapshotMaxClients = 5000 +// externalInformTimeout bounds the traffic-notify POST. Run() is scheduled +// every 5s under cron.SkipIfStillRunning, so an unbounded call to a receiver +// that accepts the connection and then stalls does not merely delay one +// notification: it holds the job, and every following tick is skipped for as +// long as the stall lasts. AddTraffic — quota enforcement, auto-renew — and +// the online/websocket work all sit in that same tick (#6115). +const externalInformTimeout = 3 * time.Second + +// externalInformClient is kept separate from fasthttp's shared default client +// so this endpoint's timeouts and connection handling cannot be influenced by, +// or influence, any other caller. Idempotent-call retries stay off on purpose: +// the payload carries per-tick deltas, so an attempt that reached the receiver +// but failed on the response leg would be counted twice if it were resent. +var externalInformClient = &fasthttp.Client{ + ReadTimeout: externalInformTimeout, + WriteTimeout: externalInformTimeout, + MaxIdleConnDuration: time.Minute, + MaxConnsPerHost: 4, +} + // NewXrayTrafficJob creates a new traffic collection job instance. func NewXrayTrafficJob() *XrayTrafficJob { return new(XrayTrafficJob) @@ -197,6 +218,9 @@ func (j *XrayTrafficJob) Run() { } func (j *XrayTrafficJob) informTrafficToExternalAPI(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) { + if len(inboundTraffics) == 0 && len(clientTraffics) == 0 { + return + } informURL, err := j.settingService.GetExternalTrafficInformURI() if err != nil { logger.Warning("get ExternalTrafficInformURI failed:", err) @@ -218,9 +242,10 @@ func (j *XrayTrafficJob) informTrafficToExternalAPI(inboundTraffics []*xray.Traf request.Header.SetContentType("application/json; charset=UTF-8") request.SetBody(requestBody) request.SetRequestURI(informURL) + request.Header.SetConnectionClose() response := fasthttp.AcquireResponse() defer fasthttp.ReleaseResponse(response) - if err := fasthttp.Do(request, response); err != nil { + if err := externalInformClient.DoTimeout(request, response, externalInformTimeout); err != nil { logger.Warning("POST ExternalTrafficInformURI failed:", err) } }