test/e2e: replace RunProcesses client sleep with log-based proxy readiness detection (#5226)

* test/e2e: replace RunProcesses client sleep with log-based proxy readiness detection

Replace the fixed 1500ms sleep in RunProcesses with event-driven proxy
registration detection by monitoring frpc log output for "start proxy
success" messages.

Key changes:
- Add thread-safe SafeBuffer to replace bytes.Buffer in Process, enabling
  concurrent read/write of process output during execution
- Add Process.WaitForOutput() to poll process output for pattern matches
  with timeout and early exit on process termination
- Add waitForClientProxyReady() that uses config.LoadClientConfig() to
  extract proxy names, then waits for each proxy's success log
- For visitor-only clients (no deterministic readiness signal), fall back
  to the original sleep with elapsed time deducted

* test/e2e: use shared deadline for proxy readiness and fix doc comment

- Use a single deadline in waitForClientProxyReady so total wait across
  all proxies does not exceed the given timeout
- Fix WaitForOutput doc comment to accurately describe single pattern
  with count semantics

test/e2e/pkg/process/process.go

@@ -4,15 +4,37 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"os/exec"
+	"strings"
 	"sync"
+	"time"
 )
 
+// SafeBuffer is a thread-safe wrapper around bytes.Buffer.
+// It is safe to call Write and String concurrently.
+type SafeBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *SafeBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *SafeBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
 type Process struct {
 	cmd         *exec.Cmd
 	cancel      context.CancelFunc
-	errorOutput *bytes.Buffer
-	stdOutput   *bytes.Buffer
+	errorOutput *SafeBuffer
+	stdOutput   *SafeBuffer
 
 	done     chan struct{}
 	closeOne sync.Once
@@ -36,8 +58,8 @@ func NewWithEnvs(path string, params []string, envs []string) *Process {
 		cancel: cancel,
 		done:   make(chan struct{}),
 	}
-	p.errorOutput = bytes.NewBufferString("")
-	p.stdOutput = bytes.NewBufferString("")
+	p.errorOutput = &SafeBuffer{}
+	p.stdOutput = &SafeBuffer{}
 	cmd.Stderr = p.errorOutput
 	cmd.Stdout = p.stdOutput
 	return p
@@ -101,3 +123,26 @@ func (p *Process) Output() string {
 func (p *Process) SetBeforeStopHandler(fn func()) {
 	p.beforeStopHandler = fn
 }
+
+// WaitForOutput polls the combined process output until the pattern is found
+// count time(s) or the timeout is reached. It also returns early if the process exits.
+func (p *Process) WaitForOutput(pattern string, count int, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		output := p.Output()
+		if strings.Count(output, pattern) >= count {
+			return nil
+		}
+		select {
+		case <-p.Done():
+			// Process exited, check one last time.
+			output = p.Output()
+			if strings.Count(output, pattern) >= count {
+				return nil
+			}
+			return fmt.Errorf("process exited before %d occurrence(s) of %q found", count, pattern)
+		case <-time.After(25 * time.Millisecond):
+		}
+	}
+	return fmt.Errorf("timeout waiting for %d occurrence(s) of %q", count, pattern)
+}