From 0dfc8548a025c43213cd3148c3f7643d9a56a55d Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 4 Nov 2026 04:16:18 +0330 Subject: [PATCH 1/4] fix(xray): let a restart abort a user sync stuck on an unresponsive core --- backend/xray/jobs.go | 1 + backend/xray/restart_stuck_sync_unix_test.go | 203 +++++++++++++++++++ backend/xray/user.go | 10 + backend/xray/xray.go | 30 +++ 4 files changed, 244 insertions(+) create mode 100644 backend/xray/restart_stuck_sync_unix_test.go diff --git a/backend/xray/jobs.go b/backend/xray/jobs.go index c1bf31b4..720f4945 100644 --- a/backend/xray/jobs.go +++ b/backend/xray/jobs.go @@ -97,6 +97,7 @@ func (x *Xray) checkXrayHealth(baseCtx context.Context) { restart := func(reason string) { // Lock order: syncMu, then mu, then core.mu, as in the user syncs. // syncMu keeps user syncs off the new core until its API answers. + x.abortRunningSync() x.syncMu.Lock() defer x.syncMu.Unlock() diff --git a/backend/xray/restart_stuck_sync_unix_test.go b/backend/xray/restart_stuck_sync_unix_test.go new file mode 100644 index 00000000..b754a280 --- /dev/null +++ b/backend/xray/restart_stuck_sync_unix_test.go @@ -0,0 +1,203 @@ +//go:build !windows + +package xray + +import ( + "context" + "net" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/xtls/xray-core/app/proxyman/command" + statsCommand "github.com/xtls/xray-core/app/stats/command" + "google.golang.org/grpc" + + "github.com/pasarguard/node/backend/xray/api" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/pkg/logstream" +) + +type hungInboundService struct { + command.UnimplementedHandlerServiceServer + reached chan struct{} + once sync.Once +} + +func (s *hungInboundService) AlterInbound(ctx context.Context, _ *command.AlterInboundRequest) (*command.AlterInboundResponse, error) { + s.once.Do(func() { close(s.reached) }) + <-ctx.Done() + return nil, ctx.Err() +} + +func newHungInboundXray(t *testing.T) (*Xray, *hungInboundService, *fakeStatsService) { + t.Helper() + + dir := t.TempDir() + executable := filepath.Join(dir, "xray") + if err := os.WriteFile(executable, []byte("#!/bin/sh\nexec sleep 600\n"), 0o755); err != nil { + t.Fatal(err) + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + hung := &hungInboundService{reached: make(chan struct{})} + stats := &fakeStatsService{} + server := grpc.NewServer() + command.RegisterHandlerServiceServer(server, hung) + statsCommand.RegisterStatsServiceServer(server, stats) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + + handler, err := api.NewXrayAPI(listener.Addr().(*net.TCPAddr).Port) + if err != nil { + t.Fatal(err) + } + t.Cleanup(handler.Close) + + cfg, err := NewConfig(recordingInbounds, nil) + if err != nil { + t.Fatal(err) + } + core := &Core{ + executablePath: executable, + assetsPath: dir, + configPath: dir, + logs: logstream.NewBuffer(100), + logPhase: logPhaseRuntime, + startupLogSize: 200, + runtimeLogs: newStartupLogRing(10), + } + t.Cleanup(core.Stop) + return &Xray{config: cfg, cfg: &config.Config{}, core: core, handler: handler}, hung, stats +} + +func startHungUserSync(t *testing.T, x *Xray, hung *hungInboundService) (context.CancelFunc, <-chan error) { + t.Helper() + + syncCtx, stopSync := context.WithCancel(context.Background()) + t.Cleanup(stopSync) + synced := make(chan error, 1) + go func() { synced <- x.SyncUser(syncCtx, recordingUser("1.alice", "vless")) }() + select { + case <-hung.reached: + case <-time.After(10 * time.Second): + t.Fatal("the user sync never reached the core") + } + return stopSync, synced +} + +func TestRestartIsNotBlockedByAUserSyncOnAnUnresponsiveCore(t *testing.T) { + x, hung, stats := newHungInboundXray(t) + stopSync, synced := startHungUserSync(t, x, hung) + + restarted := make(chan error, 1) + go func() { restarted <- x.Restart() }() + + deadline := time.Now().Add(10 * time.Second) + for !x.core.Started() && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + blocked := !x.core.Started() + + stopSync() + stats.healthy.Store(true) + select { + case <-restarted: + case <-time.After(45 * time.Second): + t.Fatal("the restart never returned after the user sync was cancelled") + } + <-synced + + if blocked { + t.Fatal("Restart did not start a new core within 10s: it is still waiting for syncMu, held by a user sync whose AlterInbound call the unresponsive core never answers") + } +} + +func TestHealthLoopRestartsAnUnresponsiveCoreWhileAUserSyncIsStuckOnIt(t *testing.T) { + x, hung, stats := newHungInboundXray(t) + stats.healthy.Store(true) + if err := x.Restart(); err != nil { + t.Fatal(err) + } + first := corePID(x) + + stopSync, synced := startHungUserSync(t, x, hung) + stats.healthy.Store(false) + + loopCtx, stopLoop := context.WithCancel(context.Background()) + t.Cleanup(stopLoop) + looped := make(chan struct{}) + go func() { + x.checkXrayHealth(loopCtx) + close(looped) + }() + + replaced := false + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + if pid := corePID(x); pid != 0 && pid != first { + replaced = true + break + } + time.Sleep(100 * time.Millisecond) + } + + stopLoop() + stopSync() + stats.healthy.Store(true) + select { + case <-looped: + case <-time.After(45 * time.Second): + t.Fatal("the health loop never returned after it was cancelled") + } + <-synced + + if !replaced { + t.Fatal("the health loop did not replace the unresponsive core within 60s: its restart is waiting for syncMu, held by a user sync whose AlterInbound call the core never answers") + } +} + +func TestHealthLoopRestartsAnUnresponsiveCoreWithNoUserSyncInFlight(t *testing.T) { + x, _, stats := newHungInboundXray(t) + stats.healthy.Store(true) + if err := x.Restart(); err != nil { + t.Fatal(err) + } + first := corePID(x) + stats.healthy.Store(false) + + loopCtx, stopLoop := context.WithCancel(context.Background()) + t.Cleanup(stopLoop) + looped := make(chan struct{}) + go func() { + x.checkXrayHealth(loopCtx) + close(looped) + }() + + replaced := false + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + if pid := corePID(x); pid != 0 && pid != first { + replaced = true + break + } + time.Sleep(100 * time.Millisecond) + } + + stopLoop() + stats.healthy.Store(true) + select { + case <-looped: + case <-time.After(45 * time.Second): + t.Fatal("the health loop never returned after it was cancelled") + } + + if !replaced { + t.Fatal("the health loop did not replace the unresponsive core within 60s") + } +} diff --git a/backend/xray/user.go b/backend/xray/user.go index fb6ecaaa..4ea21c07 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -242,6 +242,8 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { if err := ctx.Err(); err != nil { return err } + ctx, endSync := x.beginSync(ctx) + defer endSync() proxySetting, wireguardErr := setupUserAccount(user) @@ -291,6 +293,9 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { } } + if syncAborted(ctx) { + return fmt.Errorf("failed to sync user %s: %w", user.GetEmail(), errSyncAborted) + } if errMessage.String() != "" { return errors.New("failed to add user:" + errMessage.String()) } @@ -372,6 +377,8 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { if err := ctx.Err(); err != nil { return err } + ctx, endSync := x.beginSync(ctx) + defer endSync() handler := x.handler inboundByTag, updates := x.config.buildInboundUpdates(users) @@ -414,6 +421,9 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { } } + if syncAborted(ctx) { + return fmt.Errorf("failed to update users: %w", errSyncAborted) + } if errMessage.String() != "" { return errors.New("failed to update users:" + errMessage.String()) } diff --git a/backend/xray/xray.go b/backend/xray/xray.go index b9bc59c0..898e8ff7 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -26,11 +26,40 @@ type Xray struct { cancelFunc context.CancelFunc mu sync.RWMutex syncMu sync.Mutex + abortMu sync.Mutex + abortSync context.CancelCauseFunc closed bool // set by Shutdown under mu } var errXrayShutdown = errors.New("xray backend is shut down") +var errSyncAborted = errors.New("user sync aborted by a core restart") + +func (x *Xray) beginSync(ctx context.Context) (context.Context, func()) { + syncCtx, cancel := context.WithCancelCause(ctx) + x.abortMu.Lock() + x.abortSync = cancel + x.abortMu.Unlock() + return syncCtx, func() { + x.abortMu.Lock() + x.abortSync = nil + x.abortMu.Unlock() + cancel(nil) + } +} + +func (x *Xray) abortRunningSync() { + x.abortMu.Lock() + defer x.abortMu.Unlock() + if x.abortSync != nil { + x.abortSync(errSyncAborted) + } +} + +func syncAborted(ctx context.Context) bool { + return errors.Is(context.Cause(ctx), errSyncAborted) +} + // restartReadyTimeout bounds how long a restart holds syncMu waiting for the // new core's API. const restartReadyTimeout = 30 * time.Second @@ -152,6 +181,7 @@ func (x *Xray) Started() bool { // Restart restarts the core from the current config and waits until its API // answers. func (x *Xray) Restart() error { + x.abortRunningSync() x.syncMu.Lock() defer x.syncMu.Unlock() From 00d1edc668807404e4c68e93f9843ff0305146d1 Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 4 Nov 2026 04:50:06 +0330 Subject: [PATCH 2/4] fix(xray): keep user syncs off syncMu until a pending restart holds it A sync queued on syncMu could take the lock after a restart aborted the running one and wedge the restart again. While a restart waits for syncMu, new syncs now fail fast. The health loop no longer skips its restart when it had to abort a sync, so the core is rebuilt from the current config. --- backend/xray/jobs.go | 5 +- backend/xray/restart_stuck_sync_unix_test.go | 103 ++++++++++++++++++- backend/xray/xray.go | 29 +++++- 3 files changed, 127 insertions(+), 10 deletions(-) diff --git a/backend/xray/jobs.go b/backend/xray/jobs.go index 720f4945..ac69a15b 100644 --- a/backend/xray/jobs.go +++ b/backend/xray/jobs.go @@ -97,13 +97,12 @@ func (x *Xray) checkXrayHealth(baseCtx context.Context) { restart := func(reason string) { // Lock order: syncMu, then mu, then core.mu, as in the user syncs. // syncMu keeps user syncs off the new core until its API answers. - x.abortRunningSync() - x.syncMu.Lock() + syncAborted := x.lockSyncForRestart() defer x.syncMu.Unlock() // A full sync may have replaced the core while this restart waited // for syncMu: leave a core that answers alone. - if x.Started() && probe() == nil { + if !syncAborted && x.Started() && probe() == nil { log.Println("xray answers again, health restart skipped") consecutiveFailures = 0 return diff --git a/backend/xray/restart_stuck_sync_unix_test.go b/backend/xray/restart_stuck_sync_unix_test.go index b754a280..c1561b6d 100644 --- a/backend/xray/restart_stuck_sync_unix_test.go +++ b/backend/xray/restart_stuck_sync_unix_test.go @@ -4,10 +4,12 @@ package xray import ( "context" + "errors" "net" "os" "path/filepath" "sync" + "sync/atomic" "testing" "time" @@ -34,6 +36,13 @@ func (s *hungInboundService) AlterInbound(ctx context.Context, _ *command.AlterI func newHungInboundXray(t *testing.T) (*Xray, *hungInboundService, *fakeStatsService) { t.Helper() + stats := &fakeStatsService{} + x, hung := newHungInboundXrayWithStats(t, stats) + return x, hung, stats +} + +func newHungInboundXrayWithStats(t *testing.T, stats statsCommand.StatsServiceServer) (*Xray, *hungInboundService) { + t.Helper() dir := t.TempDir() executable := filepath.Join(dir, "xray") @@ -46,7 +55,6 @@ func newHungInboundXray(t *testing.T) (*Xray, *hungInboundService, *fakeStatsSer t.Fatal(err) } hung := &hungInboundService{reached: make(chan struct{})} - stats := &fakeStatsService{} server := grpc.NewServer() command.RegisterHandlerServiceServer(server, hung) statsCommand.RegisterStatsServiceServer(server, stats) @@ -73,7 +81,7 @@ func newHungInboundXray(t *testing.T) (*Xray, *hungInboundService, *fakeStatsSer runtimeLogs: newStartupLogRing(10), } t.Cleanup(core.Stop) - return &Xray{config: cfg, cfg: &config.Config{}, core: core, handler: handler}, hung, stats + return &Xray{config: cfg, cfg: &config.Config{}, core: core, handler: handler}, hung } func startHungUserSync(t *testing.T, x *Xray, hung *hungInboundService) (context.CancelFunc, <-chan error) { @@ -201,3 +209,94 @@ func TestHealthLoopRestartsAnUnresponsiveCoreWithNoUserSyncInFlight(t *testing.T t.Fatal("the health loop did not replace the unresponsive core within 60s") } } + +func TestRestartIsNotBlockedByAUserSyncQueuedBehindAStuckOne(t *testing.T) { + x, hung, stats := newHungInboundXray(t) + stopSync, synced := startHungUserSync(t, x, hung) + + queuedCtx, stopQueued := context.WithCancel(context.Background()) + t.Cleanup(stopQueued) + queued := make(chan error, 1) + go func() { queued <- x.SyncUser(queuedCtx, recordingUser("2.bob", "vless")) }() + time.Sleep(300 * time.Millisecond) + + restarted := make(chan error, 1) + go func() { restarted <- x.Restart() }() + + deadline := time.Now().Add(10 * time.Second) + for !x.core.Started() && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + blocked := !x.core.Started() + + stopSync() + stopQueued() + stats.healthy.Store(true) + select { + case <-restarted: + case <-time.After(45 * time.Second): + t.Fatal("the restart never returned after the user syncs were cancelled") + } + <-synced + <-queued + + if blocked { + t.Fatal("Restart did not start a new core within 10s: a second user sync queued on syncMu took the lock after the first one was aborted and is now stuck on the unresponsive core") + } +} + +type healingStatsService struct { + *fakeStatsService + failuresLeft atomic.Int32 +} + +func (s *healingStatsService) GetSysStats(ctx context.Context, req *statsCommand.SysStatsRequest) (*statsCommand.SysStatsResponse, error) { + if s.failuresLeft.Add(-1) >= 0 { + return nil, errors.New("core busy") + } + return s.fakeStatsService.GetSysStats(ctx, req) +} + +func TestHealthLoopReplacesTheCoreAfterAbortingAUserSyncEvenWhenTheCoreAnswersAgain(t *testing.T) { + stats := &healingStatsService{fakeStatsService: &fakeStatsService{}} + stats.healthy.Store(true) + x, hung := newHungInboundXrayWithStats(t, stats) + if err := x.Restart(); err != nil { + t.Fatal(err) + } + first := corePID(x) + + stopSync, synced := startHungUserSync(t, x, hung) + stats.failuresLeft.Store(10) + + loopCtx, stopLoop := context.WithCancel(context.Background()) + t.Cleanup(stopLoop) + looped := make(chan struct{}) + go func() { + x.checkXrayHealth(loopCtx) + close(looped) + }() + + replaced := false + deadline := time.Now().Add(45 * time.Second) + for time.Now().Before(deadline) { + if pid := corePID(x); pid != 0 && pid != first { + replaced = true + break + } + time.Sleep(100 * time.Millisecond) + } + + stopLoop() + stopSync() + select { + case <-looped: + case <-time.After(45 * time.Second): + t.Fatal("the health loop never returned after it was cancelled") + } + <-synced + + if !replaced { + t.Fatal("the health loop aborted a user sync and then skipped the restart because the core answered the re-probe: the core may now lack users that x.config holds") + } +} diff --git a/backend/xray/xray.go b/backend/xray/xray.go index 898e8ff7..815202de 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -28,6 +28,8 @@ type Xray struct { syncMu sync.Mutex abortMu sync.Mutex abortSync context.CancelCauseFunc + restarts int + aborted bool closed bool // set by Shutdown under mu } @@ -38,7 +40,12 @@ var errSyncAborted = errors.New("user sync aborted by a core restart") func (x *Xray) beginSync(ctx context.Context) (context.Context, func()) { syncCtx, cancel := context.WithCancelCause(ctx) x.abortMu.Lock() - x.abortSync = cancel + if x.restarts > 0 { + x.aborted = true + cancel(errSyncAborted) + } else { + x.abortSync = cancel + } x.abortMu.Unlock() return syncCtx, func() { x.abortMu.Lock() @@ -48,12 +55,25 @@ func (x *Xray) beginSync(ctx context.Context) (context.Context, func()) { } } -func (x *Xray) abortRunningSync() { +func (x *Xray) lockSyncForRestart() (syncAborted bool) { x.abortMu.Lock() - defer x.abortMu.Unlock() + x.restarts++ if x.abortSync != nil { x.abortSync(errSyncAborted) + x.aborted = true } + x.abortMu.Unlock() + + x.syncMu.Lock() + + x.abortMu.Lock() + defer x.abortMu.Unlock() + x.restarts-- + syncAborted = x.aborted + if x.restarts == 0 { + x.aborted = false + } + return syncAborted } func syncAborted(ctx context.Context) bool { @@ -181,8 +201,7 @@ func (x *Xray) Started() bool { // Restart restarts the core from the current config and waits until its API // answers. func (x *Xray) Restart() error { - x.abortRunningSync() - x.syncMu.Lock() + x.lockSyncForRestart() defer x.syncMu.Unlock() return x.restartLocked(context.Background()) From d814391fc3560f95672e369f7c04440df61c1fa5 Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 4 Nov 2026 14:29:55 +0330 Subject: [PATCH 3/4] fix(xray): keep the intended account when a restart aborts a user sync A restart that aborts a running SyncUser or UpdateUsers no longer drops the account from the in-memory config, so the core it rebuilds from that config carries the update. The abort flag is consumed by every restart, and an aborted WireGuard remove-first path reports the abort. --- backend/xray/jobs.go | 4 +- backend/xray/restart_stuck_sync_unix_test.go | 39 ++++++++++++++++++++ backend/xray/user.go | 8 ++-- backend/xray/xray.go | 10 ++--- 4 files changed, 49 insertions(+), 12 deletions(-) diff --git a/backend/xray/jobs.go b/backend/xray/jobs.go index ac69a15b..a232fdf2 100644 --- a/backend/xray/jobs.go +++ b/backend/xray/jobs.go @@ -97,12 +97,12 @@ func (x *Xray) checkXrayHealth(baseCtx context.Context) { restart := func(reason string) { // Lock order: syncMu, then mu, then core.mu, as in the user syncs. // syncMu keeps user syncs off the new core until its API answers. - syncAborted := x.lockSyncForRestart() + aborted := x.lockSyncForRestart() defer x.syncMu.Unlock() // A full sync may have replaced the core while this restart waited // for syncMu: leave a core that answers alone. - if !syncAborted && x.Started() && probe() == nil { + if !aborted && x.Started() && probe() == nil { log.Println("xray answers again, health restart skipped") consecutiveFailures = 0 return diff --git a/backend/xray/restart_stuck_sync_unix_test.go b/backend/xray/restart_stuck_sync_unix_test.go index c1561b6d..854918c9 100644 --- a/backend/xray/restart_stuck_sync_unix_test.go +++ b/backend/xray/restart_stuck_sync_unix_test.go @@ -300,3 +300,42 @@ func TestHealthLoopReplacesTheCoreAfterAbortingAUserSyncEvenWhenTheCoreAnswersAg t.Fatal("the health loop aborted a user sync and then skipped the restart because the core answered the re-probe: the core may now lack users that x.config holds") } } + +func TestRestartDuringAUserSyncKeepsTheUserForTheRebuiltCore(t *testing.T) { + x, hung, stats := newHungInboundXray(t) + stats.healthy.Store(true) + _, synced := startHungUserSync(t, x, hung) + + restarted := make(chan error, 1) + go func() { restarted <- x.Restart() }() + select { + case <-restarted: + case <-time.After(45 * time.Second): + t.Fatal("the restart never returned") + } + select { + case err := <-synced: + if !errors.Is(err, errSyncAborted) { + t.Fatalf("the aborted user sync returned %v, want errSyncAborted", err) + } + case <-time.After(10 * time.Second): + t.Fatal("the user sync never returned after the restart") + } + + x.mu.RLock() + cfg := x.config + x.mu.RUnlock() + for _, inbound := range cfg.InboundConfigs { + if inbound.Tag != "vless" { + continue + } + inbound.mu.RLock() + _, ok := inbound.clients["1.alice"] + inbound.mu.RUnlock() + if !ok { + t.Fatal("Restart aborted the user sync and the sync dropped 1.alice from x.config, so the rebuilt core does not have the user") + } + return + } + t.Fatal("no vless inbound in x.config") +} diff --git a/backend/xray/user.go b/backend/xray/user.go index 4ea21c07..7929b456 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -276,13 +276,13 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { if isActive { removeFirst := inbound.removeBeforeAdd(account) if removeFirst { - if err := handler.RemoveInboundUser(ctx, inbound.Tag, user.GetEmail()); err != nil && inbound.Protocol == Wireguard { + if err := handler.RemoveInboundUser(ctx, inbound.Tag, user.GetEmail()); err != nil && inbound.Protocol == Wireguard && !syncAborted(ctx) { return fmt.Errorf("failed to remove wireguard user %s from inbound %s: %w", user.GetEmail(), inbound.Tag, err) } } inbound.updateUser(account) err := handler.AddInboundUser(ctx, inbound.Tag, accountForAPI(inbound, account)) - if err != nil { + if err != nil && !syncAborted(ctx) { inbound.removeUser(user.GetEmail()) log.Println(err) errMessage.WriteString("\n" + err.Error()) @@ -404,14 +404,14 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { for n, account := range accounts { if removeFirst[n] { - if err := handler.RemoveInboundUser(ctx, tag, account.GetEmail()); err != nil && inbound.Protocol == Wireguard { + if err := handler.RemoveInboundUser(ctx, tag, account.GetEmail()); err != nil && inbound.Protocol == Wireguard && !syncAborted(ctx) { removeErr := fmt.Errorf("failed to remove wireguard user %s from inbound %s: %w", account.GetEmail(), tag, err) log.Println(removeErr) errMessage.WriteString("\n" + removeErr.Error()) continue } } - if err := handler.AddInboundUser(ctx, tag, accountForAPI(inbound, account)); err != nil { + if err := handler.AddInboundUser(ctx, tag, accountForAPI(inbound, account)); err != nil && !syncAborted(ctx) { inbound.removeUser(account.GetEmail()) log.Println(err) errMessage.WriteString("\n" + err.Error()) diff --git a/backend/xray/xray.go b/backend/xray/xray.go index 815202de..72a57c70 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -55,7 +55,7 @@ func (x *Xray) beginSync(ctx context.Context) (context.Context, func()) { } } -func (x *Xray) lockSyncForRestart() (syncAborted bool) { +func (x *Xray) lockSyncForRestart() (aborted bool) { x.abortMu.Lock() x.restarts++ if x.abortSync != nil { @@ -69,11 +69,9 @@ func (x *Xray) lockSyncForRestart() (syncAborted bool) { x.abortMu.Lock() defer x.abortMu.Unlock() x.restarts-- - syncAborted = x.aborted - if x.restarts == 0 { - x.aborted = false - } - return syncAborted + aborted = x.aborted + x.aborted = false + return aborted } func syncAborted(ctx context.Context) bool { From f9a2f8eaec7251c2a91b467a1bb63d8051ac049e Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Thu, 5 Nov 2026 00:35:11 +0330 Subject: [PATCH 4/4] fix(xray): answer a user sync aborted by a core restart as retry-later An aborted SyncUser or UpdateUsers now returns a backend.RetryLaterError, so the node answers it as gRPC Unavailable / HTTP 503 instead of Internal / 500, and a client retries the batch instead of splitting it to find refused users. errors.Is(err, errSyncAborted) still holds. --- backend/xray/restart_stuck_sync_unix_test.go | 39 ++++++ backend/xray/user.go | 5 +- controller/rest/sync_abort_unix_test.go | 135 +++++++++++++++++++ controller/rpc/sync_abort_unix_test.go | 124 +++++++++++++++++ 4 files changed, 301 insertions(+), 2 deletions(-) create mode 100644 controller/rest/sync_abort_unix_test.go create mode 100644 controller/rpc/sync_abort_unix_test.go diff --git a/backend/xray/restart_stuck_sync_unix_test.go b/backend/xray/restart_stuck_sync_unix_test.go index 854918c9..fd670d1f 100644 --- a/backend/xray/restart_stuck_sync_unix_test.go +++ b/backend/xray/restart_stuck_sync_unix_test.go @@ -17,7 +17,9 @@ import ( statsCommand "github.com/xtls/xray-core/app/stats/command" "google.golang.org/grpc" + "github.com/pasarguard/node/backend" "github.com/pasarguard/node/backend/xray/api" + "github.com/pasarguard/node/common" "github.com/pasarguard/node/config" "github.com/pasarguard/node/pkg/logstream" ) @@ -318,6 +320,9 @@ func TestRestartDuringAUserSyncKeepsTheUserForTheRebuiltCore(t *testing.T) { if !errors.Is(err, errSyncAborted) { t.Fatalf("the aborted user sync returned %v, want errSyncAborted", err) } + if !backend.IsRetryLater(err) { + t.Fatalf("the aborted user sync returned %v as a plain error, which the node answers as Internal / 500: want a backend.RetryLaterError, answered as Unavailable / 503", err) + } case <-time.After(10 * time.Second): t.Fatal("the user sync never returned after the restart") } @@ -339,3 +344,37 @@ func TestRestartDuringAUserSyncKeepsTheUserForTheRebuiltCore(t *testing.T) { } t.Fatal("no vless inbound in x.config") } + +func TestRestartDuringAUserUpdateAbortsItAsRetryLater(t *testing.T) { + x, hung, stats := newHungInboundXray(t) + stats.healthy.Store(true) + + updated := make(chan error, 1) + go func() { + updated <- x.UpdateUsers(context.Background(), []*common.User{recordingUser("1.alice", "vless")}) + }() + select { + case <-hung.reached: + case <-time.After(10 * time.Second): + t.Fatal("the user update never reached the core") + } + + restarted := make(chan error, 1) + go func() { restarted <- x.Restart() }() + select { + case <-restarted: + case <-time.After(45 * time.Second): + t.Fatal("the restart never returned") + } + select { + case err := <-updated: + if !errors.Is(err, errSyncAborted) { + t.Fatalf("the aborted user update returned %v, want errSyncAborted", err) + } + if !backend.IsRetryLater(err) { + t.Fatalf("the aborted user update returned %v as a plain error, which the node answers as Internal / 500: want a backend.RetryLaterError, answered as Unavailable / 503", err) + } + case <-time.After(10 * time.Second): + t.Fatal("the user update never returned after the restart") + } +} diff --git a/backend/xray/user.go b/backend/xray/user.go index 7929b456..f7b713a4 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "github.com/pasarguard/node/backend" "github.com/pasarguard/node/backend/xray/api" "github.com/pasarguard/node/common" ) @@ -294,7 +295,7 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { } if syncAborted(ctx) { - return fmt.Errorf("failed to sync user %s: %w", user.GetEmail(), errSyncAborted) + return backend.RetryLater(fmt.Errorf("failed to sync user %s: %w", user.GetEmail(), errSyncAborted)) } if errMessage.String() != "" { return errors.New("failed to add user:" + errMessage.String()) @@ -422,7 +423,7 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { } if syncAborted(ctx) { - return fmt.Errorf("failed to update users: %w", errSyncAborted) + return backend.RetryLater(fmt.Errorf("failed to update users: %w", errSyncAborted)) } if errMessage.String() != "" { return errors.New("failed to update users:" + errMessage.String()) diff --git a/controller/rest/sync_abort_unix_test.go b/controller/rest/sync_abort_unix_test.go new file mode 100644 index 00000000..106e9431 --- /dev/null +++ b/controller/rest/sync_abort_unix_test.go @@ -0,0 +1,135 @@ +//go:build !windows + +package rest + +import ( + "bytes" + "io" + "net/http" + "os" + "os/exec" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/google/uuid" + "google.golang.org/protobuf/proto" + + "github.com/pasarguard/node/backend" + "github.com/pasarguard/node/common" +) + +func runningBackend(t *testing.T) backend.Backend { + t.Helper() + service := sharedTestCtx.service.(*Service) + if back := service.Backend(); back != nil && back.Started() { + return back + } + configFile, err := os.ReadFile(configPath) + if err != nil { + t.Fatal(err) + } + var info common.BaseInfoResponse + if err := sharedTestCtx.createAuthenticatedRequest("POST", "/start", &common.Backend{ + Type: common.BackendType_XRAY, + Config: string(configFile), + }, &info); err != nil { + t.Fatalf("Failed to start backend: %v", err) + } + back := service.Backend() + if back == nil || !back.Started() { + t.Fatal("the backend did not start") + } + return back +} + +func stopCoreProcess(t *testing.T) { + t.Helper() + out, err := exec.Command("pgrep", "-P", strconv.Itoa(os.Getpid())).Output() + if err != nil { + t.Fatalf("no core process under the test process: %v", err) + } + fields := strings.Fields(string(out)) + if len(fields) != 1 { + t.Fatalf("want exactly one core process under the test process, got %q", fields) + } + pid, err := strconv.Atoi(fields[0]) + if err != nil { + t.Fatal(err) + } + if err := syscall.Kill(pid, syscall.SIGSTOP); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = syscall.Kill(pid, syscall.SIGCONT) }) +} + +type syncAnswer struct { + status int + body string + err error +} + +func TestREST_SyncUserAbortedByACoreRestartIsServiceUnavailable(t *testing.T) { + back := runningBackend(t) + stopCoreProcess(t) + + body, err := proto.Marshal(&common.User{ + Email: "abort_rest_user1@example.com", + Inbounds: []string{"VMESS TCP NOTLS"}, + Proxies: &common.Proxy{Vmess: &common.Vmess{Id: uuid.New().String()}}, + }) + if err != nil { + t.Fatal(err) + } + req, err := http.NewRequest("PUT", sharedTestCtx.url+"/user/sync", bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + req.Header.Set("x-api-key", apiKey.String()) + req.Header.Set("Content-Type", "application/x-protobuf") + + answered := make(chan syncAnswer, 1) + go func() { + resp, err := sharedTestCtx.client.Do(req) + if err != nil { + answered <- syncAnswer{err: err} + return + } + defer resp.Body.Close() + text, _ := io.ReadAll(resp.Body) + answered <- syncAnswer{status: resp.StatusCode, body: string(text)} + }() + select { + case got := <-answered: + t.Fatalf("the user sync returned %d %q %v while the core was stopped, want it stuck until the restart", got.status, got.body, got.err) + case <-time.After(2 * time.Second): + } + + restarted := make(chan error, 1) + go func() { restarted <- back.Restart() }() + + select { + case got := <-answered: + if got.err != nil { + t.Fatalf("the user sync request failed: %v", got.err) + } + if got.status != http.StatusServiceUnavailable { + t.Errorf("the user sync aborted by a core restart answered %d %q, want 503 so the bridge retries instead of splitting the batch", got.status, strings.TrimSpace(got.body)) + } + case <-time.After(30 * time.Second): + t.Fatal("the user sync never returned after the restart") + } + select { + case err := <-restarted: + if err != nil { + t.Fatalf("restart failed: %v", err) + } + case <-time.After(60 * time.Second): + t.Fatal("the restart never returned") + } + if !back.Started() { + t.Fatal("the backend has no running core after the restart") + } +} diff --git a/controller/rpc/sync_abort_unix_test.go b/controller/rpc/sync_abort_unix_test.go new file mode 100644 index 00000000..8418e4aa --- /dev/null +++ b/controller/rpc/sync_abort_unix_test.go @@ -0,0 +1,124 @@ +//go:build !windows + +package rpc + +import ( + "context" + "os" + "os/exec" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/google/uuid" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/pasarguard/node/backend" + "github.com/pasarguard/node/common" +) + +func runningBackend(t *testing.T) backend.Backend { + t.Helper() + service := sharedTestCtx.service.(*Service) + if back := service.Backend(); back != nil && back.Started() { + return back + } + configFile, err := os.ReadFile(configPath) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(sharedTestCtx.ctxWithSession, 30*time.Second) + defer cancel() + if _, err := sharedTestCtx.client.Start(ctx, &common.Backend{ + Type: common.BackendType_XRAY, + Config: string(configFile), + KeepAlive: 10, + }); err != nil { + t.Fatalf("Failed to start backend: %v", err) + } + back := service.Backend() + if back == nil || !back.Started() { + t.Fatal("the backend did not start") + } + return back +} + +func stopCoreProcess(t *testing.T) { + t.Helper() + out, err := exec.Command("pgrep", "-P", strconv.Itoa(os.Getpid())).Output() + if err != nil { + t.Fatalf("no core process under the test process: %v", err) + } + fields := strings.Fields(string(out)) + if len(fields) != 1 { + t.Fatalf("want exactly one core process under the test process, got %q", fields) + } + pid, err := strconv.Atoi(fields[0]) + if err != nil { + t.Fatal(err) + } + if err := syscall.Kill(pid, syscall.SIGSTOP); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = syscall.Kill(pid, syscall.SIGCONT) }) +} + +func TestGRPC_SyncUsersChunkedAbortedByACoreRestartIsUnavailable(t *testing.T) { + back := runningBackend(t) + stopCoreProcess(t) + + ctx, cancel := context.WithTimeout(sharedTestCtx.ctxWithSession, 90*time.Second) + defer cancel() + stream, err := sharedTestCtx.client.SyncUsersChunked(ctx) + if err != nil { + t.Fatalf("Failed to open chunked sync stream: %v", err) + } + if err = stream.Send(&common.UsersChunk{ + Index: 0, + Users: []*common.User{{ + Email: "abort_user1@example.com", + Inbounds: []string{"VMESS TCP NOTLS"}, + Proxies: &common.Proxy{Vmess: &common.Vmess{Id: uuid.New().String()}}, + }}, + Last: true, + }); err != nil { + t.Fatalf("Failed to send chunk: %v", err) + } + + answered := make(chan error, 1) + go func() { + _, err := stream.CloseAndRecv() + answered <- err + }() + select { + case err := <-answered: + t.Fatalf("the chunked sync returned %v while the core was stopped, want it stuck until the restart", err) + case <-time.After(2 * time.Second): + } + + restarted := make(chan error, 1) + go func() { restarted <- back.Restart() }() + + select { + case err := <-answered: + if got := status.Code(err); got != codes.Unavailable { + t.Errorf("the chunked sync aborted by a core restart answered %s (%v), want Unavailable so the bridge retries the batch instead of splitting it", got, err) + } + case <-time.After(30 * time.Second): + t.Fatal("the chunked sync never returned after the restart") + } + select { + case err := <-restarted: + if err != nil { + t.Fatalf("restart failed: %v", err) + } + case <-time.After(60 * time.Second): + t.Fatal("the restart never returned") + } + if !back.Started() { + t.Fatal("the backend has no running core after the restart") + } +}