Merge remote-tracking branch 'upstream/main' into sync-3.6.0

# Conflicts:
#	.github/workflows/claude-bot.yml
#	.github/workflows/release.yml
#	DockerInit.sh
#	frontend/package-lock.json
#	frontend/package.json
#	frontend/src/hooks/useClients.ts
#	frontend/src/layouts/AppSidebar.tsx
#	frontend/src/main.tsx
#	internal/config/version
#	internal/database/model/model.go
#	internal/web/service/client_wireguard.go
#	internal/web/service/inbound.go
This commit is contained in:
Kuzz007
2026-08-01 21:59:29 +03:00
250 changed files with 12904 additions and 5650 deletions
@@ -0,0 +1,34 @@
package job
import "testing"
// A bind that succeeds with an unusable search returns (empty, nil); the only
// guard was `err != nil`, so that answer detached the entire inbound.
func TestAutoDeleteAllowed(t *testing.T) {
cases := []struct {
name string
previous int64
fetched int
want bool
}{
{"empty fetch on a fresh job is refused", 0, 0, false},
{"empty fetch after a healthy sync is refused", 500, 0, false},
{"full fetch on a fresh job is allowed", 0, 500, true},
{"steady fetch is allowed", 500, 500, true},
{"growth is allowed", 500, 900, true},
{"shrink above the retention floor is allowed", 500, 250, true},
{"shrink below the retention floor is refused", 500, 249, false},
{"collapse to a single user is refused", 500, 1, false},
{"single user with no history is allowed", 0, 1, true},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
j := NewLdapSyncJob()
j.lastFlagCount.Store(c.previous)
if got := j.autoDeleteSafeForFetch(c.fetched); got != c.want {
t.Fatalf("autoDeleteSafeForFetch(previous=%d, fetched=%d) = %v, want %v",
c.previous, c.fetched, got, c.want)
}
})
}
}
+33 -3
View File
@@ -2,6 +2,7 @@ package job
import (
"strings"
"sync/atomic"
"time"
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
@@ -12,11 +13,16 @@ import (
var DefaultTruthyValues = []string{"true", "1", "yes", "on"}
// Share of the previous successful fetch a new one must still return to be
// trusted: a sudden collapse means a broken directory far more often than churn.
const ldapAutoDeleteMinRetainPercent = 50
type LdapSyncJob struct {
settingService service.SettingService
inboundService service.InboundService
clientService service.ClientService
xrayService service.XrayService
lastFlagCount atomic.Int64
}
// --- Helper functions for mustGet ---
@@ -77,7 +83,7 @@ func (j *LdapSyncJob) Run() {
UserFilter: mustGetString(j.settingService.GetLdapUserFilter),
UserAttr: mustGetString(j.settingService.GetLdapUserAttr),
FlagField: mustGetStringOr(j.settingService.GetLdapFlagField, mustGetString(j.settingService.GetLdapVlessField)),
TruthyVals: splitCsv(mustGetString(j.settingService.GetLdapTruthyValues)),
TruthyVals: truthyValuesOrDefault(mustGetString(j.settingService.GetLdapTruthyValues)),
Invert: mustGetBool(j.settingService.GetLdapInvertFlag),
}
@@ -157,7 +163,7 @@ func (j *LdapSyncJob) Run() {
// --- Auto delete clients not in LDAP ---
autoDelete := mustGetBool(j.settingService.GetLdapAutoDelete)
if autoDelete {
if autoDelete && j.autoDeleteSafeForFetch(len(flags)) {
ldapEmailSet := map[string]struct{}{}
for e := range flags {
ldapEmailSet[e] = struct{}{}
@@ -166,11 +172,35 @@ func (j *LdapSyncJob) Run() {
j.deleteClientsNotInLDAP(tag, ldapEmailSet)
}
}
j.lastFlagCount.Store(int64(len(flags)))
}
// FetchVlessFlags returns (empty, nil) when the bind succeeds but the search
// yields nothing — a renamed OU, a lost read grant — which is not "all gone".
func (j *LdapSyncJob) autoDeleteSafeForFetch(fetched int) bool {
if fetched == 0 {
logger.Warning("LDAP auto-delete skipped: directory returned no usable users")
return false
}
previous := j.lastFlagCount.Load()
if previous > 0 && int64(fetched)*100 < previous*ldapAutoDeleteMinRetainPercent {
logger.Warningf("LDAP auto-delete skipped: fetched %d users, previous successful sync saw %d (below %d%% retention)",
fetched, previous, ldapAutoDeleteMinRetainPercent)
return false
}
return true
}
func truthyValuesOrDefault(s string) []string {
if vals := splitCsv(s); len(vals) > 0 {
return vals
}
return DefaultTruthyValues
}
func splitCsv(s string) []string {
if s == "" {
return DefaultTruthyValues
return nil
}
parts := strings.Split(s, ",")
out := make([]string, 0, len(parts))
+1 -1
View File
@@ -63,7 +63,7 @@ func (j *NodeHeartbeatJob) Run() {
if !websocket.HasClients() {
return
}
updated, err := j.nodeService.GetNodeTree()
updated, err := j.nodeService.GetNodeTreeView()
if err != nil {
logger.Warning("node heartbeat: load nodes for broadcast failed:", err)
return
+24 -3
View File
@@ -1,6 +1,8 @@
package job
import (
"time"
"github.com/mhsanaei/3x-ui/v3/internal/logger"
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
)
@@ -13,15 +15,25 @@ type PeriodicTrafficResetJob struct {
inboundService service.InboundService
clientService service.ClientService
period Period
location *time.Location
}
// NewPeriodicTrafficResetJob creates a new periodic traffic reset job for the specified period.
func NewPeriodicTrafficResetJob(period Period) *PeriodicTrafficResetJob {
func NewPeriodicTrafficResetJob(period Period, location *time.Location) *PeriodicTrafficResetJob {
return &PeriodicTrafficResetJob{
period: period,
period: period,
location: location,
}
}
func monthlyResetDue(resetDay int, now time.Time) bool {
if resetDay < 1 {
resetDay = 1
}
lastDay := time.Date(now.Year(), now.Month()+1, 0, 0, 0, 0, 0, now.Location()).Day()
return now.Day() == min(resetDay, lastDay)
}
// Run resets traffic statistics for all inbounds that match the configured reset period.
func (j *PeriodicTrafficResetJob) Run() {
inbounds, err := j.inboundService.GetInboundsByTrafficReset(string(j.period))
@@ -30,13 +42,22 @@ func (j *PeriodicTrafficResetJob) Run() {
return
}
if j.period == "monthly" {
now := time.Now().In(j.location)
due := inbounds[:0]
for _, inbound := range inbounds {
if monthlyResetDue(inbound.TrafficResetDay, now) {
due = append(due, inbound)
}
}
inbounds = due
}
if len(inbounds) == 0 {
return
}
logger.Infof("Running periodic traffic reset job for period: %s (%d matching inbounds)", j.period, len(inbounds))
resetCount := 0
for _, inbound := range inbounds {
resetInboundErr := j.inboundService.ResetInboundTraffic(inbound.Id)
if resetInboundErr != nil {
@@ -0,0 +1,31 @@
package job
import (
"testing"
"time"
)
func TestMonthlyResetDue(t *testing.T) {
cases := []struct {
name string
resetDay int
now time.Time
want bool
}{
{"legacy default on first", 0, time.Date(2026, time.July, 1, 0, 0, 0, 0, time.UTC), true},
{"configured day", 15, time.Date(2026, time.July, 15, 0, 0, 0, 0, time.UTC), true},
{"before configured day", 15, time.Date(2026, time.July, 14, 0, 0, 0, 0, time.UTC), false},
{"month end", 31, time.Date(2026, time.January, 31, 0, 0, 0, 0, time.UTC), true},
{"short month fallback", 31, time.Date(2026, time.February, 28, 0, 0, 0, 0, time.UTC), true},
{"leap year fallback", 31, time.Date(2028, time.February, 29, 0, 0, 0, 0, time.UTC), true},
{"not before short month end", 31, time.Date(2028, time.February, 28, 0, 0, 0, 0, time.UTC), false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := monthlyResetDue(tc.resetDay, tc.now); got != tc.want {
t.Fatalf("monthlyResetDue(%d, %s) = %v, want %v", tc.resetDay, tc.now.Format(time.DateOnly), got, tc.want)
}
})
}
}
+27
View File
@@ -0,0 +1,27 @@
package job
import (
"github.com/mhsanaei/3x-ui/v3/internal/logger"
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
)
// ReapSyncOrphansJob hard-deletes the clients the node-snapshot sweep only
// soft-orphaned, once enough clean merges have agreed they are really gone.
type ReapSyncOrphansJob struct {
clientService service.ClientService
}
func NewReapSyncOrphansJob() *ReapSyncOrphansJob {
return new(ReapSyncOrphansJob)
}
func (j *ReapSyncOrphansJob) Run() {
reaped, err := j.clientService.ReapSyncOrphans()
if err != nil {
logger.Warning("reap sync orphans failed:", err)
return
}
if reaped > 0 {
logger.Infof("reap sync orphans: removed %d client(s)", reaped)
}
}
@@ -0,0 +1,91 @@
package job
import (
"errors"
"io"
"net"
"testing"
"time"
"github.com/valyala/fasthttp"
)
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)
}
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)
}
}
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")
}
}
+42 -10
View File
@@ -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,40 @@ type XrayTrafficJob struct {
// refetch for the rest.
const clientStatsSnapshotMaxClients = 5000
// splitMovedClientTraffics keeps the rows that actually moved bytes this poll,
// alongside the active-email list and set derived from the same pass.
//
// Xray reports a row for every known email whether or not it transferred
// anything, so on a large panel nearly every delta is zero. The database writes
// and the external-API inform consume the full slice before this point; the
// WebSocket frame only feeds the dashboard's live speed column, where an absent
// row and a zero row render identically. Broadcasting just the movers keeps that
// frame from growing with the client count — at 5k clients it was carrying about
// a megabyte of zeros every five seconds.
func splitMovedClientTraffics(clientTraffics []*xray.ClientTraffic) ([]*xray.ClientTraffic, []string, map[string]bool) {
moved := make([]*xray.ClientTraffic, 0, len(clientTraffics))
emails := make([]string, 0, len(clientTraffics))
active := make(map[string]bool, len(clientTraffics))
for _, ct := range clientTraffics {
if ct == nil || ct.Up+ct.Down <= 0 {
continue
}
moved = append(moved, ct)
emails = append(emails, ct.Email)
active[ct.Email] = true
}
return moved, emails, active
}
const externalInformTimeout = 3 * time.Second
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)
@@ -78,14 +113,7 @@ func (j *XrayTrafficJob) Run() {
// than the shared last_online column, which remote-node syncs also bump
// and would otherwise make a client active only on a remote node appear
// online on local inbounds.
activeEmails := make([]string, 0, len(clientTraffics))
deltaActive := make(map[string]bool, len(clientTraffics))
for _, ct := range clientTraffics {
if ct != nil && ct.Up+ct.Down > 0 {
activeEmails = append(activeEmails, ct.Email)
deltaActive[ct.Email] = true
}
}
movedTraffics, activeEmails, deltaActive := splitMovedClientTraffics(clientTraffics)
// When the core supports the online-stats API, union in connection-based
// onlines. Neither signal alone covers everything: an idle-but-connected
// client moves no bytes between polls (the delta heuristic's blind spot),
@@ -169,7 +197,7 @@ func (j *XrayTrafficJob) Run() {
}
websocket.BroadcastTraffic(map[string]any{
"traffics": traffics,
"clientTraffics": clientTraffics,
"clientTraffics": movedTraffics,
"onlineClients": onlineClients,
"onlineByGuid": j.inboundService.GetOnlineClientsByGuid(),
"activeInbounds": j.inboundService.GetActiveInboundsByGuid(),
@@ -197,6 +225,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 +249,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)
}
}
@@ -0,0 +1,59 @@
package job
import (
"slices"
"testing"
"github.com/mhsanaei/3x-ui/v3/internal/xray"
)
func TestSplitMovedClientTraffics(t *testing.T) {
rows := []*xray.ClientTraffic{
{Email: "idle@x", Up: 0, Down: 0},
{Email: "up@x", Up: 1024, Down: 0},
{Email: "down@x", Up: 0, Down: 2048},
nil,
{Email: "both@x", Up: 512, Down: 512},
{Email: "alsoidle@x", Up: 0, Down: 0},
}
moved, emails, active := splitMovedClientTraffics(rows)
t.Run("only the rows that moved bytes are broadcast", func(t *testing.T) {
got := make([]string, 0, len(moved))
for _, ct := range moved {
got = append(got, ct.Email)
}
want := []string{"up@x", "down@x", "both@x"}
if !slices.Equal(got, want) {
t.Fatalf("moved = %v, want %v", got, want)
}
})
t.Run("the active list and set agree with the broadcast rows", func(t *testing.T) {
want := []string{"up@x", "down@x", "both@x"}
if !slices.Equal(emails, want) {
t.Fatalf("activeEmails = %v, want %v", emails, want)
}
if len(active) != len(want) {
t.Fatalf("deltaActive has %d entries, want %d", len(active), len(want))
}
for _, e := range want {
if !active[e] {
t.Fatalf("deltaActive missing %q", e)
}
}
if active["idle@x"] {
t.Fatal("an idle client must not count as active")
}
})
t.Run("an all-idle poll broadcasts nothing", func(t *testing.T) {
moved, emails, active := splitMovedClientTraffics([]*xray.ClientTraffic{
{Email: "a@x"}, {Email: "b@x"},
})
if len(moved) != 0 || len(emails) != 0 || len(active) != 0 {
t.Fatalf("expected an empty split, got %d/%d/%d", len(moved), len(emails), len(active))
}
})
}