diff --git a/internal/web/runtime/manager.go b/internal/web/runtime/manager.go index cdfa39daf..ff6afc2fd 100644 --- a/internal/web/runtime/manager.go +++ b/internal/web/runtime/manager.go @@ -18,6 +18,7 @@ type Manager struct { mu sync.RWMutex remotes map[int]*Remote overrides map[int]Runtime // test-only: forces RuntimeFor to return a stub + localOverride Runtime // test-only: forces RuntimeFor(nil) to return a stub egressResolver NodeEgressResolver } @@ -44,6 +45,15 @@ func (m *Manager) SetRuntimeOverride(nodeID int, rt Runtime) { m.overrides[nodeID] = rt } +// SetLocalRuntimeOverride makes RuntimeFor(nil) return rt instead of the real +// local runtime. Test seam for exercising the local dispatch path (MTProto +// sidecar, local Xray) without a running child process; pass nil rt to clear. +func (m *Manager) SetLocalRuntimeOverride(rt Runtime) { + m.mu.Lock() + defer m.mu.Unlock() + m.localOverride = rt +} + func (m *Manager) SetNodeEgressResolver(r NodeEgressResolver) { m.mu.Lock() defer m.mu.Unlock() @@ -61,6 +71,13 @@ func (m *Manager) NodeEgressProxyURL(nodeID int) string { func (m *Manager) RuntimeFor(nodeID *int) (Runtime, error) { if nodeID == nil { + m.mu.RLock() + if m.localOverride != nil { + rt := m.localOverride + m.mu.RUnlock() + return rt, nil + } + m.mu.RUnlock() return m.local, nil } m.mu.RLock() diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index fc1149126..e2346318d 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -1303,13 +1303,16 @@ func (s *InboundService) UpdateInbound(inbound *model.Inbound) (*model.Inbound, pushable = false } } + newProtocolIsMtproto := oldInbound.Protocol == model.MTProto if pushable { - if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, payload); err2 == nil { - logger.Debug("Updated inbound applied on", rt.Name(), ":", oldInbound.Tag) - } else { - logger.Debug("Unable to update inbound on", rt.Name(), ":", err2) - if oldInbound.Protocol != model.MTProto { - needRestart = true + postCommitApply = func() { + if err2 := rt.UpdateInbound(context.Background(), &oldSnapshot, payload); err2 == nil { + logger.Debug("Updated inbound applied on", rt.Name(), ":", oldInbound.Tag) + } else { + logger.Debug("Unable to update inbound on", rt.Name(), ":", err2) + if !newProtocolIsMtproto { + needRestart = true + } } } } diff --git a/internal/web/service/inbound_mtproto_txfail_test.go b/internal/web/service/inbound_mtproto_txfail_test.go new file mode 100644 index 000000000..190c4251c --- /dev/null +++ b/internal/web/service/inbound_mtproto_txfail_test.go @@ -0,0 +1,53 @@ +package service + +import ( + "errors" + "testing" + + "gorm.io/gorm" + + "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" +) + +// A local MTProto inbound edit must not push to the managed sidecar from inside +// the serialized write transaction: that blocks the single traffic-writer +// goroutine on process/network I/O, and a later step failing the transaction +// would leave the sidecar ahead of the rolled-back database. The push belongs in +// the post-commit hook, exactly as the xray branch already does it. +func TestUpdateInboundLocalMtprotoDefersPushUntilCommit(t *testing.T) { + setupConflictDB(t) + + mgr := runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }}) + fake := &fakeNodeRuntime{} + mgr.SetLocalRuntimeOverride(fake) + runtime.SetManager(mgr) + t.Cleanup(func() { runtime.SetManager(nil) }) + + seedInboundConflict(t, "mt-txfail", "", 46150, model.MTProto, "", + `{"clients":[{"email":"mtx","secret":"`+mtprotoTestSecretA+`","enable":true}]}`) + seeded := loadInboundByTag(t, "mt-txfail") + seedClientTraffic(t, seeded.Id, "mtx", true) + + db := database.GetDB() + const cbName = "b1-05:fail-inbound-update" + if err := db.Callback().Update().After("gorm:update").Register(cbName, func(tx *gorm.DB) { + if tx.Statement != nil && tx.Statement.Table == "inbounds" { + tx.AddError(errors.New("injected transaction failure")) + } + }); err != nil { + t.Fatalf("register callback: %v", err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(cbName) }) + + update := *loadInboundByTag(t, "mt-txfail") + update.Remark = "edited" + if _, _, err := (&InboundService{}).UpdateInbound(&update); err == nil { + t.Fatal("UpdateInbound: expected the injected transaction failure") + } + + if n := fake.updateInbound.Load(); n != 0 { + t.Fatalf("the MTProto sidecar push ran %d time(s) inside the failed transaction; it must be deferred until the commit succeeds", n) + } +}