mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 11:15:03 +03:00
MHSanaei
0972ac31eb
Create, BulkCreate, Attach and BulkAttach built one marshalled copy of the client per entry of the request's inboundIds and held all of them until the fan-out returned, with repeated ids never collapsed. An id costs two bytes of request body, so a large client named on the same inbound a few thousand times needed gigabytes and the panel died. Every fan-out entry point now collapses the id list first, keeping the order of first occurrence so credentials are still minted on the first inbound.
394 lines
14 KiB
Go
394 lines
14 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
goruntime "runtime"
|
|
"slices"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/web/runtime"
|
|
)
|
|
|
|
func TestCreateAcrossManyInboundsUsesOneEmailSnapshot(t *testing.T) {
|
|
setupBulkDB(t)
|
|
svc := &ClientService{}
|
|
inboundSvc := &InboundService{}
|
|
|
|
const uuid = "bbbbbbbb-1111-2222-3333-555555555555"
|
|
ids := make([]int, 0, 6)
|
|
for i := range 6 {
|
|
ib := mkInbound(t, 23001+i, model.VLESS, `{"clients":[]}`)
|
|
ids = append(ids, ib.Id)
|
|
}
|
|
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "fan@x", ID: uuid, SubID: "sub-fan", Enable: true},
|
|
InboundIds: ids,
|
|
}); err != nil {
|
|
t.Fatalf("Create across %d inbounds: %v", len(ids), err)
|
|
}
|
|
|
|
if n := countClientRecords(t); n != 1 {
|
|
t.Fatalf("client records = %d, want 1", n)
|
|
}
|
|
rec := lookupClientRecord(t, "fan@x")
|
|
if rec.UUID != uuid || rec.SubID != "sub-fan" {
|
|
t.Fatalf("record = {uuid:%q sub:%q}, want {%q sub-fan}", rec.UUID, rec.SubID, uuid)
|
|
}
|
|
for _, id := range ids {
|
|
if !settingsHoldUUID(t, inboundSvc, id, uuid) {
|
|
t.Fatalf("inbound %d settings missing the client", id)
|
|
}
|
|
}
|
|
|
|
linked, err := svc.GetInboundIdsForRecord(rec.Id)
|
|
if err != nil {
|
|
t.Fatalf("GetInboundIdsForRecord: %v", err)
|
|
}
|
|
if len(linked) != len(ids) {
|
|
t.Fatalf("linked inbounds = %d, want %d", len(linked), len(ids))
|
|
}
|
|
}
|
|
|
|
func TestAttachAcrossManyInboundsUsesOneEmailSnapshot(t *testing.T) {
|
|
setupBulkDB(t)
|
|
svc := &ClientService{}
|
|
inboundSvc := &InboundService{}
|
|
|
|
first := mkInbound(t, 23101, model.VLESS, `{"clients":[]}`)
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "att@x", ID: "cccccccc-1111-2222-3333-666666666666", SubID: "sub-att", Enable: true},
|
|
InboundIds: []int{first.Id},
|
|
}); err != nil {
|
|
t.Fatalf("seed Create: %v", err)
|
|
}
|
|
rec := lookupClientRecord(t, "att@x")
|
|
|
|
ids := []int{first.Id}
|
|
for i := range 4 {
|
|
ib := mkInbound(t, 23102+i, model.VLESS, `{"clients":[]}`)
|
|
ids = append(ids, ib.Id)
|
|
}
|
|
|
|
if _, err := svc.Attach(inboundSvc, rec.Id, ids); err != nil {
|
|
t.Fatalf("Attach across %d inbounds: %v", len(ids), err)
|
|
}
|
|
|
|
if n := countClientRecords(t); n != 1 {
|
|
t.Fatalf("client records after attach = %d, want 1", n)
|
|
}
|
|
linked, err := svc.GetInboundIdsForRecord(rec.Id)
|
|
if err != nil {
|
|
t.Fatalf("GetInboundIdsForRecord: %v", err)
|
|
}
|
|
if len(linked) != len(ids) {
|
|
t.Fatalf("linked inbounds = %d, want %d", len(linked), len(ids))
|
|
}
|
|
}
|
|
|
|
// barrierNodeRuntime holds every AddClient until fanout of them are inside it at
|
|
// once, recording the peak overlap; a sequential caller only ever reaches one.
|
|
type barrierNodeRuntime struct {
|
|
fakeNodeRuntime
|
|
fanout int32
|
|
inFlight atomic.Int32
|
|
maxPar atomic.Int32
|
|
release chan struct{}
|
|
freed atomic.Bool
|
|
expired atomic.Bool
|
|
}
|
|
|
|
func (b *barrierNodeRuntime) free() {
|
|
if b.freed.CompareAndSwap(false, true) {
|
|
close(b.release)
|
|
}
|
|
}
|
|
|
|
func (b *barrierNodeRuntime) AddClient(ctx context.Context, ib *model.Inbound, c model.Client) error {
|
|
n := b.inFlight.Add(1)
|
|
for {
|
|
peak := b.maxPar.Load()
|
|
if n <= peak || b.maxPar.CompareAndSwap(peak, n) {
|
|
break
|
|
}
|
|
}
|
|
if n == b.fanout {
|
|
b.free()
|
|
}
|
|
select {
|
|
case <-b.release:
|
|
case <-time.After(5 * time.Second):
|
|
// Release everyone on the first timeout so a sequential regression
|
|
// fails once instead of stalling for fanout x the wait.
|
|
b.expired.Store(true)
|
|
b.free()
|
|
}
|
|
b.inFlight.Add(-1)
|
|
return b.fakeNodeRuntime.AddClient(ctx, ib, c)
|
|
}
|
|
|
|
func fanoutNodeInbounds(t *testing.T, mgr *runtime.Manager, rt runtime.Runtime, n int, basePort int) []int {
|
|
t.Helper()
|
|
ids := make([]int, 0, n)
|
|
for i := range n {
|
|
node := &model.Node{
|
|
Name: fmt.Sprintf("%s-%d", t.Name(), i), Address: "127.0.0.1", Port: 2096 + i,
|
|
ApiToken: "tok", Enable: true, Status: "online",
|
|
}
|
|
if err := database.GetDB().Create(node).Error; err != nil {
|
|
t.Fatalf("create node %d: %v", i, err)
|
|
}
|
|
mgr.SetRuntimeOverride(node.Id, rt)
|
|
ids = append(ids, nodeInbound(t, node.Id, basePort+i, nil).Id)
|
|
}
|
|
return ids
|
|
}
|
|
|
|
// TestCreateAcrossNodesPushesConcurrently pins that a client spanning several
|
|
// node inbounds pushes to them at once, up to inboundFanoutConcurrency at a time.
|
|
func TestCreateAcrossNodesPushesConcurrently(t *testing.T) {
|
|
setupBulkDB(t)
|
|
startSerializedWriter(t)
|
|
mgr := useTestRuntimeManager(t)
|
|
|
|
const nodes = inboundFanoutConcurrency + 1
|
|
bar := &barrierNodeRuntime{fanout: inboundFanoutConcurrency, release: make(chan struct{})}
|
|
ids := fanoutNodeInbounds(t, mgr, bar, nodes, 40101)
|
|
|
|
if _, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{
|
|
Client: model.Client{Email: "fanout@x", ID: "11111111-2222-3333-4444-555555555555", SubID: "sub-fanout", Enable: true},
|
|
InboundIds: ids,
|
|
}); err != nil {
|
|
t.Fatalf("Create across %d node inbounds: %v", nodes, err)
|
|
}
|
|
|
|
if got := bar.addClient.Load(); got != nodes {
|
|
t.Fatalf("AddClient pushes = %d, want %d", got, nodes)
|
|
}
|
|
if got := bar.maxPar.Load(); got < 2 || got != inboundFanoutConcurrency {
|
|
t.Fatalf("peak node pushes in flight = %d, want overlap at the %d cap (barrier timed out: %v)",
|
|
got, inboundFanoutConcurrency, bar.expired.Load())
|
|
}
|
|
}
|
|
|
|
// TestCreateRecoversPanicInOneInbound pins that a panicking inbound fails only
|
|
// itself: off the request goroutine nothing else would catch it.
|
|
func TestCreateRecoversPanicInOneInbound(t *testing.T) {
|
|
setupBulkDB(t)
|
|
startSerializedWriter(t)
|
|
mgr := useTestRuntimeManager(t)
|
|
|
|
node := &model.Node{
|
|
Name: t.Name(), 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, &panicNodeRuntime{})
|
|
boom := nodeInbound(t, node.Id, 40201, nil)
|
|
healthy := mkInbound(t, 40202, model.VLESS, `{"clients":[]}`)
|
|
|
|
const uuid = "33333333-4444-5555-6666-777777777777"
|
|
_, err := (&ClientService{}).Create(&InboundService{}, &ClientCreatePayload{
|
|
Client: model.Client{Email: "panic@x", ID: uuid, SubID: "sub-panic", Enable: true},
|
|
InboundIds: []int{boom.Id, healthy.Id},
|
|
})
|
|
if err == nil {
|
|
t.Fatal("a panicking node runtime produced no error")
|
|
}
|
|
if want := fmt.Sprintf("inbound %d: panic:", boom.Id); !strings.Contains(err.Error(), want) {
|
|
t.Fatalf("error %q does not report %q", err, want)
|
|
}
|
|
if !settingsHoldUUID(t, &InboundService{}, healthy.Id, uuid) {
|
|
t.Fatalf("healthy inbound %d did not get the client", healthy.Id)
|
|
}
|
|
}
|
|
|
|
// TestCreateLeavesHwidLimitAloneWhenCreateFails pins that a create the panel
|
|
// reported as failed never rewrites a device cap, so it can never retrim one.
|
|
func TestCreateLeavesHwidLimitAloneWhenCreateFails(t *testing.T) {
|
|
setupBulkDB(t)
|
|
startSerializedWriter(t)
|
|
svc := &ClientService{}
|
|
inboundSvc := &InboundService{}
|
|
|
|
const vipUUID = "44444444-5555-6666-7777-888888888888"
|
|
seed := mkInbound(t, 41401, model.VLESS, `{"clients":[]}`)
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "vip@x", ID: vipUUID, SubID: "sub-vip", Enable: true},
|
|
InboundIds: []int{seed.Id},
|
|
LimitHwid: 3,
|
|
}); err != nil {
|
|
t.Fatalf("seed Create: %v", err)
|
|
}
|
|
|
|
broken := mkInbound(t, 41402, model.VLESS, `{"clients":`)
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "vip@x", ID: vipUUID, SubID: "sub-vip", Enable: true},
|
|
InboundIds: []int{broken.Id},
|
|
LimitHwid: 1,
|
|
}); err == nil {
|
|
t.Fatal("re-adding to an unparsable inbound returned no error")
|
|
}
|
|
|
|
if rec := lookupClientRecord(t, "vip@x"); rec.LimitHwid != 3 {
|
|
t.Fatalf("limit_hwid = %d, want the untouched 3: a failed create retrimmed a live client", rec.LimitHwid)
|
|
}
|
|
|
|
// Same failure with the seeded inbound alongside it: that one is a dedup
|
|
// no-op returning no error, which must not read as "an inbound took it".
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "vip@x", ID: vipUUID, SubID: "sub-vip", Enable: true},
|
|
InboundIds: []int{seed.Id, broken.Id},
|
|
LimitHwid: 1,
|
|
}); err == nil {
|
|
t.Fatal("re-adding over a no-op and an unparsable inbound returned no error")
|
|
}
|
|
if rec := lookupClientRecord(t, "vip@x"); rec.LimitHwid != 3 {
|
|
t.Fatalf("limit_hwid = %d, want the untouched 3: a no-op inbound counted as applied", rec.LimitHwid)
|
|
}
|
|
|
|
// A brand new identity that only partly applies is left uncapped rather than
|
|
// capped, the deliberate safe side: the operator saw the error and retries.
|
|
healthy := mkInbound(t, 41403, model.VLESS, `{"clients":[]}`)
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "fresh@x", ID: "55555555-6666-7777-8888-999999999999", SubID: "sub-fresh", Enable: true},
|
|
InboundIds: []int{healthy.Id, broken.Id},
|
|
LimitHwid: 5,
|
|
}); err == nil {
|
|
t.Fatal("creating over an unparsable inbound returned no error")
|
|
}
|
|
if rec := lookupClientRecord(t, "fresh@x"); rec.LimitHwid != 0 {
|
|
t.Fatalf("limit_hwid = %d, want 0 on a create that failed", rec.LimitHwid)
|
|
}
|
|
}
|
|
|
|
func assertNamesFailedInbounds(t *testing.T, err error, broken []*model.Inbound, healthy *model.Inbound) {
|
|
t.Helper()
|
|
if err == nil {
|
|
t.Fatalf("applying %d unparsable inbounds returned no error", len(broken))
|
|
}
|
|
for _, ib := range broken {
|
|
if want := fmt.Sprintf("inbound %d:", ib.Id); !strings.Contains(err.Error(), want) {
|
|
t.Fatalf("error %q does not name the failing %s", err, want)
|
|
}
|
|
}
|
|
if blamed := fmt.Sprintf("inbound %d:", healthy.Id); strings.Contains(err.Error(), blamed) {
|
|
t.Fatalf("error %q blames the healthy %s", err, blamed)
|
|
}
|
|
}
|
|
|
|
// TestFanoutReportsEveryFailingInbound pins that no inbound aborts the others:
|
|
// each failure names its own inbound, and the healthy ones still get the client.
|
|
func TestFanoutReportsEveryFailingInbound(t *testing.T) {
|
|
const halfBadUUID = "22222222-3333-4444-5555-666666666666"
|
|
|
|
t.Run("create", func(t *testing.T) {
|
|
setupBulkDB(t)
|
|
startSerializedWriter(t)
|
|
svc := &ClientService{}
|
|
inboundSvc := &InboundService{}
|
|
|
|
broken := []*model.Inbound{
|
|
mkInbound(t, 41201, model.VLESS, `{"clients":`),
|
|
mkInbound(t, 41202, model.VLESS, `{"clients":`),
|
|
}
|
|
healthy := mkInbound(t, 41203, model.VLESS, `{"clients":[]}`)
|
|
|
|
_, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "halfbad@x", ID: halfBadUUID, SubID: "sub-halfbad", Enable: true},
|
|
InboundIds: []int{broken[0].Id, broken[1].Id, healthy.Id},
|
|
})
|
|
assertNamesFailedInbounds(t, err, broken, healthy)
|
|
if !settingsHoldUUID(t, inboundSvc, healthy.Id, halfBadUUID) {
|
|
t.Fatalf("healthy inbound %d did not get the client", healthy.Id)
|
|
}
|
|
})
|
|
|
|
t.Run("attach", func(t *testing.T) {
|
|
setupBulkDB(t)
|
|
startSerializedWriter(t)
|
|
svc := &ClientService{}
|
|
inboundSvc := &InboundService{}
|
|
|
|
seed := mkInbound(t, 41301, model.VLESS, `{"clients":[]}`)
|
|
if _, err := svc.Create(inboundSvc, &ClientCreatePayload{
|
|
Client: model.Client{Email: "halfbad@x", ID: halfBadUUID, SubID: "sub-halfbad", Enable: true},
|
|
InboundIds: []int{seed.Id},
|
|
}); err != nil {
|
|
t.Fatalf("seed Create: %v", err)
|
|
}
|
|
|
|
broken := []*model.Inbound{
|
|
mkInbound(t, 41302, model.VLESS, `{"clients":`),
|
|
mkInbound(t, 41303, model.VLESS, `{"clients":`),
|
|
}
|
|
healthy := mkInbound(t, 41304, model.VLESS, `{"clients":[]}`)
|
|
|
|
rec := lookupClientRecord(t, "halfbad@x")
|
|
_, err := svc.Attach(inboundSvc, rec.Id, []int{broken[0].Id, broken[1].Id, healthy.Id})
|
|
assertNamesFailedInbounds(t, err, broken, healthy)
|
|
if !settingsHoldUUID(t, inboundSvc, healthy.Id, halfBadUUID) {
|
|
t.Fatalf("healthy inbound %d did not get the client", healthy.Id)
|
|
}
|
|
})
|
|
}
|
|
|
|
// An inbound id costs two bytes of request body, so a repeated one must not cost
|
|
// the panel another copy of the client: 2000 ids x 8 MB of comment was 16 GB.
|
|
func TestRepeatedInboundIdCostsOneCopyOfTheClient(t *testing.T) {
|
|
const commentBytes = 1 << 20
|
|
client := func(email string) model.Client {
|
|
return model.Client{Email: email, Comment: strings.Repeat("x", commentBytes), Enable: true}
|
|
}
|
|
|
|
cases := map[string]func(svc *ClientService, inboundSvc *InboundService, spare int, ids []int) error{
|
|
"Create": func(svc *ClientService, inboundSvc *InboundService, _ int, ids []int) error {
|
|
_, err := svc.Create(inboundSvc, &ClientCreatePayload{Client: client("repeated@x"), InboundIds: ids})
|
|
return err
|
|
},
|
|
"BulkCreate": func(svc *ClientService, inboundSvc *InboundService, _ int, ids []int) error {
|
|
res, _, err := svc.BulkCreate(inboundSvc, []ClientCreatePayload{{Client: client("repeated@x"), InboundIds: ids}})
|
|
if err == nil && res.Created != 1 {
|
|
t.Fatalf("bulk create: created %d, skipped %+v", res.Created, res.Skipped)
|
|
}
|
|
return err
|
|
},
|
|
"Attach": func(svc *ClientService, inboundSvc *InboundService, spare int, ids []int) error {
|
|
if _, err := svc.CreateOne(inboundSvc, spare, client("repeated@x")); err != nil {
|
|
return err
|
|
}
|
|
_, err := svc.AttachByEmail(inboundSvc, "repeated@x", ids)
|
|
return err
|
|
},
|
|
}
|
|
for name, run := range cases {
|
|
t.Run(name, func(t *testing.T) {
|
|
setupBulkDB(t)
|
|
ib := mkInbound(t, 21011, model.VLESS, `{"clients":[]}`)
|
|
spare := mkInbound(t, 21012, model.VLESS, `{"clients":[]}`)
|
|
|
|
var before, after goruntime.MemStats
|
|
goruntime.ReadMemStats(&before)
|
|
if err := run(&ClientService{}, &InboundService{}, spare.Id, slices.Repeat([]int{ib.Id}, 300)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
goruntime.ReadMemStats(&after)
|
|
|
|
if allocated := after.TotalAlloc - before.TotalAlloc; allocated > 100*commentBytes {
|
|
t.Fatalf("allocated %d MiB for a 1 MiB client named on one inbound", allocated>>20)
|
|
}
|
|
if emails := settingsClientEmails(t, ib.Id); len(emails) != 1 {
|
|
t.Fatalf("clients on the inbound = %v, want the one", emails)
|
|
}
|
|
})
|
|
}
|
|
}
|