refactor(engine): killer 的上游复制段归零

engine/UPSTREAM.md

@@ -31,10 +31,10 @@
 **9,621**（对照 `f4c50506`）、**9,659**（对照 `be48afbe`），生产 3,852、
 测试 5,758。
 
-2026-11-12 的 ports 重写在此基础上继续下降：对照 `26dbcae` 为 **8,933** 行 / 664 段，
-分布在 135 个文件（生产 3,564、测试 5,369）。
+2026-11-12 的 ports 与 killer 重写在此基础上继续下降：对照 `26dbcae` 为 **8,689**
+行 / 636 段，分布在 131 个文件（生产 3,478、测试 5,211）。
 
-按包（对照 `26dbcae`，ports 重写后）：
+按包（对照 `26dbcae`，ports 与 killer 重写后）：
 
 | 包 | 生产 | 测试 | 合计 |
 | --- | ---: | ---: | ---: |
@@ -43,12 +43,13 @@
 | `internal/groups` | 336 | 506 | 842 |
 | `internal/scanner` | 220 | 540 | 760 |
 | `internal/ports` | 0 | 0 | 0 |
-| `internal/killer` | 86 | 158 | 244 |
+| `internal/killer` | 0 | 0 | 0 |
 | `internal/store` | 12 | 5 | 17 |
 
-`internal/claims`、`internal/display`、`internal/testenv`、`internal/ports` 与
-`internal/store/claims*.go` 已经归零；`internal/docker` 只剩结构体字段与表驱动
-用例数据。`internal/store` 已整包重写：生产代码从 1,076 行降到 199 行。
+`internal/claims`、`internal/display`、`internal/testenv`、`internal/ports`、
+`internal/killer` 与 `internal/store/claims*.go` 已经归零；`internal/docker` 只剩
+结构体字段与表驱动用例数据。`internal/store` 已整包重写：生产代码从 1,076 行降到
+199 行。
 
 ```sh
 git clone https://github.com/raskrebs/sonar /tmp/sonar

engine/internal/killer/escalate.go

@@ -2,6 +2,7 @@ package killer
 
 import (
 	"context"
+	"slices"
 
 	"github.com/sheathedsharp/option-berth/engine/internal/state"
 )
@@ -110,30 +111,24 @@ func noteCancellation(ctx context.Context, pending []*unit) bool {
 func (e *engine) stillUp(units, scratch []*unit) []*unit {
 	out := scratch[:0]
 	for _, u := range units {
-		if u.port > 0 {
-			if e.probe(u.port, u.bind) != portClosed {
-				out = append(out, u)
-			}
+		portUnit := u.port > 0
+		// A timed-out probe proves nothing; only a refusal means gone.
+		if portUnit && e.probe(u.port, u.bind) == portClosed {
 			continue
 		}
-		if e.anyAlive(u) {
+		if portUnit || e.anyAlive(u) {
 			out = append(out, u)
 		}
 	}
 	return out
 }
 
-// anyAlive reports whether any process of the unit is still running.
+// anyAlive answers whether the unit still has a process running.
 func (e *engine) anyAlive(u *unit) bool {
 	if u.pgid != 0 {
 		return e.alive(u.pgid)
 	}
-	for _, pid := range u.pids {
-		if e.alive(pid) {
-			return true
-		}
-	}
-	return false
+	return slices.ContainsFunc(u.pids, e.alive)
 }
 
 // hardKill sends SIGKILL to whatever of the unit is left, keeping the

engine/internal/killer/integration_test.go

@@ -9,6 +9,7 @@ import (
 	"os/exec"
 	"os/signal"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"syscall"
@@ -53,59 +54,74 @@ func TestMain(m *testing.M) {
 	os.Exit(testenv.Run(m))
 }
 
-// runHelper listens on port, optionally spawns a child helper on childPort, and
-// then blocks until it is signalled. With ignoreTerm it plays the wedged dev
-// server that only SIGKILL can stop; otherwise the default SIGTERM disposition
-// is what a well-behaved one gives us.
-//
-// Every helper reports its own pid into pidDir, named after the port it serves.
-// A helper knows its pid and its port and nothing else, so that name is the one
-// thing it can write without being told; a parent spawning a child therefore
-// does not have to pass an identity down.
+// runHelper is the listener a test starts by re-executing this binary. It
+// serves its port until it is signalled, and with ignoreTerm it plays the wedged
+// dev server only SIGKILL can stop. Before serving it writes its own pid into
+// pidDir under the port's name, because a helper knows its pid and its port and
+// nothing else; that lets a parent spawn a child without passing an identity
+// down.
 func runHelper(port, childPort, pidDir string, ignoreTerm bool) {
 	if ignoreTerm {
 		signal.Ignore(syscall.SIGTERM)
 	}
-	ln, err := net.Listen("tcp", "127.0.0.1:"+port)
+	sock, err := net.Listen("tcp", "127.0.0.1:"+port)
 	if err != nil {
 		os.Exit(1)
 	}
-	defer ln.Close()
-	// Report this process before serving, so a test never races the answer.
+	defer sock.Close()
 	if pidDir != "" {
-		if err := os.WriteFile(pidFileFor(pidDir, port), []byte(strconv.Itoa(os.Getpid())), 0o600); err != nil {
+		pid := strconv.Itoa(os.Getpid())
+		if err := os.WriteFile(pidFileFor(pidDir, port), []byte(pid), 0o600); err != nil {
 			os.Exit(1)
 		}
 	}
-
 	if childPort != "" {
-		child := exec.Command(os.Args[0])
-		child.Env = append(os.Environ(), envHelperPort+"="+childPort, envHelperChild+"=")
-		if err := child.Start(); err != nil {
-			os.Exit(1)
-		}
+		spawnHelper(childPort)
 	}
 	for {
-		conn, err := ln.Accept()
+		conn, err := sock.Accept()
 		if err != nil {
 			return
 		}
 		_ = conn.Close()
 	}
 }
 
+// spawnHelper starts one more helper on port, as this helper's child.
+func spawnHelper(port string) {
+	child := exec.Command(os.Args[0])
+	child.Env = append(os.Environ(), envHelperPort+"="+port, envHelperChild+"=")
+	if err := child.Start(); err != nil {
+		os.Exit(1)
+	}
+}
+
 // pidFileFor is the path a helper serving port reports itself to.
 func pidFileFor(dir, port string) string { return filepath.Join(dir, "helper-"+port+".pid") }
 
 // freePort asks the kernel for a port nothing is using and hands it back.
 func freePort(t *testing.T) int {
 	t.Helper()
-	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	sock, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatalf("reserving a port: %v", err)
 	}
-	defer ln.Close()
-	return ln.Addr().(*net.TCPAddr).Port
+	defer sock.Close()
+	addr, ok := sock.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("reserved %T, want a TCP address", sock.Addr())
+	}
+	return addr.Port
+}
+
+// mustExecutable is this test binary's path, which is also the helper program.
+func mustExecutable(t *testing.T) string {
+	t.Helper()
+	exe, err := os.Executable()
+	if err != nil {
+		t.Fatalf("locating the test binary: %v", err)
+	}
+	return exe
 }
 
 // portAnswers is the test's own question — "is something accepting here?" — and
@@ -116,13 +132,15 @@ func portAnswers(port int) bool { return probePort(port, "127.0.0.1") == portOpe
 func waitForPort(t *testing.T, port int, want bool) {
 	t.Helper()
 	deadline := time.Now().Add(10 * time.Second)
-	for time.Now().Before(deadline) {
+	for {
 		if portAnswers(port) == want {
 			return
 		}
+		if time.Now().After(deadline) {
+			t.Fatalf("port %d: still open=%v after 10s, wanted open=%v", port, !want, want)
+		}
 		time.Sleep(20 * time.Millisecond)
 	}
-	t.Fatalf("port %d: still open=%v after 10s, wanted open=%v", port, !want, want)
 }
 
 // readPID waits for a helper to report the pid it runs as.
@@ -153,10 +171,7 @@ type tree struct {
 // is a child of a shell, and waits until both listeners answer.
 func startTree(t *testing.T) tree {
 	t.Helper()
-	exe, err := os.Executable()
-	if err != nil {
-		t.Fatalf("locating the test binary: %v", err)
-	}
+	exe := mustExecutable(t)
 	dir := t.TempDir()
 	parentPort, childPort := freePort(t), freePort(t)
 	got := tree{parentPort: parentPort, childPort: childPort}
@@ -219,13 +234,15 @@ func stopOwnedTree(cmd *exec.Cmd) {
 // that cannot see the socket table does not turn these cases into skips.
 func snapshotFor(t *testing.T, port, knownPID int) []ports.ListeningPort {
 	t.Helper()
-	if found, err := ports.Scan(); err == nil {
-		docker.EnrichPorts(found)
-		ports.Enrich(found)
-		for i := range found {
-			if found[i].Port == port {
-				return found
-			}
+	found, err := ports.Scan()
+	if err != nil {
+		return []ports.ListeningPort{snapshotForPID(knownPID, port)}
+	}
+	docker.EnrichPorts(found)
+	ports.Enrich(found)
+	for _, row := range found {
+		if row.Port == port {
+			return found
 		}
 	}
 	return []ports.ListeningPort{snapshotForPID(knownPID, port)}
@@ -242,28 +259,34 @@ func TestKillPortsStopsARealProcessTree(t *testing.T) {
 		Options{Tree: true, Grace: 2 * time.Second, Ports: snapshotFor(t, tree.parentPort, tree.parentPID)},
 	)
 
-	if len(results) == 0 {
-		t.Fatal("no result rows")
-	}
-	for _, r := range results {
-		if !r.OK {
-			t.Errorf("row %+v failed: %s", r, r.Error)
-		}
-	}
+	requireAllOK(t, results)
 	// The listener and the port it spawned are both gone.
 	waitForPort(t, tree.parentPort, false)
 	waitForPort(t, tree.childPort, false)
 
 	// And the shell that was waiting on them has nothing left to wait for.
-	done := make(chan struct{})
-	go func() { _ = tree.shell.Wait(); close(done) }()
+	exited := make(chan struct{})
+	go func() { _ = tree.shell.Wait(); close(exited) }()
 	select {
-	case <-done:
+	case <-exited:
 	case <-time.After(10 * time.Second):
 		t.Fatal("the shell is still running after its tree was killed")
 	}
 }
 
+// requireAllOK fails the test when any row reports a stop that did not happen.
+func requireAllOK(t *testing.T, results []Result) {
+	t.Helper()
+	if len(results) == 0 {
+		t.Fatal("no result rows")
+	}
+	for i, r := range results {
+		if !r.OK {
+			t.Errorf("row %d (%+v) failed: %s", i, r, r.Error)
+		}
+	}
+}
+
 func TestKillPortsDryRunLeavesARealTreeRunning(t *testing.T) {
 	if testing.Short() {
 		t.Skip("spawns real processes")
@@ -286,30 +309,17 @@ func TestKillPortsDryRunLeavesARealTreeRunning(t *testing.T) {
 	root := results[len(results)-1].PID
 	below := scanProcessTable().Descendants(root)
 	for _, r := range results[:len(results)-1] {
-		if !contains(below, r.PID) {
+		if !slices.Contains(below, r.PID) {
 			t.Errorf("row %+v is not below the last row (PID %d)", r, root)
 		}
 	}
 }
 
-func contains(haystack []int, needle int) bool {
-	for _, v := range haystack {
-		if v == needle {
-			return true
-		}
-	}
-	return false
-}
-
 func TestKillPortsEscalatesOnAProcessThatIgnoresSIGTERM(t *testing.T) {
 	if testing.Short() {
 		t.Skip("spawns real processes")
 	}
-	port := freePort(t)
-	exe, err := os.Executable()
-	if err != nil {
-		t.Fatal(err)
-	}
+	port, exe := freePort(t), mustExecutable(t)
 	pidDir := t.TempDir()
 	wedged := exec.Command(exe)
 	wedged.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
@@ -330,14 +340,10 @@ func TestKillPortsEscalatesOnAProcessThatIgnoresSIGTERM(t *testing.T) {
 	// The listener ignores SIGTERM, so only the escalation to SIGKILL frees
 	// the port. A short grace keeps the test quick.
 	snapshot := []ports.ListeningPort{snapshotForPID(readPID(t, pidFileFor(pidDir, strconv.Itoa(port))), port)}
-	results := KillPorts(context.Background(), []Target{{Port: port}},
-		Options{Grace: 500 * time.Millisecond, Ports: snapshot})
+	results := KillPorts(context.Background(), []Target{{Port: port}}, Options{Grace: 500 * time.Millisecond, Ports: snapshot})
 
-	if len(results) != 1 {
-		t.Fatalf("results = %+v", results)
-	}
-	if results[0].Method != state.MethodSIGKILL {
-		t.Fatalf("method = %q, want sigkill after the grace period", results[0].Method)
+	if len(results) != 1 || results[0].Method != state.MethodSIGKILL {
+		t.Fatalf("results = %+v, want one sigkill row after the grace period", results)
 	}
 	waitForPort(t, port, false)
 }