From 03cc80bb9e366aaa3445d785f4f813d5d219f746 Mon Sep 17 00:00:00 2001 From: PathGao <42336971+PathGao@users.noreply.github.com> Date: Thu, 30 Jul 2026 02:56:26 +0800 Subject: [PATCH] fix(mtproto): synchronize child-process lifecycle (#6141) * fix(mtproto): synchronize child-process state Use lifecycle snapshots around the mtg command, completion signal, and exit error so Wait cannot race status and shutdown reads. * test(mtproto): cover concurrent process exit * test(mtproto): cover lifecycle field synchronization --------- Co-authored-by: PathGao --- internal/mtproto/manager_reload_test.go | 10 +++ internal/mtproto/process.go | 79 +++++++++++------- internal/mtproto/process_race_test.go | 106 ++++++++++++++++++++++++ 3 files changed, 166 insertions(+), 29 deletions(-) create mode 100644 internal/mtproto/process_race_test.go diff --git a/internal/mtproto/manager_reload_test.go b/internal/mtproto/manager_reload_test.go index 61a36ca90..7ca7423c7 100644 --- a/internal/mtproto/manager_reload_test.go +++ b/internal/mtproto/manager_reload_test.go @@ -21,6 +21,16 @@ func TestMain(m *testing.M) { fmt.Fprintf(f, "%d\n", os.Getpid()) f.Close() } + if exitFile := os.Getenv("MTG_FAKE_EXIT_FILE"); exitFile != "" { + for { + if _, err := os.Stat(exitFile); err == nil { + os.Exit(1) + } else if !os.IsNotExist(err) { + os.Exit(2) + } + time.Sleep(time.Millisecond) + } + } select {} } os.Exit(m.Run()) diff --git a/internal/mtproto/process.go b/internal/mtproto/process.go index 81d964e4c..44979a8c1 100644 --- a/internal/mtproto/process.go +++ b/internal/mtproto/process.go @@ -109,6 +109,7 @@ func (w *procLogWriter) LastLine() string { // Process wraps a single mtg process invocation for one mtproto inbound. type Process struct { + mu sync.RWMutex cmd *exec.Cmd done chan struct{} configPath string @@ -126,20 +127,20 @@ func newProcess(configPath, label string) *Process { // IsRunning reports whether the mtg process is currently running. func (p *Process) IsRunning() bool { - if p.cmd == nil || p.cmd.Process == nil { + p.mu.RLock() + cmd, done := p.cmd, p.done + p.mu.RUnlock() + if cmd == nil || cmd.Process == nil { return false } - if p.done != nil { + if done != nil { select { - case <-p.done: + case <-done: return false default: } } - if p.cmd.ProcessState == nil { - return true - } - return false + return true } // GetResult returns the last log line or the exit error from the mtg process. @@ -147,8 +148,11 @@ func (p *Process) GetResult() string { if line := p.logWriter.LastLine(); line != "" { return line } - if p.exitErr != nil { - return p.exitErr.Error() + p.mu.RLock() + exitErr := p.exitErr + p.mu.RUnlock() + if exitErr != nil { + return exitErr.Error() } return "" } @@ -161,22 +165,27 @@ func (p *Process) Start() error { cmd := exec.CommandContext(context.Background(), GetBinaryPath(), "run", p.configPath) cmd.Stdout = p.logWriter cmd.Stderr = p.logWriter + done := make(chan struct{}) + p.mu.Lock() p.cmd = cmd - p.done = make(chan struct{}) + p.done = done p.exitErr = nil + p.mu.Unlock() p.intentionalStop.Store(false) if err := cmd.Start(); err != nil { - close(p.done) + close(done) + p.mu.Lock() p.cmd = nil + p.mu.Unlock() return err } attachChildLifetime(cmd) - go p.wait(cmd) + go p.wait(cmd, done) return nil } -func (p *Process) wait(cmd *exec.Cmd) { - defer close(p.done) +func (p *Process) wait(cmd *exec.Cmd, done chan struct{}) { + defer close(done) err := cmd.Wait() p.logWriter.Flush() if err == nil || p.intentionalStop.Load() { @@ -184,12 +193,18 @@ func (p *Process) wait(cmd *exec.Cmd) { } if runtime.GOOS == "windows" { if strings.Contains(strings.ToLower(err.Error()), "exit status 1") { - p.exitErr = err + p.setExitErr(err) return } } logger.Errorf("mtproto: mtg process exited: %v", err) + p.setExitErr(err) +} + +func (p *Process) setExitErr(err error) { + p.mu.Lock() p.exitErr = err + p.mu.Unlock() } // Stop terminates the running mtg process gracefully, falling back to a kill. @@ -198,40 +213,46 @@ func (p *Process) Stop() error { return errors.New("mtg is not running") } p.intentionalStop.Store(true) - - if runtime.GOOS == "windows" { - if err := p.cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { - return err - } - return p.waitForExit(forceStopTimeout) + p.mu.RLock() + cmd, done := p.cmd, p.done + p.mu.RUnlock() + if cmd == nil || cmd.Process == nil { + return errors.New("mtg is not running") } - if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { + if runtime.GOOS == "windows" { + if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + return err + } + return waitForExit(done, forceStopTimeout) + } + + if err := cmd.Process.Signal(syscall.SIGTERM); err != nil { if errors.Is(err, os.ErrProcessDone) { - return p.waitForExit(forceStopTimeout) + return waitForExit(done, forceStopTimeout) } return err } - if err := p.waitForExit(gracefulStopTimeout); err == nil { + if err := waitForExit(done, gracefulStopTimeout); err == nil { return nil } logger.Warning("mtproto: mtg did not stop after SIGTERM, killing process") - if err := p.cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { + if err := cmd.Process.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) { return err } - return p.waitForExit(forceStopTimeout) + return waitForExit(done, forceStopTimeout) } -func (p *Process) waitForExit(timeout time.Duration) error { - if p.done == nil { +func waitForExit(done <-chan struct{}, timeout time.Duration) error { + if done == nil { return nil } timer := time.NewTimer(timeout) defer timer.Stop() select { - case <-p.done: + case <-done: return nil case <-timer.C: return fmt.Errorf("timed out waiting for mtg process to stop after %s", timeout) diff --git a/internal/mtproto/process_race_test.go b/internal/mtproto/process_race_test.go new file mode 100644 index 000000000..2cd06d978 --- /dev/null +++ b/internal/mtproto/process_race_test.go @@ -0,0 +1,106 @@ +package mtproto + +import ( + "errors" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func TestProcessLifecycleFieldsRaceSafe(t *testing.T) { + proc := newProcess("", "test") + stop := make(chan struct{}) + var workers sync.WaitGroup + defer func() { + close(stop) + workers.Wait() + }() + + workers.Go(func() { + for { + select { + case <-stop: + return + default: + } + proc.mu.Lock() + proc.cmd = &exec.Cmd{} + proc.done = make(chan struct{}) + proc.mu.Unlock() + proc.setExitErr(errors.New("exit")) + } + }) + for range 4 { + workers.Go(func() { + for { + select { + case <-stop: + return + default: + } + _ = proc.IsRunning() + _ = proc.GetResult() + } + }) + } + + time.Sleep(50 * time.Millisecond) +} + +func TestProcessStatusDuringExit(t *testing.T) { + pidFile := installFakeMtg(t) + exitFile := filepath.Join(t.TempDir(), "exit") + t.Setenv("MTG_FAKE_EXIT_FILE", exitFile) + configPath := filepath.Join(t.TempDir(), "mtg.toml") + if err := os.WriteFile(configPath, nil, 0o600); err != nil { + t.Fatalf("write config: %v", err) + } + proc := newProcess(configPath, "test") + if err := proc.Start(); err != nil { + t.Fatalf("start process: %v", err) + } + t.Cleanup(func() { + _ = proc.Stop() + }) + waitSpawnCount(t, pidFile, 1) + + stopReads := make(chan struct{}) + var readers sync.WaitGroup + defer func() { + close(stopReads) + readers.Wait() + }() + for range 4 { + readers.Go(func() { + for { + select { + case <-stopReads: + return + default: + _ = proc.IsRunning() + _ = proc.GetResult() + } + } + }) + } + + proc.mu.RLock() + done := proc.done + proc.mu.RUnlock() + if err := os.WriteFile(exitFile, nil, 0o600); err != nil { + t.Fatalf("trigger exit: %v", err) + } + if err := waitForExit(done, time.Second); err != nil { + t.Fatalf("wait for process exit: %v", err) + } + if proc.IsRunning() { + t.Fatal("process must not be running after exit") + } + if got := proc.GetResult(); !strings.Contains(got, "exit status 1") { + t.Fatalf("GetResult after an unexpected exit = %q, want exit status", got) + } +}