fix: improve process termination and cleanup (#347)

2026-03-02 16:11:13 +00:00
parent 22c0fd2a57
commit 2062f1dbfe
6 changed files with 272 additions and 12 deletions
@@ -15,6 +15,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"

@@ -125,8 +126,12 @@ func run(cfg *config.Config, log *logger.Logger) {
 		}
 	}()

-	// Wait for shutdown signal
-	waitForShutdown(log, func(ctx context.Context) {
+	// Monitor parent liveness via inherited pipe (standalone mode only).
+	// Returns a channel that closes when the parent dies.
+	parentDied := monitorParentLiveness(log)
+
+	// Wait for shutdown signal (OS signal or parent death)
+	waitForShutdown(log, parentDied, func(ctx context.Context) {
 		// Flush pending traces before stopping instances
 		if err := shared.ShutdownTracing(ctx); err != nil {
 			log.Error("error shutting down tracing", zap.Error(err))
@@ -141,11 +146,24 @@ func run(cfg *config.Config, log *logger.Logger) {
 	})
 }

-// waitForShutdown waits for shutdown signal and calls cleanup
-func waitForShutdown(log *logger.Logger, cleanup func(ctx context.Context)) {
-	quit := make(chan os.Signal, 1)
-	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
-	<-quit
+// waitForShutdown waits for a shutdown trigger (OS signal or parent death) and
+// runs the cleanup function. parentDied may be nil when no parent monitor is active.
+func waitForShutdown(log *logger.Logger, parentDied <-chan struct{}, cleanup func(ctx context.Context)) {
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+
+	if parentDied == nil {
+		// No parent monitor — wait for OS signal only.
+		sig := <-sigCh
+		log.Info("received signal", zap.String("signal", sig.String()))
+	} else {
+		select {
+		case sig := <-sigCh:
+			log.Info("received signal", zap.String("signal", sig.String()))
+		case <-parentDied:
+			log.Warn("parent process died, initiating shutdown")
+		}
+	}

 	log.Info("shutting down agentctl...")

@@ -156,3 +174,40 @@ func waitForShutdown(log *logger.Logger, cleanup func(ctx context.Context)) {

 	log.Info("agentctl stopped")
 }
+
+// monitorParentLiveness watches a pipe inherited from the parent process.
+// The parent (kandev backend) passes the read-end of a pipe via ExtraFiles
+// and sets KANDEV_PARENT_PIPE_FD to the FD number. A goroutine blocks on
+// reading the pipe. When the parent dies — even via SIGKILL — the kernel
+// closes the write-end, the read returns, and the returned channel is closed.
+// Returns nil when the env var is absent (Docker, manual start, remote executors).
+func monitorParentLiveness(log *logger.Logger) <-chan struct{} {
+	fdStr := os.Getenv("KANDEV_PARENT_PIPE_FD")
+	if fdStr == "" {
+		return nil
+	}
+
+	fd, err := strconv.Atoi(fdStr)
+	if err != nil {
+		log.Warn("invalid KANDEV_PARENT_PIPE_FD", zap.String("value", fdStr), zap.Error(err))
+		return nil
+	}
+
+	pipe := os.NewFile(uintptr(fd), "parent-liveness-pipe")
+	if pipe == nil {
+		log.Warn("failed to open parent liveness pipe", zap.Int("fd", fd))
+		return nil
+	}
+
+	log.Info("parent liveness monitor started", zap.Int("fd", fd))
+
+	ch := make(chan struct{})
+	go func() {
+		buf := make([]byte, 1)
+		_, _ = pipe.Read(buf) // blocks until pipe breaks
+		_ = pipe.Close()
+		close(ch)
+	}()
+
+	return ch
+}
@@ -0,0 +1,74 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/common/logger"
+)
+
+func newTestLogger(t *testing.T) *logger.Logger {
+	t.Helper()
+	log, err := logger.NewLogger(logger.LoggingConfig{
+		Level:      "warn",
+		Format:     "console",
+		OutputPath: "stdout",
+	})
+	if err != nil {
+		t.Fatalf("failed to create logger: %v", err)
+	}
+	return log
+}
+
+func TestMonitorParentLiveness_NoEnvVar(t *testing.T) {
+	t.Setenv("KANDEV_PARENT_PIPE_FD", "")
+	ch := monitorParentLiveness(newTestLogger(t))
+	if ch != nil {
+		t.Fatal("expected nil channel when env var is absent")
+	}
+}
+
+func TestMonitorParentLiveness_InvalidEnvVar(t *testing.T) {
+	t.Setenv("KANDEV_PARENT_PIPE_FD", "not-a-number")
+	ch := monitorParentLiveness(newTestLogger(t))
+	if ch != nil {
+		t.Fatal("expected nil channel for invalid FD value")
+	}
+}
+
+func TestMonitorParentLiveness_PipeBreak(t *testing.T) {
+	// Create a pipe: r is the read-end (what agentctl would inherit),
+	// w is the write-end (what the parent keeps open).
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Pipe: %v", err)
+	}
+	defer func() { _ = r.Close() }()
+
+	// Point the env var at the read-end's FD.
+	t.Setenv("KANDEV_PARENT_PIPE_FD", fmt.Sprintf("%d", r.Fd()))
+
+	ch := monitorParentLiveness(newTestLogger(t))
+	if ch == nil {
+		t.Fatal("expected non-nil channel")
+	}
+
+	// Channel should not be closed yet.
+	select {
+	case <-ch:
+		t.Fatal("channel closed before pipe break")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	// Close write-end to simulate parent death.
+	_ = w.Close()
+
+	select {
+	case <-ch:
+		// success
+	case <-time.After(2 * time.Second):
+		t.Fatal("channel not closed after pipe break")
+	}
+}
@@ -33,6 +33,11 @@ type Launcher struct {

 	// For clean shutdown
 	stopping bool
+
+	// Parent liveness pipe: write-end kept open by the launcher.
+	// When the backend dies (even SIGKILL), the kernel closes this FD,
+	// breaking the pipe. agentctl detects the break and self-terminates.
+	parentPipe *os.File
 }

 // Config holds configuration for the launcher.
@@ -144,9 +149,19 @@ func (l *Launcher) Start(ctx context.Context) error {

 	// Set process attributes:
 	// - Pdeathsig on Linux: kernel sends SIGTERM to child when parent dies.
-	// - Setpgid: create new process group so Ctrl+C doesn't propagate directly.
 	l.cmd.SysProcAttr = buildSysProcAttr()

+	// Parent liveness pipe: agentctl inherits the read-end (FD 3 via ExtraFiles)
+	// and blocks on it. When this process dies — even via SIGKILL — the kernel
+	// closes the write-end, breaking the pipe. agentctl detects the break and
+	// initiates graceful shutdown. This is the portable equivalent of Pdeathsig.
+	pipeRead, pipeWrite, err := os.Pipe()
+	if err != nil {
+		return fmt.Errorf("failed to create parent liveness pipe: %w", err)
+	}
+	l.cmd.ExtraFiles = []*os.File{pipeRead} // child FD 3
+	l.cmd.Env = append(l.cmd.Env, "KANDEV_PARENT_PIPE_FD=3")
+
 	// Capture stdout and stderr
 	stdout, err := l.cmd.StdoutPipe()
 	if err != nil {
@@ -159,9 +174,16 @@ func (l *Launcher) Start(ctx context.Context) error {

 	// Start the process
 	if err := l.cmd.Start(); err != nil {
+		_ = pipeRead.Close()
+		_ = pipeWrite.Close()
 		return fmt.Errorf("failed to start agentctl: %w", err)
 	}

+	// Close read-end in parent (child inherited it). Keep write-end open —
+	// it closes automatically when this process exits, signaling agentctl.
+	_ = pipeRead.Close()
+	l.parentPipe = pipeWrite
+
 	l.logger.Info("agentctl process started", zap.Int("pid", l.cmd.Process.Pid))

 	// Pipe stdout/stderr to logger in background
@@ -177,6 +199,7 @@ func (l *Launcher) Start(ctx context.Context) error {
 		if killErr := l.cmd.Process.Kill(); killErr != nil {
 			l.logger.Warn("failed to kill agentctl process after failed health check", zap.Error(killErr))
 		}
+		l.closeParentPipeLocked()
 		return fmt.Errorf("agentctl failed to become healthy: %w", err)
 	}

@@ -208,6 +231,10 @@ func (l *Launcher) Stop(ctx context.Context) error {

 	l.logger.Info("stopping agentctl subprocess", zap.Int("pid", pid))

+	// Close the liveness pipe first — this signals agentctl that the parent
+	// is shutting down, complementing the SIGTERM that follows.
+	l.closeParentPipe()
+
 	// Send graceful stop signal (SIGTERM on Unix, interrupt on Windows)
 	if err := l.gracefulStop(pid); err != nil {
 		return err
@@ -231,6 +258,22 @@ func (l *Launcher) Stop(ctx context.Context) error {
 	}
 }

+// closeParentPipe closes and nils the parent liveness pipe if open.
+// Safe for concurrent use.
+func (l *Launcher) closeParentPipe() {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.closeParentPipeLocked()
+}
+
+// closeParentPipeLocked is like closeParentPipe but assumes mu is already held.
+func (l *Launcher) closeParentPipeLocked() {
+	if l.parentPipe != nil {
+		_ = l.parentPipe.Close()
+		l.parentPipe = nil
+	}
+}
+
 // checkPortAvailable verifies the given port is not in use.
 // It checks by attempting a wildcard bind (matching what agentctl does with ":port").
 func checkPortAvailable(port int) error {
@@ -0,0 +1,50 @@
+//go:build !windows
+
+package launcher
+
+import (
+	"os"
+	"testing"
+)
+
+func TestCloseParentPipe_ClosesAndNils(t *testing.T) {
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Pipe: %v", err)
+	}
+	defer func() { _ = r.Close() }()
+
+	l := &Launcher{parentPipe: w}
+	l.closeParentPipe()
+
+	if l.parentPipe != nil {
+		t.Error("parentPipe should be nil after closeParentPipe")
+	}
+
+	// Writing to the closed write-end should fail.
+	_, err = w.Write([]byte("x"))
+	if err == nil {
+		t.Error("expected error writing to closed pipe")
+	}
+}
+
+func TestCloseParentPipe_NilIsNoop(t *testing.T) {
+	l := &Launcher{}
+	// Must not panic.
+	l.closeParentPipe()
+	if l.parentPipe != nil {
+		t.Error("parentPipe should remain nil")
+	}
+}
+
+func TestCloseParentPipe_Idempotent(t *testing.T) {
+	_, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Pipe: %v", err)
+	}
+
+	l := &Launcher{parentPipe: w}
+	l.closeParentPipe()
+	// Second call must not panic (double close).
+	l.closeParentPipe()
+}
@@ -20,8 +20,6 @@ func (l *Launcher) gracefulStop(pid int) error {
 }

 // forceKill sends SIGKILL to the agentctl process.
-// agentctl shares the backend's process group, so we kill the single process
-// rather than the group (which would kill the backend itself).
 func (l *Launcher) forceKill(pid int) {
 	_ = syscall.Kill(pid, syscall.SIGKILL)
 }
@@ -51,6 +51,45 @@ function killProcess(proc: ChildProcess): Promise<void> {
  });
 }

+/**
+ * Kills an entire process group. Used for the backend process which is spawned
+ * with `detached: true` so it becomes a process group leader. Sending signals
+ * to the negative PID targets all processes in that group (backend + agentctl).
+ * The 7s grace period gives agentctl time to cascade cleanup to agent process groups.
+ */
+function killProcessGroup(proc: ChildProcess): Promise<void> {
+  return new Promise<void>((resolve) => {
+    if (!proc.pid) {
+      resolve();
+      return;
+    }
+
+    const pid = proc.pid;
+
+    try {
+      process.kill(-pid, "SIGTERM");
+    } catch {
+      // Process group may already be gone
+      resolve();
+      return;
+    }
+
+    const timeout = setTimeout(() => {
+      try {
+        process.kill(-pid, "SIGKILL");
+      } catch {
+        // Already dead
+      }
+      resolve();
+    }, 7_000);
+
+    proc.on("exit", () => {
+      clearTimeout(timeout);
+      resolve();
+    });
+  });
+}
+
 /**
 * Worker-scoped fixture that spawns an isolated backend process and
 * a dedicated Next.js frontend. Each Playwright worker gets its own
@@ -113,6 +152,7 @@ export const backendFixture = base.extend<object, { backend: BackendContext }>({
      const backendProc: ChildProcess = spawn(KANDEV_BIN, [], {
        env: backendEnv as unknown as NodeJS.ProcessEnv,
        stdio: ["ignore", "pipe", "pipe"],
+        detached: true, // Own process group so we can kill backend + agentctl together
      });

      const logFile = debug ? fs.createWriteStream(`/tmp/e2e-backend-${backendPort}.log`) : null;
@@ -167,9 +207,9 @@ export const backendFixture = base.extend<object, { backend: BackendContext }>({
      try {
        await use({ port: backendPort, baseUrl, frontendPort, frontendUrl, tmpDir });
      } finally {
-        // Shutdown frontend first, then backend
+        // Shutdown frontend first (simple process), then backend (process group)
        await killProcess(frontendProc);
-        await killProcess(backendProc);
+        await killProcessGroup(backendProc);

        // Cleanup temp directory — ignore errors (backend may still hold files briefly)
        try {