magefile.go

//go:build mage

// Celeris Benchmark Suite build tasks.
// Install mage: go install github.com/magefile/mage@latest
// Run: mage [target]
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/magefile/mage/mg"
	"github.com/magefile/mage/sh"
)

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

const (
	binDir       = "bin"
	hostsFile    = "config/hosts.json"
	goModFile    = "go.mod"
	buildLdflags = "-s -w"

	// Remote operation defaults
	serverPort     = "8080"
	controlPort    = 9999
	healthPath     = "/health"
	shutdownPath   = "/shutdown"
	healthTimeout  = 60 * time.Second
	healthInterval = 2 * time.Second
	serverLogPath  = "/tmp/server.log"
	benchUser      = "benchmark"
	remoteBinDir   = "/home/benchmark/bin"

	// Ulimit / sysctl constants
	ulimitNofile    = 2097152
	ulimitConf      = "/etc/security/limits.d/99-benchmark.conf"
	sysctlConf      = "/etc/sysctl.d/99-benchmark.conf"
	goProfileScript = "/etc/profile.d/go.sh"
)

// sysctlConfig is the kernel tuning written to /etc/sysctl.d/99-benchmark.conf.
const sysctlConfig = `# Celeris benchmark kernel tuning
net.core.somaxconn = 65535
net.core.netdev_max_backlog = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 10
net.ipv4.ip_local_port_range = 1024 65535
net.ipv4.tcp_fastopen = 3
net.ipv4.tcp_mtu_probing = 1
fs.file-max = 2097152
fs.nr_open = 2097152
vm.swappiness = 1
net.core.rmem_max = 16777216
net.core.wmem_max = 16777216
net.ipv4.tcp_rmem = 4096 87380 16777216
net.ipv4.tcp_wmem = 4096 87380 16777216`

// irqAffinityScript is run via sudo to pin NIC IRQs to dedicated CPU cores.
const irqAffinityScript = `
set -euo pipefail
IFACE=""
for dev in /sys/class/net/*; do
  name=$(basename "$dev")
  [ "$name" = "lo" ] && continue
  [ "$(cat "$dev/carrier" 2>/dev/null)" = "1" ] || continue
  IFACE="$name"
  break
done
if [ -z "$IFACE" ]; then
  echo "[tune] No active network interface found, skipping IRQ affinity"
  exit 0
fi
IRQS=$(grep "$IFACE" /proc/interrupts 2>/dev/null | awk '{print $1}' | tr -d ':')
if [ -z "$IRQS" ]; then
  echo "[tune] No IRQs found for $IFACE, skipping affinity pinning"
  exit 0
fi
CORE=0
NCPUS=$(nproc)
echo "[tune] Pinning $IFACE IRQs to dedicated CPU cores..."
for irq in $IRQS; do
  mask=$(printf "%x" $((1 << CORE)))
  echo "$mask" > "/proc/irq/$irq/smp_affinity" 2>/dev/null || true
  CORE=$(( (CORE + 1) % NCPUS ))
done
echo "[tune] IRQ affinity pinned for $IFACE"
`

// ---------------------------------------------------------------------------
// Color helpers
// ---------------------------------------------------------------------------

var (
	green = "\033[0;32m"
	nc    = "\033[0m" // No Color
)

func printGreen(format string, args ...any) {
	fmt.Printf("%s%s%s\n", green, fmt.Sprintf(format, args...), nc)
}

// ---------------------------------------------------------------------------
// Data model
// ---------------------------------------------------------------------------

// Machine describes a single benchmark host.
type Machine struct {
	Name        string `json:"-"` // populated from the map key
	Host        string `json:"host"`
	User        string `json:"user"`
	Arch        string `json:"arch"`
	CPU         string `json:"cpu"`
	Cores       int    `json:"cores"`
	Threads     int    `json:"threads"`
	RAMGB       int    `json:"ram_gb"`
	Network     string `json:"network"`
	Role        string `json:"role"`
	Description string `json:"description"`
}

// HostConfig is the top-level structure of config/hosts.json.
type HostConfig struct {
	Machines map[string]Machine `json:"machines"`
}

var (
	hostCfg     HostConfig
	hostOnce    sync.Once
	hostLoadErr error
)

// loadHosts reads and caches config/hosts.json. It populates each Machine's
// Name field from the map key.
func loadHosts() (HostConfig, error) {
	hostOnce.Do(func() {
		data, err := os.ReadFile(hostsFile)
		if err != nil {
			hostLoadErr = fmt.Errorf("reading %s: %w", hostsFile, err)
			return
		}
		if err := json.Unmarshal(data, &hostCfg); err != nil {
			hostLoadErr = fmt.Errorf("parsing %s: %w", hostsFile, err)
			return
		}
		for k, m := range hostCfg.Machines {
			m.Name = k
			hostCfg.Machines[k] = m
		}
	})
	return hostCfg, hostLoadErr
}

// parseGoVersion reads go.mod and returns the Go version string (e.g. "1.26.0").
func parseGoVersion() (string, error) {
	f, err := os.Open(goModFile)
	if err != nil {
		return "", fmt.Errorf("opening %s: %w", goModFile, err)
	}
	defer f.Close()

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		parts := strings.Fields(line)
		if len(parts) >= 2 && parts[0] == "go" {
			return parts[1], nil
		}
	}
	if err := scanner.Err(); err != nil {
		return "", fmt.Errorf("scanning %s: %w", goModFile, err)
	}
	return "", fmt.Errorf("no 'go' directive found in %s", goModFile)
}

// parseCelerisVersion reads go.mod and returns the version of the
// github.com/goceleris/celeris dependency (e.g. "v1.2.0").
func parseCelerisVersion() (string, error) {
	f, err := os.Open(goModFile)
	if err != nil {
		return "", fmt.Errorf("opening %s: %w", goModFile, err)
	}
	defer f.Close()

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		parts := strings.Fields(line)
		if len(parts) >= 2 && parts[0] == "github.com/goceleris/celeris" {
			return parts[1], nil
		}
	}
	if err := scanner.Err(); err != nil {
		return "", fmt.Errorf("scanning %s: %w", goModFile, err)
	}
	return "", fmt.Errorf("no celeris dependency found in %s", goModFile)
}

// ---------------------------------------------------------------------------
// Filtering helper
// ---------------------------------------------------------------------------

// filterMachines returns machines with the given role. If filter is non-empty
// only machines whose name key contains the filter substring are returned.
func filterMachines(cfg HostConfig, role, filter string) []Machine {
	var out []Machine
	for _, m := range cfg.Machines {
		if m.Role != role {
			continue
		}
		if filter != "" && !strings.Contains(m.Name, filter) {
			continue
		}
		out = append(out, m)
	}
	return out
}

// ---------------------------------------------------------------------------
// Architecture helper
// ---------------------------------------------------------------------------

// goArch maps a hosts.json arch value to a GOARCH value.
func goArch(arch string) string {
	switch arch {
	case "x86":
		return "amd64"
	default:
		return arch // "arm64" passes through
	}
}

// ---------------------------------------------------------------------------
// SSH / SCP helpers (private — NOT mage targets)
// ---------------------------------------------------------------------------

// sshKeyFile returns the SSH private key path from BENCH_SSH_KEY_FILE or the
// default ~/.ssh/bench_key.
func sshKeyFile() string {
	if v := os.Getenv("BENCH_SSH_KEY_FILE"); v != "" {
		return v
	}
	home, err := os.UserHomeDir()
	if err != nil {
		return filepath.Join("~", ".ssh", "bench_key")
	}
	return filepath.Join(home, ".ssh", "bench_key")
}

// sshArgs returns the standard SSH flags for a machine.
func sshArgs(m Machine) []string {
	return []string{
		"-o", "StrictHostKeyChecking=no",
		"-o", "BatchMode=yes",
		"-o", "UserKnownHostsFile=/dev/null",
		"-o", "ConnectTimeout=10",
		"-i", sshKeyFile(),
	}
}

// sshRun executes a command on the remote machine via system ssh, forwarding
// stdout and stderr to the local terminal.
func sshRun(m Machine, cmd string) error {
	args := sshArgs(m)
	args = append(args, fmt.Sprintf("%s@%s", m.User, m.Host), cmd)
	c := exec.Command("ssh", args...)
	c.Stdout = os.Stdout
	c.Stderr = os.Stderr
	if err := c.Run(); err != nil {
		return fmt.Errorf("ssh %s@%s: %w", m.User, m.Host, err)
	}
	return nil
}

// sshRunSudo executes a command on the remote machine with sudo.
func sshRunSudo(m Machine, cmd string) error {
	return sshRun(m, "sudo bash -c "+shellQuote(cmd))
}

// scpTo copies a local file to a remote path.
func scpTo(m Machine, local, remote string) error {
	args := sshArgs(m)
	args = append(args, local, fmt.Sprintf("%s@%s:%s", m.User, m.Host, remote))
	c := exec.Command("scp", args...)
	c.Stdout = os.Stdout
	c.Stderr = os.Stderr
	return c.Run()
}

// shellQuote wraps s in single quotes, escaping embedded single quotes.
func shellQuote(s string) string {
	return "'" + strings.ReplaceAll(s, "'", "'\"'\"'") + "'"
}

// ---------------------------------------------------------------------------
// Health check helper
// ---------------------------------------------------------------------------

// waitForHealth polls GET http://host:port/health until it returns 200 or the
// timeout elapses.
func waitForHealth(host string, port int, timeout time.Duration) error {
	url := fmt.Sprintf("http://%s:%d%s", host, port, healthPath)
	deadline := time.Now().Add(timeout)
	client := &http.Client{Timeout: healthInterval}

	for time.Now().Before(deadline) {
		resp, err := client.Get(url)
		if err == nil {
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				return nil
			}
		}
		time.Sleep(healthInterval)
	}
	return fmt.Errorf("health check timed out after %s for %s", timeout, url)
}

// ---------------------------------------------------------------------------
// Default target
// ---------------------------------------------------------------------------

// Default target when running mage without arguments.
var Default = Build

// ---------------------------------------------------------------------------
// Build targets
// ---------------------------------------------------------------------------

// Build builds all binaries (server, bench).
func Build() error {
	mg.Deps(BuildServer, BuildBench)
	return nil
}

// BuildServer builds the server binary.
func BuildServer() error {
	printGreen("Building server...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	if err := sh.Run("go", "build", "-o", filepath.Join(binDir, "server"), "./cmd/server"); err != nil {
		return err
	}
	printGreen("Server build complete: %s/server", binDir)
	return nil
}

// BuildBench builds the benchmark tool.
func BuildBench() error {
	printGreen("Building benchmark tool...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	if err := sh.Run("go", "build", "-o", filepath.Join(binDir, "bench"), "./cmd/bench"); err != nil {
		return err
	}
	printGreen("Benchmark tool build complete: %s/bench", binDir)
	return nil
}

// BuildLinux cross-compiles all binaries for Linux amd64.
func BuildLinux() error {
	printGreen("Building for Linux amd64...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	env := map[string]string{"GOOS": "linux", "GOARCH": "amd64"}
	for _, cmd := range []struct{ name, path string }{
		{"server", "./cmd/server"},
		{"bench", "./cmd/bench"},
	} {
		if err := sh.RunWith(env, "go", "build", "-ldflags", buildLdflags, "-o", filepath.Join(binDir, cmd.name+"-linux-amd64"), cmd.path); err != nil {
			return err
		}
	}
	printGreen("Linux amd64 build complete")
	return nil
}

// BuildLinuxArm cross-compiles all binaries for Linux arm64.
func BuildLinuxArm() error {
	printGreen("Building for Linux arm64...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	env := map[string]string{"GOOS": "linux", "GOARCH": "arm64"}
	for _, cmd := range []struct{ name, path string }{
		{"server", "./cmd/server"},
		{"bench", "./cmd/bench"},
	} {
		if err := sh.RunWith(env, "go", "build", "-ldflags", buildLdflags, "-o", filepath.Join(binDir, cmd.name+"-linux-arm64"), cmd.path); err != nil {
			return err
		}
	}
	printGreen("Linux arm64 build complete")
	return nil
}

// BuildLinuxServer cross-compiles only the server binary for Linux amd64.
func BuildLinuxServer() error {
	printGreen("Building server for Linux amd64...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	env := map[string]string{"GOOS": "linux", "GOARCH": "amd64"}
	if err := sh.RunWith(env, "go", "build", "-ldflags", buildLdflags, "-o", filepath.Join(binDir, "server-linux-amd64"), "./cmd/server"); err != nil {
		return err
	}
	printGreen("Linux amd64 server build complete")
	return nil
}

// BuildLinuxArmServer cross-compiles only the server binary for Linux arm64.
func BuildLinuxArmServer() error {
	printGreen("Building server for Linux arm64...")
	if err := os.MkdirAll(binDir, 0755); err != nil {
		return err
	}
	env := map[string]string{"GOOS": "linux", "GOARCH": "arm64"}
	if err := sh.RunWith(env, "go", "build", "-ldflags", buildLdflags, "-o", filepath.Join(binDir, "server-linux-arm64"), "./cmd/server"); err != nil {
		return err
	}
	printGreen("Linux arm64 server build complete")
	return nil
}

// BuildAll cross-compiles for all supported platforms.
func BuildAll() error {
	mg.Deps(BuildLinux, BuildLinuxArm)
	return nil
}

// ---------------------------------------------------------------------------
// Code quality targets
// ---------------------------------------------------------------------------

// Lint runs golangci-lint.
func Lint() error {
	printGreen("Running golangci-lint...")
	if err := ensureGolangciLint(); err != nil {
		return err
	}
	if err := sh.Run("golangci-lint", "run", "--timeout=5m", "./..."); err != nil {
		return err
	}
	printGreen("Linting complete")
	return nil
}

// Fmt formats Go code.
func Fmt() error {
	printGreen("Formatting Go code...")
	if err := sh.Run("gofmt", "-s", "-w", "."); err != nil {
		return err
	}
	printGreen("Formatting complete")
	return nil
}

// Vet runs go vet.
func Vet() error {
	printGreen("Running go vet...")
	if err := sh.Run("go", "vet", "./..."); err != nil {
		return err
	}
	printGreen("Vet complete")
	return nil
}

// Test runs unit tests.
func Test() error {
	printGreen("Running tests...")
	if err := sh.Run("go", "test", "-v", "./..."); err != nil {
		return err
	}
	printGreen("Tests complete")
	return nil
}

// ---------------------------------------------------------------------------
// Benchmark targets
// ---------------------------------------------------------------------------

// Benchmark runs benchmarks with 30s duration.
func Benchmark() error {
	mg.Deps(Build)
	printGreen("Running benchmarks...")
	return sh.Run(filepath.Join(binDir, "bench"), "-mode", "baseline", "-duration", "30s")
}

// BenchmarkQuick runs a quick benchmark for validation (5s).
func BenchmarkQuick() error {
	mg.Deps(Build)
	printGreen("Running quick benchmark validation...")
	return sh.Run(filepath.Join(binDir, "bench"), "-mode", "baseline", "-duration", "5s")
}

// ---------------------------------------------------------------------------
// Dependency management
// ---------------------------------------------------------------------------

// Deps downloads Go dependencies.
func Deps() error {
	printGreen("Downloading dependencies...")
	if err := sh.Run("go", "mod", "download"); err != nil {
		return err
	}
	if err := sh.Run("go", "mod", "tidy"); err != nil {
		return err
	}
	printGreen("Dependencies ready")
	return nil
}

// ---------------------------------------------------------------------------
// Meta targets
// ---------------------------------------------------------------------------

// Check runs all checks (deps, lint, vet, build).
func Check() error {
	mg.SerialDeps(Deps, Lint, Vet, Build)
	printGreen("All checks passed")
	return nil
}

// Clean removes build artifacts.
func Clean() error {
	printGreen("Cleaning...")
	if err := os.RemoveAll(binDir); err != nil {
		return err
	}
	patterns := []string{"results/*.json", "results/*.png", "results/charts", "results/x86", "results/arm64"}
	for _, pattern := range patterns {
		matches, _ := filepath.Glob(pattern)
		for _, match := range matches {
			os.RemoveAll(match)
		}
	}
	printGreen("Clean complete")
	return nil
}

// ---------------------------------------------------------------------------
// Remote operation targets
// ---------------------------------------------------------------------------

// Setup provisions server machines: installs packages, Go, sets ulimits, and
// creates the benchmark user. Pass a machine name filter or empty string for
// all servers. Runs in parallel across machines.
func Setup(filter string) error {
	cfg, err := loadHosts()
	if err != nil {
		return err
	}

	goVer, err := parseGoVersion()
	if err != nil {
		return err
	}

	machines := filterMachines(cfg, "server", filter)
	if len(machines) == 0 {
		return fmt.Errorf("no server machines matched filter %q", filter)
	}

	type result struct {
		name string
		err  error
	}

	ch := make(chan result, len(machines))
	for _, m := range machines {
		go func(m Machine) {
			var err error
			defer func() {
				if r := recover(); r != nil {
					err = fmt.Errorf("panic: %v", r)
				}
				ch <- result{name: m.Name, err: err}
			}()
			err = setupMachine(m, goVer)
		}(m)
	}

	var errs []string
	for range machines {
		r := <-ch
		if r.err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", r.name, r.err))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("setup failed:\n  %s", strings.Join(errs, "\n  "))
	}

	printGreen("Setup complete for %d machine(s)", len(machines))
	return nil
}

// goVerRegexp validates Go version strings before shell interpolation.
var goVerRegexp = regexp.MustCompile(`^[0-9]+\.[0-9]+(\.[0-9]+)?$`)

// setupMachine runs the idempotent provisioning steps on a single machine.
func setupMachine(m Machine, goVer string) error {
	if !goVerRegexp.MatchString(goVer) {
		return fmt.Errorf("invalid Go version %q: must match %s", goVer, goVerRegexp.String())
	}

	printGreen("[%s] Installing packages...", m.Name)
	installPkgs := `
set -euo pipefail
NEEDED=()
for pkg in build-essential curl; do
  dpkg -s "$pkg" &>/dev/null || NEEDED+=("$pkg")
done
if [ ${#NEEDED[@]} -eq 0 ]; then
  echo "[setup] System packages already installed"
else
  echo "[setup] Installing packages: ${NEEDED[*]}"
  apt-get update -qq
  DEBIAN_FRONTEND=noninteractive apt-get install -y -qq "${NEEDED[@]}"
fi
`
	if err := sshRunSudo(m, installPkgs); err != nil {
		return fmt.Errorf("install packages: %w", err)
	}

	printGreen("[%s] Installing Go %s...", m.Name, goVer)
	installGo := fmt.Sprintf(`
set -euo pipefail
WANT="%s"
CURRENT=""
if command -v go &>/dev/null; then
  CURRENT=$(go version | awk '{print $3}' | sed 's/go//')
fi
if [ "$CURRENT" = "$WANT" ]; then
  echo "[setup] Go $WANT already installed"
  exit 0
fi
echo "[setup] Installing Go $WANT (current: ${CURRENT:-none})..."
ARCH=$(dpkg --print-architecture 2>/dev/null || uname -m)
case "$ARCH" in
  amd64|x86_64) ARCH="amd64" ;;
  arm64|aarch64) ARCH="arm64" ;;
  *) echo "[setup] ERROR: unsupported architecture: $ARCH"; exit 1 ;;
esac
TARBALL="go${WANT}.linux-${ARCH}.tar.gz"
URL="https://go.dev/dl/${TARBALL}"
cd /tmp
curl -fsSLO "$URL"
rm -rf /usr/local/go
tar -C /usr/local -xzf "$TARBALL"
rm -f "$TARBALL"
if ! grep -q '/usr/local/go/bin' %s 2>/dev/null; then
  echo 'export PATH=$PATH:/usr/local/go/bin' > %s
fi
export PATH=$PATH:/usr/local/go/bin
echo "[setup] Go $(go version | awk '{print $3}') installed"
`, goVer, goProfileScript, goProfileScript)
	if err := sshRunSudo(m, installGo); err != nil {
		return fmt.Errorf("install go: %w", err)
	}

	printGreen("[%s] Setting ulimits...", m.Name)
	setUlimits := fmt.Sprintf(`
set -euo pipefail
TARGET=%d
CONF="%s"
if [ -f "$CONF" ] && grep -q "$TARGET" "$CONF"; then
  echo "[setup] Ulimits already configured"
  exit 0
fi
echo "[setup] Setting ulimits (nofile=$TARGET)..."
cat > "$CONF" <<ULEOF
* soft nofile $TARGET
* hard nofile $TARGET
root soft nofile $TARGET
root hard nofile $TARGET
ULEOF
if [ "$(cat /proc/sys/fs/nr_open)" -lt "$TARGET" ]; then
  echo "$TARGET" > /proc/sys/fs/nr_open
fi
echo "[setup] Ulimits configured"
`, ulimitNofile, ulimitConf)
	if err := sshRunSudo(m, setUlimits); err != nil {
		return fmt.Errorf("set ulimits: %w", err)
	}

	printGreen("[%s] Ensuring benchmark user...", m.Name)
	ensureUser := fmt.Sprintf(`
set -euo pipefail
if id %s &>/dev/null; then
  echo "[setup] Benchmark user already exists"
  exit 0
fi
echo "[setup] Creating benchmark user..."
useradd -m -s /bin/bash %s
echo "[setup] Benchmark user created"
`, benchUser, benchUser)
	if err := sshRunSudo(m, ensureUser); err != nil {
		return fmt.Errorf("ensure user: %w", err)
	}

	printGreen("[%s] Setup complete", m.Name)
	return nil
}

// Tune applies kernel tuning to server machines: sysctl, CPU governor, turbo
// boost disable, IRQ affinity, and NUMA configuration. Pass a machine name
// filter or empty string for all servers.
func Tune(filter string) error {
	cfg, err := loadHosts()
	if err != nil {
		return err
	}

	machines := filterMachines(cfg, "server", filter)
	if len(machines) == 0 {
		return fmt.Errorf("no server machines matched filter %q", filter)
	}

	type result struct {
		name string
		err  error
	}

	ch := make(chan result, len(machines))
	for _, m := range machines {
		go func(m Machine) {
			var err error
			defer func() {
				if r := recover(); r != nil {
					err = fmt.Errorf("panic: %v", r)
				}
				ch <- result{name: m.Name, err: err}
			}()
			err = tuneMachine(m)
		}(m)
	}

	var errs []string
	for range machines {
		r := <-ch
		if r.err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", r.name, r.err))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("tune failed:\n  %s", strings.Join(errs, "\n  "))
	}

	printGreen("Tune complete for %d machine(s)", len(machines))
	return nil
}

// tuneMachine applies kernel tuning to a single machine.
func tuneMachine(m Machine) error {
	printGreen("[%s] Applying sysctl tuning...", m.Name)
	applySysctl := fmt.Sprintf(`
set -euo pipefail
CONF="%s"
DESIRED=$(cat <<'SYSEOF'
%s
SYSEOF
)
if [ -f "$CONF" ] && diff -q <(printf '%%s\n' "$DESIRED") "$CONF" &>/dev/null; then
  echo "[tune] Sysctl already configured"
else
  echo "[tune] Applying sysctl tuning..."
  printf '%%s\n' "$DESIRED" > "$CONF"
  sysctl --system >/dev/null 2>&1
  echo "[tune] Sysctl applied"
fi
`, sysctlConf, sysctlConfig)
	if err := sshRunSudo(m, applySysctl); err != nil {
		return fmt.Errorf("sysctl: %w", err)
	}

	printGreen("[%s] Setting CPU governor...", m.Name)
	setGovernor := `
set -euo pipefail
if [ ! -d /sys/devices/system/cpu/cpu0/cpufreq ]; then
  echo "[tune] CPU frequency scaling not available, skipping governor"
  exit 0
fi
CURRENT=$(cat /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor 2>/dev/null || echo "unknown")
if [ "$CURRENT" = "performance" ]; then
  echo "[tune] CPU governor already set to performance"
  exit 0
fi
echo "[tune] Setting CPU governor to performance (was: $CURRENT)..."
if ! command -v cpufreq-set &>/dev/null; then
  apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq cpufrequtils 2>/dev/null || true
fi
for cpu_dir in /sys/devices/system/cpu/cpu*/cpufreq; do
  echo "performance" > "$cpu_dir/scaling_governor" 2>/dev/null || true
done
echo "[tune] CPU governor set to performance"
`
	if err := sshRunSudo(m, setGovernor); err != nil {
		return fmt.Errorf("cpu governor: %w", err)
	}

	printGreen("[%s] Disabling turbo boost...", m.Name)
	disableTurbo := `
set -euo pipefail
if [ -f /sys/devices/system/cpu/cpufreq/boost ]; then
  CURRENT=$(cat /sys/devices/system/cpu/cpufreq/boost)
  if [ "$CURRENT" = "0" ]; then
    echo "[tune] AMD turbo boost already disabled"
  else
    echo "[tune] Disabling AMD turbo boost..."
    echo 0 > /sys/devices/system/cpu/cpufreq/boost
    echo "[tune] AMD turbo boost disabled"
  fi
elif [ -f /sys/devices/system/cpu/intel_pstate/no_turbo ]; then
  CURRENT=$(cat /sys/devices/system/cpu/intel_pstate/no_turbo)
  if [ "$CURRENT" = "1" ]; then
    echo "[tune] Intel turbo boost already disabled"
  else
    echo "[tune] Disabling Intel turbo boost..."
    echo 1 > /sys/devices/system/cpu/intel_pstate/no_turbo
    echo "[tune] Intel turbo boost disabled"
  fi
else
  echo "[tune] Turbo boost control not available, skipping"
fi
`
	if err := sshRunSudo(m, disableTurbo); err != nil {
		return fmt.Errorf("disable turbo: %w", err)
	}

	printGreen("[%s] Pinning IRQ affinity...", m.Name)
	if err := sshRunSudo(m, irqAffinityScript); err != nil {
		return fmt.Errorf("irq affinity: %w", err)
	}

	printGreen("[%s] Configuring NUMA...", m.Name)
	configureNuma := `
set -euo pipefail
if ! command -v numactl &>/dev/null; then
  apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq numactl 2>/dev/null || true
fi
if ! command -v numactl &>/dev/null; then
  echo "[tune] numactl not available, skipping NUMA config"
  exit 0
fi
NODES=$(numactl --hardware 2>/dev/null | grep "^available:" | awk '{print $2}')
if [ "${NODES:-0}" -le 1 ]; then
  echo "[tune] Single NUMA node, no NUMA tuning needed"
  exit 0
fi
echo "[tune] NUMA topology: $NODES nodes detected"
echo 1 > /proc/sys/vm/zone_reclaim_mode 2>/dev/null || true
echo "[tune] NUMA zone_reclaim_mode set to 1 (prefer local allocation)"
`
	if err := sshRunSudo(m, configureNuma); err != nil {
		return fmt.Errorf("numa config: %w", err)
	}

	printGreen("[%s] Tune complete", m.Name)
	return nil
}

// Deploy cross-compiles binaries and deploys them to server machines. For each
// server, the architecture-appropriate binary is SCP'd and the control daemon
// is started. Pass a machine name filter or empty string for all servers.
func Deploy(filter string) error {
	mg.Deps(BuildLinuxServer, BuildLinuxArmServer)

	cfg, err := loadHosts()
	if err != nil {
		return err
	}

	machines := filterMachines(cfg, "server", filter)
	if len(machines) == 0 {
		return fmt.Errorf("no server machines matched filter %q", filter)
	}

	type result struct {
		name string
		err  error
	}

	ch := make(chan result, len(machines))
	for _, m := range machines {
		go func(m Machine) {
			var err error
			defer func() {
				if r := recover(); r != nil {
					err = fmt.Errorf("panic: %v", r)
				}
				ch <- result{name: m.Name, err: err}
			}()
			err = deployMachine(m)
		}(m)
	}

	var errs []string
	for range machines {
		r := <-ch
		if r.err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", r.name, r.err))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("deploy failed:\n  %s", strings.Join(errs, "\n  "))
	}

	printGreen("Deploy complete for %d machine(s)", len(machines))
	return nil
}

// deployMachine deploys the server binary to a single machine and starts the
// control daemon.
func deployMachine(m Machine) error {
	ga := goArch(m.Arch)
	localBin := filepath.Join(binDir, "server-linux-"+ga)
	remoteBin := remoteBinDir + "/server"

	// Stop any existing daemon/server before deploying (avoids EADDRINUSE).
	// Use a variable for the pattern to avoid pkill matching its own SSH session.
	printGreen("[%s] Stopping any existing daemon...", m.Name)
	_ = sshRun(m, fmt.Sprintf("PATTERN='%s'; pkill -f \"$PATTERN\" 2>/dev/null; sleep 1", remoteBin))

	printGreen("[%s] Creating remote bin directory...", m.Name)
	if err := sshRun(m, fmt.Sprintf("mkdir -p %s", remoteBinDir)); err != nil {
		return fmt.Errorf("mkdir: %w", err)
	}

	printGreen("[%s] Uploading server binary (%s)...", m.Name, ga)
	if err := scpTo(m, localBin, remoteBin); err != nil {
		return fmt.Errorf("scp: %w", err)
	}

	if err := sshRun(m, fmt.Sprintf("chmod +x %s", remoteBin)); err != nil {
		return fmt.Errorf("chmod: %w", err)
	}

	printGreen("[%s] Starting control daemon...", m.Name)
	startCmd := fmt.Sprintf(
		"nohup %s -mode control -port %s -control-port %d > %s 2>&1 &",
		remoteBin, serverPort, controlPort, serverLogPath,
	)
	if err := sshRun(m, startCmd); err != nil {
		return fmt.Errorf("start daemon: %w", err)
	}

	printGreen("[%s] Waiting for health check...", m.Name)
	if err := waitForHealth(m.Host, controlPort, healthTimeout); err != nil {
		return fmt.Errorf("health: %w", err)
	}

	printGreen("[%s] Deploy complete — healthy", m.Name)
	return nil
}

// BenchRemote runs benchmarks against the remote server fleet. Parameters:
//   - mode: benchmark mode (default "all")
//   - duration: per-benchmark duration (default "30s")
//   - execution: "parallel" or "sequential" (default "parallel")
//
// The BENCH_LEVEL environment variable controls the benchmark level
// ("standard" or "full"). Default: "standard".
func BenchRemote(mode, duration, execution string) error {
	if mode == "" {
		mode = "all"
	}
	if duration == "" {
		duration = "30s"
	}
	if execution == "" {
		execution = "parallel"
	}

	// Validate parameters
	switch mode {
	case "all", "baseline", "theoretical", "celeris":
		// valid
	default:
		return fmt.Errorf("invalid mode %q: must be one of: all, baseline, theoretical, celeris", mode)
	}

	switch execution {
	case "parallel", "sequential":
		// valid
	default:
		return fmt.Errorf("invalid execution %q: must be 'parallel' or 'sequential'", execution)
	}

	mg.Deps(BuildBench)

	cfg, err := loadHosts()
	if err != nil {
		return err
	}

	servers := filterMachines(cfg, "server", "")
	if len(servers) == 0 {
		return fmt.Errorf("no server machines found in hosts config")
	}

	// Find client machine description for dashboard metadata.
	var clientDesc string
	for _, m := range cfg.Machines {
		if m.Role == "client" {
			clientDesc = m.Description
			break
		}
	}

	benchBin := filepath.Join(binDir, "bench")

	versionTag, err := parseCelerisVersion()
	if err != nil {
		return fmt.Errorf("parse celeris version: %w", err)
	}

	benchLevel := os.Getenv("BENCH_LEVEL")
	if benchLevel == "" {
		benchLevel = "standard"
	}
	switch benchLevel {
	case "standard", "full":
		// valid
	default:
		return fmt.Errorf("invalid BENCH_LEVEL %q: must be 'standard' or 'full'", benchLevel)
	}

	runBench := func(m Machine) error {
		outputDir := filepath.Join("results", m.Arch)

		printGreen("Running bench for %s against %s (mode=%s, duration=%s, level=%s)...", m.Arch, m.Host, mode, duration, benchLevel)
		return sh.Run(benchBin,
			"-mode", mode,
			"-duration", duration,
			"-bench-level", benchLevel,
			"-server-ip", m.Host,
			"-control-port", fmt.Sprintf("%d", controlPort),
			"-arch", m.Arch,
			"-output", outputDir,
			"-dashboard-output",
			"-server-machine", m.Description,
			"-client-machine", clientDesc,
			"-version-tag", versionTag,
			"-server-cpus", fmt.Sprintf("%d", m.Threads),
		)
	}

	if execution == "parallel" {
		type result struct {
			name string
			err  error
		}
		ch := make(chan result, len(servers))
		for _, m := range servers {
			go func(m Machine) {
				var err error
				defer func() {
					if r := recover(); r != nil {
						err = fmt.Errorf("panic: %v", r)
					}
					ch <- result{name: m.Name, err: err}
				}()
				err = runBench(m)
			}(m)
		}
		var errs []string
		for range servers {
			r := <-ch
			if r.err != nil {
				errs = append(errs, fmt.Sprintf("%s: %v", r.name, r.err))
			}
		}
		if len(errs) > 0 {
			return fmt.Errorf("bench failed:\n  %s", strings.Join(errs, "\n  "))
		}
	} else {
		for _, m := range servers {
			if err := runBench(m); err != nil {
				return fmt.Errorf("bench %s: %w", m.Name, err)
			}
		}
	}

	printGreen("BenchRemote complete")
	return nil
}

// Cleanup shuts down control daemons on server machines by posting to their
// /shutdown endpoint. Errors are logged but do not fail the target. Pass a
// machine name filter or empty string for all servers.
func Cleanup(filter string) error {
	cfg, err := loadHosts()
	if err != nil {
		return err
	}

	machines := filterMachines(cfg, "server", filter)
	if len(machines) == 0 {
		printGreen("No server machines matched filter %q — nothing to clean up", filter)
		return nil
	}

	client := &http.Client{Timeout: 5 * time.Second}
	for _, m := range machines {
		url := fmt.Sprintf("http://%s:%d%s", m.Host, controlPort, shutdownPath)
		printGreen("[%s] Sending shutdown to %s...", m.Name, url)
		resp, err := client.Post(url, "", nil)
		if err != nil {
			fmt.Printf("[%s] shutdown request failed (ignored): %v\n", m.Name, err)
			continue
		}
		resp.Body.Close()
		printGreen("[%s] Shutdown sent (status %d)", m.Name, resp.StatusCode)
	}

	printGreen("Cleanup complete")
	return nil
}

// Run executes the full remote benchmark pipeline: Setup, Tune, Deploy,
// BenchRemote, Cleanup. Cleanup always runs (via defer). Parameters match
// BenchRemote.
func Run(mode, duration, execution string) error {
	// Fail fast: verify go.mod has the celeris dependency before doing any work.
	if _, err := parseCelerisVersion(); err != nil {
		return fmt.Errorf("pre-flight check: %w", err)
	}

	// Cleanup always runs, even if earlier stages fail.
	defer func() {
		printGreen("Running cleanup...")
		if err := Cleanup(""); err != nil {
			fmt.Printf("Cleanup error (ignored): %v\n", err)
		}
	}()

	printGreen("=== Phase 1/4: Setup ===")
	if err := Setup(""); err != nil {
		return fmt.Errorf("setup: %w", err)
	}

	printGreen("=== Phase 2/4: Tune ===")
	if err := Tune(""); err != nil {
		return fmt.Errorf("tune: %w", err)
	}

	printGreen("=== Phase 3/4: Deploy ===")
	if err := Deploy(""); err != nil {
		return fmt.Errorf("deploy: %w", err)
	}

	printGreen("=== Phase 4/4: Bench ===")
	if err := BenchRemote(mode, duration, execution); err != nil {
		return fmt.Errorf("bench: %w", err)
	}

	printGreen("=== Run complete ===")
	return nil
}

// ---------------------------------------------------------------------------
// Internal helpers
// ---------------------------------------------------------------------------

func ensureGolangciLint() error {
	_, err := exec.LookPath("golangci-lint")
	if err != nil {
		printGreen("golangci-lint not installed. Installing...")
		return sh.Run("go", "install", "github.com/golangci/golangci-lint/cmd/golangci-lint@latest")
	}
	return nil
}