fix(tuic): client speed display and certificate button layout (#6723)

* fix(tuic): restore client speed and certificate layout

* docs(tuic): clarify native runtime and protocol behavior

* fix(websocket): preserve traffic updates from independent sources

* fix(docs): sync websocket traffic schema and drop restating TUIC tests

docs/public/openapi.json still described the old traffic event, without
clientTrafficSource/clientTrafficIntervalMs or the TUIC oneOf branch, so
the docs site showed a payload the panel no longer sends. No check
covers that copy.

The TUIC certificate layout test and the TUIC speed payload test only
read back the literals the code writes, so neither could fail on a real
regression. Both are removed, along with the className that existed only
for the layout test.

---------

Co-authored-by: MHSanaei <ho3ein.sanaei@gmail.com>
This commit is contained in:
Egor
2026-10-05 22:46:46 +05:00
committed by GitHub
parent e897b0957a
commit aacfaebab8
14 changed files with 310 additions and 50 deletions
+37 -1
View File
@@ -2,16 +2,22 @@ package job
import (
"fmt"
"sync"
"time"
"github.com/mhsanaei/3x-ui/v3/internal/logger"
"github.com/mhsanaei/3x-ui/v3/internal/tuic"
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
"github.com/mhsanaei/3x-ui/v3/internal/web/websocket"
"github.com/mhsanaei/3x-ui/v3/internal/xray"
)
const defaultTuicSpeedSampleInterval = 10 * time.Second
type TuicJob struct {
inboundService service.InboundService
inboundService service.InboundService
runMu sync.Mutex
lastSpeedSample time.Time
}
func NewTuicJob() *TuicJob {
@@ -19,6 +25,9 @@ func NewTuicJob() *TuicJob {
}
func (j *TuicJob) Run() {
j.runMu.Lock()
defer j.runMu.Unlock()
tuicJournalMu.Lock()
journalErr := j.replayTuicJournal()
tuicJournalMu.Unlock()
@@ -44,20 +53,28 @@ func (j *TuicJob) Run() {
onlineEmails, _ := mgr.GetActiveClients(30 * time.Second)
clientTraffics := aggregateTuicClientTraffic(clientDeltas, onlineEmails)
sampledAt := time.Now()
sampleInterval := tuicSpeedSampleInterval(j.lastSpeedSample, sampledAt)
// Inbound total traffic is already metered through the loopback SOCKS relay
// by xray_traffic_job (matching mtproto); only per-client deltas are submitted here.
persisted := true
if len(clientTraffics) > 0 {
needRestart, _, err := j.inboundService.AddTraffic(nil, clientTraffics)
if err != nil {
logger.Warning("tuic job: add traffic failed:", err)
mgr.RequeueClientTraffic(clientDeltas)
persisted = false
} else if needRestart {
if desired, err := j.inboundService.DesiredTuicInstances(); err == nil {
mgr.Reconcile(desired)
}
}
}
if persisted {
websocket.BroadcastTraffic(tuicSpeedPayload(clientTraffics, sampleInterval))
j.lastSpeedSample = sampledAt
}
if len(onlineEmails) > 0 {
if err := j.inboundService.BumpClientsLastOnline(onlineEmails); err != nil {
@@ -68,6 +85,25 @@ func (j *TuicJob) Run() {
j.inboundService.RefreshLocalOnlineClients(onlineEmails, activeTags)
}
func tuicSpeedSampleInterval(previous, current time.Time) time.Duration {
if previous.IsZero() || !current.After(previous) {
return defaultTuicSpeedSampleInterval
}
return current.Sub(previous)
}
func tuicSpeedPayload(clientTraffics []*xray.ClientTraffic, sampleInterval time.Duration) map[string]any {
intervalMs := sampleInterval.Milliseconds()
if intervalMs < 1 {
intervalMs = 1
}
return map[string]any{
"clientTraffics": clientTraffics,
"clientTrafficSource": "tuic",
"clientTrafficIntervalMs": intervalMs,
}
}
// FlushStoppedTraffic persists counters drained when the TUIC manager stops its
// listeners. Call it after scheduled jobs have stopped and before the traffic
// writer shuts down.
+11
View File
@@ -180,3 +180,14 @@ func TestAggregateTuicClientTrafficPreservesStableIdentityAcrossEmailRename(t *t
t.Fatalf("aggregate counters = (%d,%d), want (40,60)", got[0].Up, got[0].Down)
}
}
func TestTuicSpeedSampleIntervalUsesElapsedPollTime(t *testing.T) {
current := time.Date(2026, time.October, 3, 12, 0, 10, 0, time.UTC)
previous := current.Add(-12 * time.Second)
if got := tuicSpeedSampleInterval(previous, current); got != 12*time.Second {
t.Fatalf("sample interval = %s, want 12s", got)
}
if got := tuicSpeedSampleInterval(time.Time{}, current); got != defaultTuicSpeedSampleInterval {
t.Fatalf("initial sample interval = %s, want %s", got, defaultTuicSpeedSampleInterval)
}
}
+3 -1
View File
@@ -94,10 +94,12 @@ func NewHub() *Hub {
}
}
// Traffic messages carry independent partial updates from Xray, TUIC, and node
// polling jobs. Throttling by message type would silently discard one source
// when two jobs publish within the throttle window.
var throttledMessageTypes = map[MessageType]struct{}{
MessageTypeInbounds: {},
MessageTypeOutbounds: {},
MessageTypeTraffic: {},
MessageTypeClientStats: {},
}
+43 -14
View File
@@ -81,21 +81,45 @@ func TestHub_BroadcastDeliversToClient(t *testing.T) {
waitClientCount(t, h, 1)
h.Broadcast(MessageTypeStatus, map[string]string{"k": "v"})
select {
case raw := <-c.Send:
var m Message
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatalf("payload is not valid JSON: %v\n%s", err, raw)
var message Message
if err := json.Unmarshal(raw, &message); err != nil {
t.Fatalf("payload is not valid JSON: %v", err)
}
if m.Type != MessageTypeStatus {
t.Fatalf("Type = %q, want %q", m.Type, MessageTypeStatus)
if message.Type != MessageTypeStatus {
t.Fatalf("message type = %q, want %q", message.Type, MessageTypeStatus)
}
if m.Time == 0 {
if message.Time == 0 {
t.Fatal("Time should be set to a non-zero unix-millis value")
}
case <-time.After(500 * time.Millisecond):
t.Fatal("timed out waiting for broadcast to reach client")
t.Fatal("timed out waiting for status broadcast to reach client")
}
for _, source := range []string{"tuic", "xray"} {
h.Broadcast(MessageTypeTraffic, map[string]string{"source": source})
}
for _, wantSource := range []string{"tuic", "xray"} {
select {
case raw := <-c.Send:
var message struct {
Type MessageType `json:"type"`
Payload map[string]string `json:"payload"`
}
if err := json.Unmarshal(raw, &message); err != nil {
t.Fatalf("traffic event is not valid JSON: %v", err)
}
if message.Type != MessageTypeTraffic {
t.Fatalf("message type = %q, want %q", message.Type, MessageTypeTraffic)
}
if got := message.Payload["source"]; got != wantSource {
t.Fatalf("traffic source = %q, want %q", got, wantSource)
}
case <-time.After(500 * time.Millisecond):
t.Fatalf("timed out waiting for %q traffic event", wantSource)
}
}
}
@@ -156,23 +180,28 @@ func TestHub_ShouldThrottle(t *testing.T) {
t.Fatal("non-gated message type should never throttle on second call")
}
if h.shouldThrottle(MessageTypeTraffic) {
if h.shouldThrottle(MessageTypeInbounds) {
t.Fatal("first call for gated type should not throttle")
}
if !h.shouldThrottle(MessageTypeTraffic) {
if !h.shouldThrottle(MessageTypeInbounds) {
t.Fatal("immediate second call for gated type should throttle")
}
for i := range 2 {
if h.shouldThrottle(MessageTypeTraffic) {
t.Fatalf("traffic event %d must not be throttled", i+1)
}
}
}
func TestHub_ShouldThrottle_DistinctTypesIndependent(t *testing.T) {
h := NewHub()
defer h.Stop()
if h.shouldThrottle(MessageTypeTraffic) {
t.Fatal("first Traffic call should not throttle")
}
if h.shouldThrottle(MessageTypeInbounds) {
t.Fatal("first Inbounds call should not throttle even after Traffic")
t.Fatal("first Inbounds call should not throttle")
}
if h.shouldThrottle(MessageTypeOutbounds) {
t.Fatal("first Outbounds call should not throttle even after Inbounds")
}
}