diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 43d265bfc..00f8e3277 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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: diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 38be9a2ff..a5d8e7355 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -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 { diff --git a/internal/web/service/bulk_traffic_test.go b/internal/web/service/bulk_traffic_test.go index e0716e3d2..959d88fbd 100644 --- a/internal/web/service/bulk_traffic_test.go +++ b/internal/web/service/bulk_traffic_test.go @@ -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) diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index cafd52778..bdfba3c60 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -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 diff --git a/internal/web/service/inbound_durable_postgres_test.go b/internal/web/service/inbound_durable_postgres_test.go new file mode 100644 index 000000000..5ec4a47bd --- /dev/null +++ b/internal/web/service/inbound_durable_postgres_test.go @@ -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") + } +} diff --git a/internal/web/service/node.go b/internal/web/service/node.go index 870547f37..f2e6e61a6 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -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 } diff --git a/internal/web/service/node_bulk_dispatch_test.go b/internal/web/service/node_bulk_dispatch_test.go index 8069ffc31..3e071e4b4 100644 --- a/internal/web/service/node_bulk_dispatch_test.go +++ b/internal/web/service/node_bulk_dispatch_test.go @@ -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") + } +} diff --git a/internal/web/service/node_client_breakdown_test.go b/internal/web/service/node_client_breakdown_test.go index 09c17bcde..56522dc4c 100644 --- a/internal/web/service/node_client_breakdown_test.go +++ b/internal/web/service/node_client_breakdown_test.go @@ -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) diff --git a/internal/web/service/node_tree_test.go b/internal/web/service/node_tree_test.go index a8024aaf8..4e7548cd7 100644 --- a/internal/web/service/node_tree_test.go +++ b/internal/web/service/node_tree_test.go @@ -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)