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 <[email protected]>

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())

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)
+	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 runtime.GOOS == "windows" {
-		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)
 	}
 
-	if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+	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)

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)
+	}
+}