core/auth/ui/e2e/process.go

package e2e

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"strings"
	"sync"
	"time"
)

type processSignal int

const (
	processTerminate processSignal = iota
	processKill
)

type supervisedProcess struct {
	name       string
	cmd        *exec.Cmd
	logPath    string
	logFile    *os.File
	pgid       int
	completed  chan struct{}
	resultMu   sync.RWMutex
	result     error
	stop       sync.Once
	stopErr    error
	redactions []string
}

func startProcess(name, logPath, dir, executable string, args ...string) (*supervisedProcess, error) {
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
	if err != nil {
		return nil, err
	}
	cmd := exec.Command(executable, args...)
	cmd.Dir = dir
	cmd.Stdout = logFile
	cmd.Stderr = logFile
	if err := configureOwnedProcess(cmd); err != nil {
		_ = logFile.Close()
		return nil, fmt.Errorf("configure %s process ownership: %w", name, err)
	}
	if err := cmd.Start(); err != nil {
		_ = logFile.Close()
		return nil, fmt.Errorf("start %s: %w", name, err)
	}
	p := &supervisedProcess{name: name, cmd: cmd, logPath: logPath, logFile: logFile, pgid: cmd.Process.Pid, completed: make(chan struct{})}
	go func() {
		err := cmd.Wait()
		_ = logFile.Close()
		p.resultMu.Lock()
		p.result = err
		p.resultMu.Unlock()
		close(p.completed)
	}()
	return p, nil
}

func (p *supervisedProcess) alive() bool {
	if p == nil {
		return false
	}
	select {
	case <-p.completed:
		return false
	default:
		return true
	}
}

func (p *supervisedProcess) completion() error {
	p.resultMu.RLock()
	defer p.resultMu.RUnlock()
	return p.result
}

func (p *supervisedProcess) stopAndWait(timeout time.Duration) error {
	if p == nil {
		return nil
	}
	p.stop.Do(func() {
		// Signal the group even when the leader has already exited: descendants may
		// still own the process group and are part of this harness's ownership.
		if err := signalOwnedProcessGroup(p.pgid, processTerminate); err != nil && !errors.Is(err, os.ErrProcessDone) {
			p.stopErr = fmt.Errorf("terminate %s process group: %w", p.name, err)
		}
		deadline := time.Now().Add(timeout)
		for ownedProcessGroupAlive(p.pgid) && time.Now().Before(deadline) {
			time.Sleep(10 * time.Millisecond)
		}
		if ownedProcessGroupAlive(p.pgid) {
			if err := signalOwnedProcessGroup(p.pgid, processKill); err != nil && !errors.Is(err, os.ErrProcessDone) && p.stopErr == nil {
				p.stopErr = fmt.Errorf("kill %s process group: %w", p.name, err)
			}
		}

		postKill := time.NewTimer(2 * time.Second)
		defer postKill.Stop()
		select {
		case <-p.completed:
		case <-postKill.C:
			if p.stopErr == nil {
				p.stopErr = fmt.Errorf("wait for %s leader after process-group kill: timeout", p.name)
			}
		}
		groupDeadline := time.Now().Add(2 * time.Second)
		for ownedProcessGroupAlive(p.pgid) && time.Now().Before(groupDeadline) {
			time.Sleep(10 * time.Millisecond)
		}
		if ownedProcessGroupAlive(p.pgid) && p.stopErr == nil {
			p.stopErr = fmt.Errorf("%s process group %d survived cleanup", p.name, p.pgid)
		}
		// A non-zero leader exit belongs to startup/readiness diagnostics, not
		// cleanup. Once the owned group is gone, explicit cleanup succeeded.
	})
	return p.stopErr
}

func (p *supervisedProcess) exited() (bool, error) {
	select {
	case <-p.completed:
		return true, p.completion()
	default:
		return false, nil
	}
}

func (p *supervisedProcess) logTail(limit int64) string {
	data, err := os.ReadFile(p.logPath)
	if err != nil {
		return fmt.Sprintf("<read log: %v>", err)
	}
	if int64(len(data)) > limit {
		data = data[len(data)-int(limit):]
	}
	text := boundedText(data, int(limit))
	for _, secret := range p.redactions {
		if secret != "" {
			text = strings.ReplaceAll(text, secret, "[REDACTED]")
		}
	}
	return text
}

func waitHTTPReady(ctx context.Context, p *supervisedProcess, endpoints ...string) error {
	client := &http.Client{Timeout: 750 * time.Millisecond}
	defer client.CloseIdleConnections()
	for {
		allReady := true
		for _, endpoint := range endpoints {
			req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
			if err != nil {
				return err
			}
			resp, err := client.Do(req)
			if err != nil {
				allReady = false
				break
			}
			_, _ = io.CopyN(io.Discard, resp.Body, 4096)
			_ = resp.Body.Close()
			if resp.StatusCode < 200 || resp.StatusCode >= 300 {
				allReady = false
				break
			}
		}
		if allReady {
			return nil
		}
		if exited, err := p.exited(); exited {
			return fmt.Errorf("%s exited during readiness: %v\nlog tail:\n%s", p.name, err, p.logTail(8192))
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("%s readiness: %w\nlog tail:\n%s", p.name, ctx.Err(), p.logTail(8192))
		case <-time.After(100 * time.Millisecond):
		}
	}
}

func waitTCPReady(ctx context.Context, p *supervisedProcess, addresses ...string) error {
	pending := append([]string(nil), addresses...)
	for len(pending) > 0 {
		next := pending[:0]
		for _, address := range pending {
			conn, err := (&net.Dialer{Timeout: 500 * time.Millisecond}).DialContext(ctx, "tcp", address)
			if err == nil {
				_ = conn.Close()
				continue
			}
			next = append(next, address)
		}
		pending = next
		if len(pending) == 0 {
			return nil
		}
		if exited, exitErr := p.exited(); exited {
			return fmt.Errorf("%s exited during readiness: %v\nlog tail:\n%s", p.name, exitErr, p.logTail(8192))
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("%s readiness for %s: %w\nlog tail:\n%s", p.name, strings.Join(pending, ", "), ctx.Err(), p.logTail(8192))
		case <-time.After(100 * time.Millisecond):
		}
	}
	return nil
}

func boundedText(data []byte, limit int) string {
	if len(data) > limit {
		data = data[len(data)-limit:]
	}
	return strings.ToValidUTF8(string(data), "�")
}