mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-07-08 21:56:08 +00:00
fix(inbounds): apply runtime changes after the DB commit (#5768)
* fix(inbounds): apply runtime changes after commit * ci: fix staticcheck findings
This commit is contained in:
@@ -37,6 +37,39 @@ jobs:
|
||||
go list ./... | grep -v '/frontend/node_modules/' > /tmp/go-packages.txt
|
||||
go test -shuffle=on -count=1 $(cat /tmp/go-packages.txt)
|
||||
|
||||
postgres-durable-first:
|
||||
runs-on: ubuntu-latest
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:16
|
||||
env:
|
||||
POSTGRES_USER: postgres
|
||||
POSTGRES_PASSWORD: postgres
|
||||
POSTGRES_DB: xui_durable
|
||||
ports:
|
||||
- 5432:5432
|
||||
options: >-
|
||||
--health-cmd "pg_isready -U postgres -d xui_durable"
|
||||
--health-interval 10s
|
||||
--health-timeout 5s
|
||||
--health-retries 5
|
||||
steps:
|
||||
- uses: actions/checkout@v7
|
||||
- uses: actions/setup-go@v6
|
||||
with:
|
||||
go-version-file: go.mod
|
||||
cache: true
|
||||
- name: Stub internal/web/dist for go:embed
|
||||
run: mkdir -p internal/web/dist && touch internal/web/dist/.gitkeep
|
||||
- name: PostgreSQL durable-first tests
|
||||
run: |
|
||||
set -o pipefail
|
||||
XUI_DB_TYPE=postgres XUI_DB_DSN="host=127.0.0.1 port=5432 user=postgres password=postgres dbname=xui_durable sslmode=disable" \
|
||||
go test ./internal/web/service -run 'PostgresCommitFailure' -count=1 -v | tee /tmp/postgres-durable-first.log
|
||||
if grep -q -- '--- SKIP' /tmp/postgres-durable-first.log; then
|
||||
exit 1
|
||||
fi
|
||||
|
||||
codegen:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
|
||||
@@ -52,11 +52,10 @@ func InitLogger(level logging.Level) {
|
||||
backends := make([]logging.Backend, 0, 2)
|
||||
|
||||
// Console/syslog backend with configurable level
|
||||
if consoleBackend := initDefaultBackend(); consoleBackend != nil {
|
||||
leveledBackend := logging.AddModuleLevel(consoleBackend)
|
||||
leveledBackend.SetLevel(level, "x-ui")
|
||||
backends = append(backends, leveledBackend)
|
||||
}
|
||||
consoleBackend := initDefaultBackend()
|
||||
leveledBackend := logging.AddModuleLevel(consoleBackend)
|
||||
leveledBackend.SetLevel(level, "x-ui")
|
||||
backends = append(backends, leveledBackend)
|
||||
|
||||
// File backend with DEBUG level for comprehensive logging
|
||||
if fileBackend := initFileBackend(); fileBackend != nil {
|
||||
|
||||
@@ -139,6 +139,7 @@ func TestGetClientTrafficByEmailReadsClientsTable(t *testing.T) {
|
||||
}
|
||||
if tr == nil {
|
||||
t.Fatalf("expected traffic, got nil")
|
||||
return
|
||||
}
|
||||
if tr.UUID != "11111111-1111-1111-1111-111111111111" {
|
||||
t.Fatalf("UUID not enriched from clients table, got %q", tr.UUID)
|
||||
|
||||
+144
-178
@@ -790,114 +790,85 @@ func (s *InboundService) AddInbound(inbound *model.Inbound) (*model.Inbound, boo
|
||||
}
|
||||
|
||||
db := database.GetDB()
|
||||
tx := db.Begin()
|
||||
markDirty := false
|
||||
defer func() {
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
return
|
||||
}
|
||||
if markDirty && inbound.NodeID != nil {
|
||||
if dErr := (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID); dErr != nil {
|
||||
err = dErr
|
||||
tx.Rollback()
|
||||
return
|
||||
}
|
||||
}
|
||||
tx.Commit()
|
||||
}()
|
||||
|
||||
// Omit the ClientStats has-many association: GORM's cascade would INSERT
|
||||
// those rows with an ON CONFLICT target on the primary key only, which
|
||||
// collides with the globally-unique client_traffics.email when an imported
|
||||
// inbound carries clients that another inbound already created (e.g.
|
||||
// importing two inbounds that share the same clients). We insert the stats
|
||||
// ourselves below with the same email-conflict guard AddClientStat uses.
|
||||
err = tx.Omit("ClientStats").Save(inbound).Error
|
||||
if err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
// Imported stats first, so their traffic counters survive; emails that
|
||||
// already own a (shared) row are skipped instead of tripping the unique
|
||||
// constraint.
|
||||
for i := range inbound.ClientStats {
|
||||
if inbound.ClientStats[i].Email == "" {
|
||||
continue
|
||||
}
|
||||
inbound.ClientStats[i].Id = 0
|
||||
inbound.ClientStats[i].InboundId = inbound.Id
|
||||
if err = tx.Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "email"}},
|
||||
DoNothing: true,
|
||||
}).Create(&inbound.ClientStats[i]).Error; err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
}
|
||||
// Then make sure every client has a stats row. AddClientStat is a no-op
|
||||
// where one exists (including the rows just inserted), and fills the gap
|
||||
// for clients an import payload didn't carry stats for.
|
||||
for _, client := range clients {
|
||||
if err = s.AddClientStat(tx, inbound.Id, &client); err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
}
|
||||
|
||||
if err = s.clientService.SyncInbound(tx, inbound.Id, clients); err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
|
||||
// Legacy import: an inbound exported from a build that predated the hosts
|
||||
// table carries its external proxies inline in streamSettings.externalProxy.
|
||||
// The startup migration that converts those to host rows runs once and is
|
||||
// gated off afterwards, so it never sees a freshly imported inbound —
|
||||
// reproduce it here. No-op for inbounds without externalProxy (everything the
|
||||
// current UI builds), so this only fires on such imports.
|
||||
if _, err = database.CreateHostsFromExternalProxy(tx, inbound.Id, inbound.StreamSettings); err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
|
||||
// Before the deferred commit, so a node in "selected" sync mode cannot
|
||||
// sweep the new central row in the gap before its tag is allowed.
|
||||
if inbound.NodeID != nil {
|
||||
if aErr := (&NodeService{}).EnsureInboundTagAllowed(*inbound.NodeID, inbound.Tag); aErr != nil {
|
||||
logger.Warning("allow inbound tag on node failed:", aErr)
|
||||
}
|
||||
}
|
||||
|
||||
needRestart := false
|
||||
if inbound.Enable {
|
||||
rt, push, dirty, perr := s.nodePushPlan(inbound)
|
||||
if perr != nil {
|
||||
err = perr
|
||||
return inbound, false, err
|
||||
var postCommitApply func()
|
||||
err = db.Transaction(func(tx *gorm.DB) error {
|
||||
markDirty := false
|
||||
if err := tx.Omit("ClientStats").Save(inbound).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if dirty {
|
||||
markDirty = true
|
||||
}
|
||||
if push {
|
||||
payload := inbound
|
||||
pushable := true
|
||||
if inbound.NodeID == nil && inbound.Protocol == model.MTProto {
|
||||
if built, bErr := s.buildRuntimeInboundForAPI(tx, inbound); bErr == nil {
|
||||
payload = built
|
||||
} else {
|
||||
logger.Debug("Unable to prepare runtime inbound config:", bErr)
|
||||
pushable = false
|
||||
}
|
||||
for i := range inbound.ClientStats {
|
||||
if inbound.ClientStats[i].Email == "" {
|
||||
continue
|
||||
}
|
||||
if pushable {
|
||||
if err1 := rt.AddInbound(context.Background(), payload); err1 == nil {
|
||||
logger.Debug("New inbound added on", rt.Name(), ":", inbound.Tag)
|
||||
} else {
|
||||
logger.Debug("Unable to add inbound on", rt.Name(), ":", err1)
|
||||
if inbound.NodeID != nil {
|
||||
markDirty = true
|
||||
} else if inbound.Protocol != model.MTProto {
|
||||
needRestart = true
|
||||
inbound.ClientStats[i].Id = 0
|
||||
inbound.ClientStats[i].InboundId = inbound.Id
|
||||
if err := tx.Clauses(clause.OnConflict{
|
||||
Columns: []clause.Column{{Name: "email"}},
|
||||
DoNothing: true,
|
||||
}).Create(&inbound.ClientStats[i]).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for _, client := range clients {
|
||||
if err := s.AddClientStat(tx, inbound.Id, &client); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := s.clientService.SyncInbound(tx, inbound.Id, clients); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := database.CreateHostsFromExternalProxy(tx, inbound.Id, inbound.StreamSettings); err != nil {
|
||||
return err
|
||||
}
|
||||
if inbound.NodeID != nil {
|
||||
nodeID := *inbound.NodeID
|
||||
if err := (&NodeService{}).EnsureInboundTagAllowedTx(tx, nodeID, inbound.Tag); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if inbound.Enable {
|
||||
if inbound.NodeID != nil {
|
||||
markDirty = true
|
||||
} else {
|
||||
rt, push, _, perr := s.nodePushPlan(inbound)
|
||||
if perr != nil {
|
||||
return perr
|
||||
}
|
||||
if push {
|
||||
payload := inbound
|
||||
pushable := true
|
||||
if inbound.Protocol == model.MTProto {
|
||||
if built, bErr := s.buildRuntimeInboundForAPI(tx, inbound); bErr == nil {
|
||||
payload = built
|
||||
} else {
|
||||
logger.Debug("Unable to prepare runtime inbound config:", bErr)
|
||||
pushable = false
|
||||
}
|
||||
}
|
||||
if pushable {
|
||||
postCommitApply = func() {
|
||||
if err1 := rt.AddInbound(context.Background(), payload); err1 == nil {
|
||||
logger.Debug("New inbound added on", rt.Name(), ":", inbound.Tag)
|
||||
} else {
|
||||
logger.Debug("Unable to add inbound on", rt.Name(), ":", err1)
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if markDirty && inbound.NodeID != nil {
|
||||
return (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return inbound, false, err
|
||||
}
|
||||
if postCommitApply != nil {
|
||||
postCommitApply()
|
||||
}
|
||||
|
||||
// A routed mtproto inbound is not an Xray inbound itself, so the runtime
|
||||
@@ -914,31 +885,41 @@ func (s *InboundService) DelInbound(id int) (bool, error) {
|
||||
db := database.GetDB()
|
||||
|
||||
needRestart := false
|
||||
markDirty := false
|
||||
var postCommitApply func()
|
||||
var ib model.Inbound
|
||||
loadErr := db.Model(model.Inbound{}).Where("id = ?", id).First(&ib).Error
|
||||
if loadErr == nil {
|
||||
shouldPushToRuntime := ib.NodeID != nil || ib.Enable
|
||||
if shouldPushToRuntime {
|
||||
rt, push, dirty, perr := s.nodePushPlan(&ib)
|
||||
if perr != nil {
|
||||
logger.Warning("DelInbound: node lookup failed, deleting central row anyway:", perr)
|
||||
markDirty = true
|
||||
} else if push {
|
||||
if err1 := rt.DelInbound(context.Background(), &ib); err1 == nil {
|
||||
logger.Debug("Inbound deleted on", rt.Name(), ":", ib.Tag)
|
||||
} else {
|
||||
logger.Warning("DelInbound on", rt.Name(), "failed, deleting central row anyway:", err1)
|
||||
if ib.NodeID == nil {
|
||||
needRestart = true
|
||||
} else {
|
||||
markDirty = true
|
||||
if ib.NodeID != nil {
|
||||
rt, push, _, perr := s.nodePushPlan(&ib)
|
||||
if perr != nil {
|
||||
logger.Warning("DelInbound: node runtime lookup failed, deleting central row anyway:", perr)
|
||||
} else if push {
|
||||
postCommitApply = func() {
|
||||
if err1 := rt.DelInbound(context.Background(), &ib); err1 == nil {
|
||||
logger.Debug("Inbound deleted on", rt.Name(), ":", ib.Tag)
|
||||
} else {
|
||||
logger.Warning("DelInbound on", rt.Name(), "failed after commit:", err1)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if ib.NodeID == nil {
|
||||
needRestart = true
|
||||
} else if dirty {
|
||||
markDirty = true
|
||||
} else {
|
||||
rt, push, _, perr := s.nodePushPlan(&ib)
|
||||
if perr != nil {
|
||||
logger.Warning("DelInbound: runtime lookup failed, deleting central row anyway:", perr)
|
||||
} else if push {
|
||||
postCommitApply = func() {
|
||||
if err1 := rt.DelInbound(context.Background(), &ib); err1 == nil {
|
||||
logger.Debug("Inbound deleted on", rt.Name(), ":", ib.Tag)
|
||||
} else {
|
||||
logger.Warning("DelInbound on", rt.Name(), "failed after commit:", err1)
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
} else {
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
} else {
|
||||
logger.Debug("DelInbound: skipping runtime push for disabled local inbound id:", id)
|
||||
@@ -947,12 +928,26 @@ func (s *InboundService) DelInbound(id int) (bool, error) {
|
||||
logger.Debug("DelInbound: inbound not found, id:", id)
|
||||
}
|
||||
|
||||
if err := s.clientService.DetachInbound(db, id); err != nil {
|
||||
return false, err
|
||||
if err := db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := s.clientService.DetachInbound(tx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Delete(model.Inbound{}, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Where("inbound_id = ?", id).Delete(&model.Host{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if loadErr == nil && ib.NodeID != nil {
|
||||
return (&NodeService{}).MarkNodeDirtyTx(tx, *ib.NodeID)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return needRestart, err
|
||||
}
|
||||
if postCommitApply != nil {
|
||||
postCommitApply()
|
||||
}
|
||||
|
||||
// Drop the deleted inbound's tag from any routing rules / loopback outbounds
|
||||
// in xrayTemplateConfig so they don't point at a tag that no longer exists.
|
||||
if loadErr == nil && ib.Tag != "" {
|
||||
if routingChanged, syncErr := (&XraySettingService{}).RemoveInboundTagReferences(ib.Tag); syncErr != nil {
|
||||
logger.Warning("DelInbound: sync routing on inbound delete failed:", syncErr)
|
||||
@@ -960,22 +955,6 @@ func (s *InboundService) DelInbound(id int) (bool, error) {
|
||||
needRestart = true
|
||||
}
|
||||
}
|
||||
|
||||
if err := db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Delete(model.Inbound{}, id).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
// Hosts have no hard FK; drop the inbound's hosts alongside it.
|
||||
if err := tx.Where("inbound_id = ?", id).Delete(&model.Host{}).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if markDirty && ib.NodeID != nil {
|
||||
return (&NodeService{}).MarkNodeDirtyTx(tx, *ib.NodeID)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return needRestart, err
|
||||
}
|
||||
if !database.IsPostgres() {
|
||||
var count int64
|
||||
if err := db.Model(&model.Inbound{}).Count(&count).Error; err != nil {
|
||||
@@ -1152,18 +1131,8 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
oldTagWasAuto := isAutoGeneratedTag(tag, oldInbound.Port, oldInbound.NodeID, oldBits)
|
||||
|
||||
needRestart := false
|
||||
var postCommitApply func()
|
||||
|
||||
// Persist the client-stat sync, settings munging, runtime push and inbound
|
||||
// save as one transaction routed through the serial traffic writer, so it
|
||||
// never runs concurrently with the @every 5s traffic poll. Both touch
|
||||
// client_traffics and inbounds in opposite order, which Postgres aborts as a
|
||||
// deadlock (40P01); serializing removes the contention (runSerializedTx).
|
||||
//
|
||||
// The runtime push stays inside the transaction here (unlike the client-edit
|
||||
// paths that apply it after commit): EnsureInboundTagAllowed must reach the
|
||||
// node before the central row is committed, or a "selected"-mode node would
|
||||
// sweep the renamed inbound on its next pull. Inbound edits are rare, so
|
||||
// holding the writer across the node call is an acceptable trade.
|
||||
txErr := runSerializedTx(func(tx *gorm.DB) error {
|
||||
if err := s.updateClientTraffics(tx, oldInbound, inbound); err != nil {
|
||||
return err
|
||||
@@ -1277,11 +1246,11 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
oldInbound.Tag = resolvedTag
|
||||
inbound.Tag = oldInbound.Tag
|
||||
|
||||
rt, push, _, perr := s.nodePushPlan(oldInbound)
|
||||
if perr != nil {
|
||||
return perr
|
||||
}
|
||||
if oldInbound.NodeID == nil {
|
||||
rt, push, _, perr := s.nodePushPlan(oldInbound)
|
||||
if perr != nil {
|
||||
return perr
|
||||
}
|
||||
if !push {
|
||||
needRestart = true
|
||||
} else if oldProtocol == model.MTProto || oldInbound.Protocol == model.MTProto {
|
||||
@@ -1311,15 +1280,23 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
} else {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
|
||||
logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
|
||||
}
|
||||
var runtimeInbound *model.Inbound
|
||||
if inbound.Enable {
|
||||
runtimeInbound, err2 := s.buildRuntimeInboundForAPI(tx, oldInbound)
|
||||
var err2 error
|
||||
runtimeInbound, err2 = s.buildRuntimeInboundForAPI(tx, oldInbound)
|
||||
if err2 != nil {
|
||||
logger.Debug("Unable to prepare runtime inbound config:", err2)
|
||||
needRestart = true
|
||||
} else if err2 := rt.AddInbound(context.Background(), runtimeInbound); err2 == nil {
|
||||
}
|
||||
}
|
||||
postCommitApply = func() {
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 == nil {
|
||||
logger.Debug("Old inbound deleted on", rt.Name(), ":", tag)
|
||||
}
|
||||
if runtimeInbound == nil {
|
||||
return
|
||||
}
|
||||
if err2 := rt.AddInbound(context.Background(), runtimeInbound); err2 == nil {
|
||||
logger.Debug("Updated inbound added on", rt.Name(), ":", oldInbound.Tag)
|
||||
} else {
|
||||
logger.Debug("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
@@ -1327,24 +1304,10 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if push {
|
||||
oldSnapshot := *oldInbound
|
||||
oldSnapshot.Tag = tag
|
||||
if !inbound.Enable {
|
||||
if err2 := rt.DelInbound(context.Background(), &oldSnapshot); err2 != nil {
|
||||
logger.Warning("Unable to disable inbound on", rt.Name(), ":", err2)
|
||||
}
|
||||
} else if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, oldInbound); err2 != nil {
|
||||
logger.Warning("Unable to update inbound on", rt.Name(), ":", err2)
|
||||
}
|
||||
}
|
||||
|
||||
// A rename must allow the new tag before the inbound row is committed, or a
|
||||
// node in "selected" sync mode would sweep the renamed central row on the
|
||||
// next pull.
|
||||
if oldInbound.NodeID != nil {
|
||||
if aErr := (&NodeService{}).EnsureInboundTagAllowed(*oldInbound.NodeID, oldInbound.Tag); aErr != nil {
|
||||
logger.Warning("allow inbound tag on node failed:", aErr)
|
||||
} else {
|
||||
nodeID := *oldInbound.NodeID
|
||||
if err := (&NodeService{}).EnsureInboundTagAllowedTx(tx, nodeID, oldInbound.Tag); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1374,6 +1337,9 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound,
|
||||
if txErr != nil {
|
||||
return inbound, false, txErr
|
||||
}
|
||||
if postCommitApply != nil {
|
||||
postCommitApply()
|
||||
}
|
||||
// After the rename is committed, point any routing rules / loopback outbounds
|
||||
// in xrayTemplateConfig at the new tag (oldInbound.Tag now holds the resolved
|
||||
// new tag; tag holds the pre-edit one). Done post-commit so a sync failure
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/op/go-logging"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||
xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger"
|
||||
)
|
||||
|
||||
func durablePostgresDB(t *testing.T) *gorm.DB {
|
||||
t.Helper()
|
||||
if os.Getenv("XUI_DB_TYPE") != "postgres" || strings.TrimSpace(os.Getenv("XUI_DB_DSN")) == "" {
|
||||
t.Skip("set XUI_DB_TYPE=postgres and XUI_DB_DSN to run commit-failure injection")
|
||||
}
|
||||
portConflictLoggerOnce.Do(func() { xuilogger.InitLogger(logging.ERROR) })
|
||||
if err := database.InitDB(""); err != nil {
|
||||
t.Fatalf("InitDB: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = database.CloseDB() })
|
||||
return database.GetDB()
|
||||
}
|
||||
|
||||
func durableTestInbound(nodeID *int, tag string, port int) *model.Inbound {
|
||||
return &model.Inbound{
|
||||
UserId: 1,
|
||||
NodeID: nodeID,
|
||||
Tag: tag,
|
||||
Remark: tag,
|
||||
Enable: true,
|
||||
Port: port,
|
||||
Protocol: model.VLESS,
|
||||
StreamSettings: `{"network":"tcp","security":"tls"}`,
|
||||
Settings: `{"clients":[],"decryption":"none"}`,
|
||||
}
|
||||
}
|
||||
|
||||
func installDeferredCommitFailure(t *testing.T, db *gorm.DB, callbackKind, callbackName, triggerTable, parentTable, childTable string) {
|
||||
t.Helper()
|
||||
_ = db.Exec("DROP TABLE IF EXISTS " + childTable).Error
|
||||
_ = db.Exec("DROP TABLE IF EXISTS " + parentTable).Error
|
||||
if err := db.Exec("CREATE TABLE " + parentTable + " (id bigint PRIMARY KEY)").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Exec("CREATE TABLE " + childTable + " (id bigint PRIMARY KEY, parent_id bigint REFERENCES " + parentTable + "(id) DEFERRABLE INITIALLY DEFERRED)").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = db.Exec("DROP TABLE IF EXISTS " + childTable).Error
|
||||
_ = db.Exec("DROP TABLE IF EXISTS " + parentTable).Error
|
||||
})
|
||||
|
||||
cb := func(tx *gorm.DB) {
|
||||
if tx.Statement == nil || tx.Statement.Table != triggerTable {
|
||||
return
|
||||
}
|
||||
res := tx.Session(&gorm.Session{NewDB: true}).Exec("INSERT INTO " + childTable + " (id, parent_id) VALUES (1, 999999)")
|
||||
if res.Error != nil {
|
||||
tx.AddError(res.Error)
|
||||
}
|
||||
}
|
||||
switch callbackKind {
|
||||
case "create":
|
||||
if err := db.Callback().Create().After("gorm:create").Register(callbackName, cb); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Callback().Create().Remove(callbackName) })
|
||||
case "update":
|
||||
if err := db.Callback().Update().After("gorm:update").Register(callbackName, cb); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Callback().Update().Remove(callbackName) })
|
||||
default:
|
||||
t.Fatalf("unknown callback kind %q", callbackKind)
|
||||
}
|
||||
}
|
||||
|
||||
func cleanupDurableInboundFixtures(t *testing.T, db *gorm.DB, tags ...string) {
|
||||
t.Helper()
|
||||
if len(tags) == 0 {
|
||||
return
|
||||
}
|
||||
cleanup := func() {
|
||||
_ = db.Exec("DELETE FROM hosts WHERE inbound_id IN (SELECT id FROM inbounds WHERE tag IN ?)", tags).Error
|
||||
_ = db.Exec("DELETE FROM client_inbounds WHERE inbound_id IN (SELECT id FROM inbounds WHERE tag IN ?)", tags).Error
|
||||
_ = db.Where("tag IN ?", tags).Delete(&model.Inbound{}).Error
|
||||
}
|
||||
cleanup()
|
||||
t.Cleanup(cleanup)
|
||||
}
|
||||
|
||||
func TestAddInbound_PostgresCommitFailureMakesNoRuntimeCall(t *testing.T) {
|
||||
db := durablePostgresDB(t)
|
||||
nodeID, fake := setupNodeRuntime(t)
|
||||
tag := "durable-add-" + strings.NewReplacer("/", "-", " ", "-").Replace(t.Name())
|
||||
cleanupDurableInboundFixtures(t, db, tag)
|
||||
installDeferredCommitFailure(t, db, "create", "durable:add_inbound_commit_failure", "inbounds", "durable_add_parent", "durable_add_child")
|
||||
|
||||
inbound := durableTestInbound(&nodeID, tag, 25443)
|
||||
_, _, err := (&InboundService{}).AddInbound(inbound)
|
||||
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "foreign key") {
|
||||
t.Fatalf("AddInbound error = %v, want deferred foreign-key commit failure", err)
|
||||
}
|
||||
if fake.addInbound.Load() != 0 || fake.updateInbound.Load() != 0 || fake.delInbound.Load() != 0 {
|
||||
t.Fatalf("runtime calls before failed commit: add=%d update=%d del=%d, want zero", fake.addInbound.Load(), fake.updateInbound.Load(), fake.delInbound.Load())
|
||||
}
|
||||
var rows int64
|
||||
if err := db.Model(&model.Inbound{}).Where("tag = ?", inbound.Tag).Count(&rows).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rows != 0 {
|
||||
t.Fatalf("rolled-back inbound rows = %d, want 0", rows)
|
||||
}
|
||||
var persisted model.Node
|
||||
if err := db.First(&persisted, nodeID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if persisted.ConfigDirty {
|
||||
t.Fatal("node dirty flag changed across failed commit")
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateInbound_PostgresCommitFailureMakesNoRuntimeCall(t *testing.T) {
|
||||
db := durablePostgresDB(t)
|
||||
nodeID, fake := setupNodeRuntime(t)
|
||||
tag := "durable-update-" + strings.NewReplacer("/", "-", " ", "-").Replace(t.Name())
|
||||
cleanupDurableInboundFixtures(t, db, tag)
|
||||
original := durableTestInbound(&nodeID, tag, 25444)
|
||||
original.Remark = "before"
|
||||
if err := db.Create(original).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
installDeferredCommitFailure(t, db, "update", "durable:update_inbound_commit_failure", "inbounds", "durable_update_parent", "durable_update_child")
|
||||
|
||||
update := durableTestInbound(&nodeID, original.Tag, original.Port)
|
||||
update.Id = original.Id
|
||||
update.Remark = "after"
|
||||
_, _, err := (&InboundService{}).UpdateInbound(update)
|
||||
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "foreign key") {
|
||||
t.Fatalf("UpdateInbound error = %v, want deferred foreign-key commit failure", err)
|
||||
}
|
||||
if fake.addInbound.Load() != 0 || fake.updateInbound.Load() != 0 || fake.delInbound.Load() != 0 {
|
||||
t.Fatalf("runtime calls before failed commit: add=%d update=%d del=%d, want zero", fake.addInbound.Load(), fake.updateInbound.Load(), fake.delInbound.Load())
|
||||
}
|
||||
var persisted model.Inbound
|
||||
if err := db.First(&persisted, original.Id).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if persisted.Remark != "before" {
|
||||
t.Fatalf("remark after failed commit = %q, want before", persisted.Remark)
|
||||
}
|
||||
var persistedNode model.Node
|
||||
if err := db.First(&persistedNode, nodeID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if persistedNode.ConfigDirty {
|
||||
t.Fatal("node dirty flag changed across failed commit")
|
||||
}
|
||||
}
|
||||
@@ -500,13 +500,19 @@ func (r staticEgressResolver) NodeEgressProxyURL(int) string { return string(r)
|
||||
// reporting the old tag until the remote update lands, and a leftover entry
|
||||
// that matches nothing is harmless.
|
||||
func (s *NodeService) EnsureInboundTagAllowed(nodeID int, tag string) error {
|
||||
return s.EnsureInboundTagAllowedTx(database.GetDB(), nodeID, tag)
|
||||
}
|
||||
|
||||
func (s *NodeService) EnsureInboundTagAllowedTx(tx *gorm.DB, nodeID int, tag string) error {
|
||||
tag = strings.TrimSpace(tag)
|
||||
if nodeID <= 0 || tag == "" {
|
||||
return nil
|
||||
}
|
||||
db := database.GetDB()
|
||||
if tx == nil {
|
||||
tx = database.GetDB()
|
||||
}
|
||||
node := &model.Node{}
|
||||
if err := db.Where("id = ?", nodeID).First(node).Error; err != nil {
|
||||
if err := tx.Where("id = ?", nodeID).First(node).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if node.InboundSyncMode != "selected" {
|
||||
@@ -519,7 +525,7 @@ func (s *NodeService) EnsureInboundTagAllowed(nodeID int, tag string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return db.Model(model.Node{}).Where("id = ?", nodeID).
|
||||
return tx.Model(model.Node{}).Where("id = ?", nodeID).
|
||||
Updates(map[string]any{"inbound_tags": string(buf)}).Error
|
||||
}
|
||||
|
||||
|
||||
@@ -16,21 +16,36 @@ import (
|
||||
// fakeNodeRuntime is a runtime.Runtime stub that counts the per-client dispatch
|
||||
// calls so a test can assert a bulk op does NOT stream one RPC per client.
|
||||
type fakeNodeRuntime struct {
|
||||
addClient atomic.Int32
|
||||
deleteUser atomic.Int32
|
||||
deleteClient atomic.Int32
|
||||
updateUser atomic.Int32
|
||||
addInbound atomic.Int32
|
||||
delInbound atomic.Int32
|
||||
addClient atomic.Int32
|
||||
deleteClient atomic.Int32
|
||||
deleteUser atomic.Int32
|
||||
updateInbound atomic.Int32
|
||||
updateUser atomic.Int32
|
||||
}
|
||||
|
||||
func (f *fakeNodeRuntime) Name() string { return "fake-node" }
|
||||
|
||||
func (f *fakeNodeRuntime) AddInbound(context.Context, *model.Inbound) error { return nil }
|
||||
func (f *fakeNodeRuntime) DelInbound(context.Context, *model.Inbound) error { return nil }
|
||||
func (f *fakeNodeRuntime) UpdateInbound(context.Context, *model.Inbound, *model.Inbound) error {
|
||||
func (f *fakeNodeRuntime) AddInbound(context.Context, *model.Inbound) error {
|
||||
f.addInbound.Add(1)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeNodeRuntime) DelInbound(context.Context, *model.Inbound) error {
|
||||
f.delInbound.Add(1)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeNodeRuntime) UpdateInbound(context.Context, *model.Inbound, *model.Inbound) error {
|
||||
f.updateInbound.Add(1)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeNodeRuntime) AddUser(context.Context, *model.Inbound, map[string]any) error { return nil }
|
||||
func (f *fakeNodeRuntime) RemoveUser(context.Context, *model.Inbound, string) error { return nil }
|
||||
|
||||
func (f *fakeNodeRuntime) RemoveUser(context.Context, *model.Inbound, string) error { return nil }
|
||||
|
||||
func (f *fakeNodeRuntime) UpdateUser(context.Context, *model.Inbound, string, model.Client) error {
|
||||
f.updateUser.Add(1)
|
||||
return nil
|
||||
@@ -67,10 +82,13 @@ func setupNodeRuntime(t *testing.T) (int, *fakeNodeRuntime) {
|
||||
runtime.SetManager(mgr)
|
||||
t.Cleanup(func() { runtime.SetManager(prev) })
|
||||
|
||||
node := &model.Node{Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Status: "online"}
|
||||
node := &model.Node{Name: "n1-" + 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)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = database.GetDB().Where("id = ?", node.Id).Delete(&model.Node{}).Error
|
||||
})
|
||||
fake := &fakeNodeRuntime{}
|
||||
mgr.SetRuntimeOverride(node.Id, fake)
|
||||
return node.Id, fake
|
||||
@@ -250,3 +268,39 @@ func TestNodeBulk_LargeDeleteFoldsToDirty(t *testing.T) {
|
||||
t.Fatal("large delete must mark the node dirty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDelInbound_NodeSelectedModeDeletesRemoteImmediately(t *testing.T) {
|
||||
setupBulkDB(t)
|
||||
nodeID, fake := setupNodeRuntime(t)
|
||||
if err := database.GetDB().Model(&model.Node{}).Where("id = ?", nodeID).
|
||||
Updates(map[string]any{
|
||||
"inbound_sync_mode": "selected",
|
||||
"inbound_tags": []string{"other-tag"},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("set selected mode: %v", err)
|
||||
}
|
||||
ib := nodeInbound(t, nodeID, 30004, makeNodeClients(1))
|
||||
|
||||
needRestart, err := (&InboundService{}).DelInbound(ib.Id)
|
||||
if err != nil {
|
||||
t.Fatalf("DelInbound: %v", err)
|
||||
}
|
||||
if needRestart {
|
||||
t.Fatal("node-owned delete should not request local restart")
|
||||
}
|
||||
if got := fake.delInbound.Load(); got != 1 {
|
||||
t.Fatalf("node-owned delete streamed %d DelInbound RPCs, want 1", got)
|
||||
}
|
||||
var count int64
|
||||
if err := database.GetDB().Model(&model.Inbound{}).Where("id = ?", ib.Id).Count(&count).Error; err != nil {
|
||||
t.Fatalf("count inbound: %v", err)
|
||||
}
|
||||
if count != 0 {
|
||||
t.Fatalf("deleted inbound row count = %d, want 0", count)
|
||||
}
|
||||
if _, _, dirty, _, err := (&NodeService{}).NodeSyncState(nodeID); err != nil {
|
||||
t.Fatalf("NodeSyncState: %v", err)
|
||||
} else if !dirty {
|
||||
t.Fatal("node-owned delete should still mark the node dirty as reconcile backup")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,6 +71,7 @@ func TestGetAll_ClientBreakdownMatchesByEmailNotStaleInboundId(t *testing.T) {
|
||||
}
|
||||
if n == nil {
|
||||
t.Fatal("node 1 not found")
|
||||
return
|
||||
}
|
||||
if n.ClientCount != 3 {
|
||||
t.Errorf("ClientCount = %d, want 3", n.ClientCount)
|
||||
|
||||
@@ -62,6 +62,7 @@ func TestGetNodeTree_SurfacesTransitiveNodeNestedUnderParent(t *testing.T) {
|
||||
}
|
||||
if node2 == nil || node3 == nil {
|
||||
t.Fatalf("expected Node2 + transitive Node3, got %d nodes", len(tree))
|
||||
return
|
||||
}
|
||||
if node2.ParentGuid != selfGuid {
|
||||
t.Errorf("Node2 parent = %q, want this panel's GUID %q", node2.ParentGuid, selfGuid)
|
||||
|
||||
Reference in New Issue
Block a user