Merge pull request #1 from Gladium-AI/fix/drain-cloudflared-stdout

Fix: drain cloudflared stdout/stderr to prevent tunnel error 1033

Author: paoloanzn

internal/exec/runner.go

@@ -1,32 +1,93 @@
 package exec
 
 import (
-	"context"
+	"bytes"
 	"fmt"
 	"io"
 	"os"
 	"os/exec"
+	"sync"
 	"syscall"
 	"time"
 )
 
 // Runner manages subprocess execution.
 type Runner struct {
-	cmd     *exec.Cmd
-	logPipe io.ReadCloser
+	cmd *exec.Cmd
+
+	// exitCh is closed when the process exits.
+	exitCh chan struct{}
+	// exitErr holds the result of cmd.Wait().
+	exitErr error
+
+	mu     sync.Mutex
+	logBuf *ringBuffer // Circular buffer that stores recent log output.
+	logFile *os.File   // Optional persistent log file on disk.
+}
+
+// ringBuffer is a fixed-size circular buffer that implements io.Writer.
+// Older data is silently discarded when the buffer wraps.
+type ringBuffer struct {
+	mu   sync.Mutex
+	buf  []byte
+	pos  int  // next write position
+	full bool // whether the buffer has wrapped at least once
+}
+
+func newRingBuffer(size int) *ringBuffer {
+	return &ringBuffer{buf: make([]byte, size)}
+}
+
+func (rb *ringBuffer) Write(p []byte) (int, error) {
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
+	n := len(p)
+	for len(p) > 0 {
+		k := copy(rb.buf[rb.pos:], p)
+		rb.pos += k
+		p = p[k:]
+		if rb.pos == len(rb.buf) {
+			rb.pos = 0
+			rb.full = true
+		}
+	}
+	return n, nil
+}
+
+// Bytes returns the buffered data in chronological order.
+func (rb *ringBuffer) Bytes() []byte {
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
+	if !rb.full {
+		return bytes.Clone(rb.buf[:rb.pos])
+	}
+	out := make([]byte, len(rb.buf))
+	n := copy(out, rb.buf[rb.pos:])
+	copy(out[n:], rb.buf[:rb.pos])
+	return out
 }
 
 // RunOpts configures a subprocess.
 type RunOpts struct {
-	Name string            // Binary name or path.
-	Args []string          // Command arguments.
-	Dir  string            // Working directory (optional).
-	Env  map[string]string // Additional environment variables.
+	Name    string            // Binary name or path.
+	Args    []string          // Command arguments.
+	Dir     string            // Working directory (optional).
+	Env     map[string]string // Additional environment variables.
+	LogFile string            // Optional path to persist subprocess logs to disk.
 }
 
 // Start launches a subprocess with its own process group.
-func Start(ctx context.Context, opts RunOpts) (*Runner, error) {
-	cmd := exec.CommandContext(ctx, opts.Name, opts.Args...)
+//
+// IMPORTANT: We intentionally use exec.Command (NOT exec.CommandContext).
+// exec.CommandContext sends SIGKILL to the subprocess when the context is
+// cancelled, which kills cloudflared instantly without any chance for graceful
+// shutdown or log capture. Instead, we manage the process lifecycle ourselves
+// via the Stop() method which sends SIGTERM first, then SIGKILL after a timeout.
+//
+// Stdout/stderr are continuously drained into a 1 MB ring buffer (and optionally
+// to a persistent log file on disk) so the subprocess never blocks on log output.
+func Start(_ /* ctx not used intentionally */ interface{}, opts RunOpts) (*Runner, error) {
+	cmd := exec.Command(opts.Name, opts.Args...)
 
 	if opts.Dir != "" {
 		cmd.Dir = opts.Dir
@@ -41,24 +102,79 @@ func Start(ctx context.Context, opts RunOpts) (*Runner, error) {
 	// Own process group so we can kill the tree.
 	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
 
-	// Capture combined output.
-	pr, pw := io.Pipe()
-	cmd.Stdout = pw
-	cmd.Stderr = pw
+	// Use OS pipes (buffered by kernel, typically 64KB+) instead of io.Pipe
+	// (which has zero buffering and blocks the writer immediately).
+	stdoutR, stdoutW, err := os.Pipe()
+	if err != nil {
+		return nil, fmt.Errorf("creating stdout pipe: %w", err)
+	}
+	stderrR, stderrW, err := os.Pipe()
+	if err != nil {
+		stdoutR.Close()
+		stdoutW.Close()
+		return nil, fmt.Errorf("creating stderr pipe: %w", err)
+	}
+
+	cmd.Stdout = stdoutW
+	cmd.Stderr = stderrW
 
 	if err := cmd.Start(); err != nil {
-		pw.Close()
-		pr.Close()
+		stdoutR.Close()
+		stdoutW.Close()
+		stderrR.Close()
+		stderrW.Close()
 		return nil, fmt.Errorf("starting %s: %w", opts.Name, err)
 	}
 
-	// Close the write end when the process exits so readers see EOF.
+	// Close the write ends in our process — the child owns them now.
+	stdoutW.Close()
+	stderrW.Close()
+
+	r := &Runner{
+		cmd:    cmd,
+		logBuf: newRingBuffer(1 << 20), // 1 MB ring buffer
+		exitCh: make(chan struct{}),
+	}
+
+	// Open persistent log file if requested.
+	var logFileWriter io.Writer
+	if opts.LogFile != "" {
+		f, err := os.OpenFile(opts.LogFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
+		if err != nil {
+			// Non-fatal: log to ring buffer only.
+			logFileWriter = nil
+		} else {
+			r.logFile = f
+			logFileWriter = f
+		}
+	}
+
+	// Build the combined writer: ring buffer + optional log file.
+	var sink io.Writer
+	if logFileWriter != nil {
+		sink = io.MultiWriter(r.logBuf, logFileWriter)
+	} else {
+		sink = r.logBuf
+	}
+
+	// Continuously drain stdout and stderr into the sink.
+	// This runs until the read ends return EOF (i.e., the process exits).
+	var drainWg sync.WaitGroup
+	drainWg.Add(2)
+	go func() { defer drainWg.Done(); io.Copy(sink, stdoutR); stdoutR.Close() }()
+	go func() { defer drainWg.Done(); io.Copy(sink, stderrR); stderrR.Close() }()
+
+	// Wait for the process to exit, then close exitCh to notify watchers.
 	go func() {
-		cmd.Wait()
-		pw.Close()
+		r.exitErr = cmd.Wait()
+		drainWg.Wait() // ensure all output is captured before signalling
+		if r.logFile != nil {
+			r.logFile.Close()
+		}
+		close(r.exitCh)
 	}()
 
-	return &Runner{cmd: cmd, logPipe: pr}, nil
+	return r, nil
 }
 
 // PID returns the process ID.
@@ -70,8 +186,34 @@ func (r *Runner) PID() int {
 }
 
 // Logs returns a reader for the process's combined stdout/stderr.
+// The returned reader contains a snapshot of the most recent log output
+// (up to 1 MB) from the ring buffer.
 func (r *Runner) Logs() io.ReadCloser {
-	return r.logPipe
+	return io.NopCloser(bytes.NewReader(r.logBuf.Bytes()))
+}
+
+// LogFilePath returns the path to the persistent log file, or "" if none.
+func (r *Runner) LogFilePath() string {
+	if r.logFile != nil {
+		return r.logFile.Name()
+	}
+	return ""
+}
+
+// ExitCh returns a channel that is closed when the process exits.
+func (r *Runner) ExitCh() <-chan struct{} {
+	return r.exitCh
+}
+
+// ExitError returns the exit error after the process has finished.
+// Returns nil if the process hasn't exited yet or exited cleanly.
+func (r *Runner) ExitError() error {
+	select {
+	case <-r.exitCh:
+		return r.exitErr
+	default:
+		return nil
+	}
 }
 
 // Stop sends SIGTERM to the process group, waits up to 10s, then SIGKILL.
@@ -80,47 +222,50 @@ func (r *Runner) Stop() error {
 		return nil
 	}
 
+	// If the process already exited, nothing to do.
+	select {
+	case <-r.exitCh:
+		return nil
+	default:
+	}
+
 	// Send SIGTERM to the process group.
 	pgid := -r.cmd.Process.Pid
 	if err := syscall.Kill(pgid, syscall.SIGTERM); err != nil {
 		// Process may have already exited.
 		return nil
 	}
 
-	// Wait for exit with timeout.
-	done := make(chan error, 1)
-	go func() {
-		done <- r.cmd.Wait()
-	}()
-
 	select {
-	case <-done:
+	case <-r.exitCh:
 		return nil
 	case <-time.After(10 * time.Second):
 		// Force kill.
 		_ = syscall.Kill(pgid, syscall.SIGKILL)
-		<-done // Wait for the process to be reaped.
+		<-r.exitCh
 		return fmt.Errorf("process did not exit after SIGTERM; sent SIGKILL")
 	}
 }
 
 // Wait blocks until the process exits and returns its error.
 func (r *Runner) Wait() error {
-	return r.cmd.Wait()
+	<-r.exitCh
+	return r.exitErr
 }
 
 // Running returns true if the process is still alive.
 func (r *Runner) Running() bool {
-	if r.cmd.Process == nil {
+	select {
+	case <-r.exitCh:
 		return false
+	default:
+		return r.cmd.Process != nil
 	}
-	// Signal 0 checks if the process exists.
-	return r.cmd.Process.Signal(syscall.Signal(0)) == nil
 }
 
 // Run executes a command synchronously and returns its combined output.
-func Run(ctx context.Context, name string, args ...string) (string, error) {
-	cmd := exec.CommandContext(ctx, name, args...)
+func Run(_ interface{}, name string, args ...string) (string, error) {
+	cmd := exec.Command(name, args...)
 	out, err := cmd.CombinedOutput()
 	return string(out), err
 }

internal/pipeline/serve.go

@@ -3,6 +3,8 @@ package pipeline
 import (
 	"context"
 	"fmt"
+	"io"
+	"os"
 	"strings"
 	"time"
 
@@ -232,6 +234,9 @@ func (p *Pipeline) Serve(ctx context.Context, params ServeParams) error {
 		p.connector.Stop(cleanupCtx)
 	})
 	ui.PrintSuccess("cloudflared running (PID %d)", pid)
+	if logPath := p.connector.LogFilePath(); logPath != "" {
+		ui.PrintInfo("cloudflared logs: %s", logPath)
+	}
 
 	// Step 8: Print success summary.
 	fmt.Println()
@@ -257,11 +262,37 @@ func (p *Pipeline) waitForShutdown(ctx context.Context, sess *session.Session, p
 		ttlTimer = time.After(params.TTL)
 	}
 
+	// Get cloudflared's exit channel so we detect crashes immediately.
+	exitCh := p.connector.ExitCh()
+
 	select {
 	case <-ctx.Done():
 		ui.PrintInfo("Shutdown signal received...")
 	case <-ttlTimer:
 		ui.PrintInfo("TTL expired, shutting down...")
+	case <-exitCh:
+		ui.PrintWarning("cloudflared process exited unexpectedly!")
+		// Dump last 50 lines of cloudflared logs for diagnostics.
+		if logs := p.connector.Logs(); logs != nil {
+			data, _ := io.ReadAll(logs)
+			logs.Close()
+			if len(data) > 0 {
+				ui.PrintWarning("Last cloudflared output:")
+				// Print last ~50 lines.
+				lines := strings.Split(strings.TrimSpace(string(data)), "\n")
+				start := 0
+				if len(lines) > 50 {
+					start = len(lines) - 50
+				}
+				for _, line := range lines[start:] {
+					fmt.Fprintf(os.Stderr, "  %s\n", line)
+				}
+			}
+		}
+		if logPath := p.connector.LogFilePath(); logPath != "" {
+			ui.PrintInfo("Full cloudflared logs at: %s", logPath)
+		}
+		ui.PrintInfo("Shutting down session...")
 	}
 
 	return p.Teardown(sess, cleanups)

internal/testutil/mocks.go

@@ -85,6 +85,7 @@ type MockConnector struct {
 	FnStop    func(ctx context.Context) error
 	FnLogs    func() io.ReadCloser
 	FnHealthy func() error
+	FnExitCh  func() <-chan struct{}
 
 	RunCalls   []string
 	StopCalled bool
@@ -126,6 +127,18 @@ func (m *MockConnector) Healthy() error {
 	return nil
 }
 
+func (m *MockConnector) ExitCh() <-chan struct{} {
+	if m.FnExitCh != nil {
+		return m.FnExitCh()
+	}
+	// Default: return a channel that never closes (process stays alive).
+	return make(chan struct{})
+}
+
+func (m *MockConnector) LogFilePath() string {
+	return ""
+}
+
 // --- MockAccessManager ---
 
 type MockAccessManager struct {