diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 6c7a9cb..879e786 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -43,5 +43,8 @@ jobs: - name: Run regular test suite run: TEST_INTEGRATION=true go test ./... -v -p 1 + - name: Check Xray process status races + run: go test -race ./backend/xray -run '^TestCoreStarted' -count=10 -timeout=120s + - name: Run wireguard integration tests run: sudo -E make test-integration-wireguard diff --git a/backend/xray/core.go b/backend/xray/core.go index 75627c0..bd7d6f0 100644 --- a/backend/xray/core.go +++ b/backend/xray/core.go @@ -114,13 +114,24 @@ func (c *Core) Version() string { } func (c *Core) Started() bool { - if c.process == nil || c.process.Process == nil { + c.mu.Lock() + defer c.mu.Unlock() + return c.startedLocked() +} + +// startedLocked requires c.mu. Wait closes waitDone without taking c.mu, so +// callers can observe process exit even while Stop holds the lock to reap it. +func (c *Core) startedLocked() bool { + if c.process == nil || c.process.Process == nil || c.waitDone == nil { return false } - if c.process.ProcessState == nil { + // exec.Cmd.Wait writes ProcessState outside c.mu; never read it here. + select { + case <-c.waitDone: + return false + default: return true } - return false } func (c *Core) Stopping() bool { @@ -197,7 +208,7 @@ func (c *Core) Start(xConfig *Config, debugMode bool) error { defer c.mu.Unlock() // Check if already started after acquiring lock to prevent race condition - if c.Started() { + if c.startedLocked() { return errors.New("xray is started already") } @@ -293,7 +304,7 @@ func (c *Core) Stop() { c.mu.Lock() defer c.mu.Unlock() - started := c.Started() + started := c.startedLocked() if !started && c.process == nil && c.cancelFunc == nil && c.logger == nil && len(c.unixSocketPaths) == 0 { return } diff --git a/backend/xray/core_started_test.go b/backend/xray/core_started_test.go new file mode 100644 index 0000000..984e6dc --- /dev/null +++ b/backend/xray/core_started_test.go @@ -0,0 +1,151 @@ +package xray + +import ( + "io" + "os" + "os/exec" + "runtime" + "sync" + "testing" + "time" +) + +// Run the test binary as a child that stays alive until its stdin is closed. +// This exercises exec.Cmd.Wait without requiring an installed Xray binary. +func TestCoreStartedHelperProcess(t *testing.T) { + if os.Getenv("PG_NODE_STARTED_TEST_HELPER") != "1" { + return + } + _, _ = os.Stdout.Write([]byte{1}) + _, _ = io.Copy(io.Discard, os.Stdin) + os.Exit(0) +} + +func startCoreTestProcess(t *testing.T, core *Core) (io.WriteCloser, <-chan struct{}) { + t.Helper() + cmd := exec.Command(os.Args[0], "-test.run=^TestCoreStartedHelperProcess$") + cmd.Env = append(os.Environ(), "PG_NODE_STARTED_TEST_HELPER=1") + stdin, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = stdin.Close() + t.Fatal(err) + } + if err := cmd.Start(); err != nil { + _ = stdin.Close() + _ = stdout.Close() + t.Fatal(err) + } + + waitDone := make(chan struct{}) + core.mu.Lock() + core.process = cmd + core.waitDone = waitDone + core.mu.Unlock() + go func() { + _ = cmd.Wait() + close(waitDone) + }() + t.Cleanup(func() { + _ = stdin.Close() + _ = cmd.Process.Kill() + waitForCoreTestProcess(t, waitDone) + }) + + if _, err := io.ReadFull(stdout, make([]byte, 1)); err != nil { + t.Fatalf("child did not become ready: %v", err) + } + return stdin, waitDone +} + +func waitForCoreTestProcess(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for child process") + } +} + +func pollCoreStarted(core *Core) func() { + done := make(chan struct{}) + var ready, workers sync.WaitGroup + for range 4 { + ready.Add(1) + workers.Add(1) + go func() { + defer workers.Done() + core.Started() + ready.Done() + for { + select { + case <-done: + return + default: + core.Started() + runtime.Gosched() + } + } + }() + } + ready.Wait() + return func() { + close(done) + workers.Wait() + } +} + +func TestCoreStartedProcessExit(t *testing.T) { + core := &Core{} + if core.Started() { + t.Fatal("new core reports running") + } + stdin, waitDone := startCoreTestProcess(t, core) + if !core.Started() { + t.Fatal("live child reports stopped") + } + stopPolling := pollCoreStarted(core) + defer stopPolling() + + // Process exit must be observable without the exit handler acquiring mu. + // Stop also holds mu while waiting for the process watcher to reap the child. + core.mu.Lock() + _ = stdin.Close() + select { + case <-waitDone: + case <-time.After(5 * time.Second): + core.mu.Unlock() + t.Fatal("process watcher blocked while core mutex was held") + } + started := core.startedLocked() + core.mu.Unlock() + if started || core.Started() { + t.Fatal("exited child reports running") + } +} + +func TestCoreStartedConcurrentStopAndReplacement(t *testing.T) { + core := &Core{} + stopPolling := pollCoreStarted(core) + defer stopPolling() + + for range 8 { + _, waitDone := startCoreTestProcess(t, core) + if !core.Started() { + t.Fatal("replacement child reports stopped") + } + // Start already owns mu here: calling the locking Started method from + // this path would deadlock instead of rejecting the duplicate start. + if err := core.Start(&Config{}, false); err == nil { + t.Fatal("duplicate start succeeded") + } + core.Stop() + waitForCoreTestProcess(t, waitDone) + if core.Started() { + t.Fatal("stopped child reports running") + } + } +}