fix(shutdown): scope a local drain to its own node and stop counting failed phases as done (#780)

* fix(shutdown): scope a local drain to its own node and stop counting failed phases as done A node's spinifex-shutdown.service publishes its GATE and DRAIN requests to the shared spinifex.cluster.shutdown.* subjects, which every daemon subscribes to as a fan-out. The node filter only decided which ACK the stopping node waited for, so by the time it applied, every other node had already stopped its gateway, UI and VPC daemon and powered down its guests. Stopping one node took down every guest in the cluster. Requests now carry the node they apply to, and a daemon drops one addressed elsewhere without ACKing. A cluster shutdown leaves the field empty, so it still reaches every node. The STORAGE phase also ran 'pkill -f nbdkit', which matches every nbdkit on the host: another node's backends on a shared host, and a live guest's backend that survived a daemon restart under KillMode=process. Cleanup now sweeps this node's own nbdkit-vol-*.pid files and signals a PID only while /proc still agrees it is nbdkit, so a recycled PID is never hit. Finally, the coordinator counted errored ACKs toward its per-phase quorum, so a node that failed to drain reported N/N and the run advanced into STORAGE and PERSIST with guests still running. Only clean ACKs count now, and any errored one aborts the phase unless --force. * test(shutdown): cover the phase gate, every handler's target check and the real proc paths Diff coverage came in at 52.9% against a 60% threshold. The gaps were the per-phase target checks beyond GATE, the real procComm and signalProcess bodies that every cleanup test stubs out, and the coordinator's abort branch, which sat inside runClusterShutdown where no unit test could reach it. The abort decision moves into shutdownPhaseOutcome, a pure function returning the completed count and the message to print, so the rule that an errored node does not count toward the quorum is testable on its own rather than only reachable through a full cluster shutdown.
2026-10-19 10:52:04 +00:00 · 2026-08-13 11:50:13 +10:00
parent 6b71fb876d
commit 2fe885441b
6 changed files with 532 additions and 11 deletions
@@ -248,7 +248,9 @@ func runClusterShutdown(cmd *cobra.Command, args []string) {

 		// Unsubscribe from progress
 		if progressSub != nil {
-			progressSub.Unsubscribe()
+			if err := progressSub.Unsubscribe(); err != nil {
+				fmt.Fprintf(os.Stderr, "Warning: failed to unsubscribe from progress: %v\n", err)
+			}
 		}

 		// Print results
@@ -262,10 +264,9 @@ func runClusterShutdown(cmd *cobra.Command, args []string) {
 			}
 		}

-		ackedCount := len(acks)
-		if ackedCount < nodeCount && !force {
-			fmt.Fprintf(os.Stderr, "[%s] Only %d/%d nodes responded. Use --force to continue.\n",
-				strings.ToUpper(phase), ackedCount, nodeCount)
+		ackedCount, abort := shutdownPhaseOutcome(phase, acks, nodeCount, force)
+		if abort != "" {
+			fmt.Fprintln(os.Stderr, abort)
 			os.Exit(1)
 		}

@@ -319,7 +320,7 @@ func runNodeDrainLocal(cmd *cobra.Command, args []string) {

 	for _, phase := range []string{"gate", "drain"} {
 		topic := "spinifex.cluster.shutdown." + phase
-		req := daemon.ShutdownRequest{Phase: phase, Timeout: int(timeout.Seconds())}
+		req := localDrainRequest(phase, node, timeout)
 		reqData, err := json.Marshal(req)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "Error marshaling request: %v\n", err)
@@ -422,6 +423,51 @@ var (
 	localShutdownACKRetryMaxDelay  = 2 * time.Second
 )

+// localDrainRequest builds a phase request scoped to one node. Target is what
+// keeps it local: the phase subjects are fan-out, so an untargeted request
+// gates and drains every node in the cluster off one node's target stop.
+func localDrainRequest(phase, node string, timeout time.Duration) daemon.ShutdownRequest {
+	return daemon.ShutdownRequest{
+		Phase:   phase,
+		Timeout: int(timeout.Seconds()),
+		Target:  node,
+	}
+}
+
+// shutdownPhaseOutcome decides whether a phase may advance. A node that
+// answered with an error has not completed the phase, and counting it as done
+// would tear storage out from under guests that never stopped. Returns the
+// completed count and, when the run must stop, the message to print.
+func shutdownPhaseOutcome(phase string, acks []daemon.ShutdownACK, nodeCount int, force bool) (completed int, abort string) {
+	completed, failed := summariseShutdownACKs(acks)
+	if force {
+		return completed, ""
+	}
+
+	if len(failed) > 0 {
+		return completed, fmt.Sprintf("[%s] %d node(s) failed the phase: %s. Use --force to continue.",
+			strings.ToUpper(phase), len(failed), strings.Join(failed, ", "))
+	}
+	if completed < nodeCount {
+		return completed, fmt.Sprintf("[%s] Only %d/%d nodes completed. Use --force to continue.",
+			strings.ToUpper(phase), completed, nodeCount)
+	}
+	return completed, ""
+}
+
+// summariseShutdownACKs splits a phase's ACKs into the count that completed
+// cleanly and the names of the nodes that answered with an error.
+func summariseShutdownACKs(acks []daemon.ShutdownACK) (completed int, failed []string) {
+	for _, ack := range acks {
+		if ack.Error != "" {
+			failed = append(failed, ack.Node)
+			continue
+		}
+		completed++
+	}
+	return completed, failed
+}
+
 // collectShutdownACKsFn indirects collectShutdownACKs so tests can simulate a
 // slow-to-resubscribe daemon without a real NATS round trip.
 var collectShutdownACKsFn = collectShutdownACKs
@@ -367,6 +367,98 @@ func shrinkLocalShutdownACKBackoff(t *testing.T) {
 	})
 }

+// TestLocalDrainRequestIsNodeScoped pins the field that keeps a local drain
+// local. The phase subjects are fan-out, so an empty Target here drains the
+// guests on every node in the cluster, not just the one being stopped.
+func TestLocalDrainRequestIsNodeScoped(t *testing.T) {
+	req := localDrainRequest("drain", "node2", 90*time.Second)
+
+	assert.Equal(t, "node2", req.Target, "a local drain must name the node it applies to")
+	assert.Equal(t, "drain", req.Phase)
+	assert.Equal(t, 90, req.Timeout)
+}
+
+// TestSummariseShutdownACKs covers phase accounting by outcome: a node that
+// answered with an error has not completed the phase, and counting it as done
+// advances the shutdown into STORAGE with guests still running.
+func TestSummariseShutdownACKs(t *testing.T) {
+	tests := []struct {
+		name      string
+		acks      []daemon.ShutdownACK
+		completed int
+		failed    []string
+	}{
+		{
+			name:      "all clean",
+			acks:      []daemon.ShutdownACK{{Node: "node1"}, {Node: "node2"}},
+			completed: 2,
+		},
+		{
+			name: "one errored is not counted as completed",
+			acks: []daemon.ShutdownACK{
+				{Node: "node1"},
+				{Node: "node2", Error: "failed to stop VMs"},
+				{Node: "node3"},
+			},
+			completed: 2,
+			failed:    []string{"node2"},
+		},
+		{
+			name:      "all errored",
+			acks:      []daemon.ShutdownACK{{Node: "node1", Error: "boom"}, {Node: "node2", Error: "boom"}},
+			completed: 0,
+			failed:    []string{"node1", "node2"},
+		},
+		{
+			name:      "no acks",
+			acks:      nil,
+			completed: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			completed, failed := summariseShutdownACKs(tt.acks)
+			assert.Equal(t, tt.completed, completed)
+			assert.Equal(t, tt.failed, failed)
+		})
+	}
+}
+
+// TestShutdownPhaseOutcome covers the gate between phases. Advancing past a
+// node that failed to drain is what tears storage out from under a guest that
+// never stopped, so only --force may do it.
+func TestShutdownPhaseOutcome(t *testing.T) {
+	clean := []daemon.ShutdownACK{{Node: "node1"}, {Node: "node2"}, {Node: "node3"}}
+	mixed := []daemon.ShutdownACK{{Node: "node1"}, {Node: "node2", Error: "failed to stop VMs"}, {Node: "node3"}}
+
+	t.Run("every node clean advances", func(t *testing.T) {
+		completed, abort := shutdownPhaseOutcome("drain", clean, 3, false)
+		assert.Equal(t, 3, completed)
+		assert.Empty(t, abort)
+	})
+
+	t.Run("an errored node aborts and is named", func(t *testing.T) {
+		completed, abort := shutdownPhaseOutcome("drain", mixed, 3, false)
+		assert.Equal(t, 2, completed)
+		assert.Contains(t, abort, "node2")
+		assert.Contains(t, abort, "failed the phase")
+		assert.Contains(t, abort, "--force")
+	})
+
+	t.Run("a silent node aborts on the count", func(t *testing.T) {
+		completed, abort := shutdownPhaseOutcome("storage", clean, 5, false)
+		assert.Equal(t, 3, completed)
+		assert.Contains(t, abort, "3/5")
+	})
+
+	t.Run("force advances past both", func(t *testing.T) {
+		completed, abort := shutdownPhaseOutcome("drain", mixed, 5, true)
+		assert.Equal(t, 2, completed)
+		assert.Empty(t, abort, "--force is the documented way to continue despite errors")
+	})
+}
+
 // TestCollectLocalShutdownACKRetriesUntilACK covers the actual bug: the
 // daemon may not have re-subscribed right after a restart, so the first
 // attempt(s) get no ACK. The ACK on the third attempt must still succeed.
@@ -23,6 +23,7 @@ resources:
 - [Overview](#overview)
 - [The Contract](#the-contract)
 - [Older Releases: `systemctl stop spinifex.target`](#older-releases-systemctl-stop-spinifextarget)
+- [Older Releases: A Local Drain Drained the Whole Cluster](#older-releases-a-local-drain-drained-the-whole-cluster)
 - [Why Guests Outlive the Daemon](#why-guests-outlive-the-daemon)
 - [Checking State](#checking-state)
 - [Maintenance Runbooks](#maintenance-runbooks)
@@ -119,6 +120,36 @@ This does not apply to `systemctl restart spinifex.target` — the restart row
 in the contract table above is unaffected, since the same gate is what makes
 guests survive a restart intact.

+## Older Releases: A Local Drain Drained the Whole Cluster
+
+> [!WARNING]
+> **Releases up to the one carrying this section drain every node in the
+> cluster when any single node stops its target.** Stopping or rebooting one
+> node powers down every guest on every other node, stops their API gateway,
+> UI and VPC daemon, and leaves each daemon refusing new work until it is
+> restarted. On a multi-node cluster this reads as an unexplained cluster-wide
+> outage triggered by routine single-node maintenance.
+
+**Cause:** the local drain published its GATE and DRAIN requests to the shared
+`spinifex.cluster.shutdown.*` subjects, which every daemon subscribes to as a
+fan-out. The node filter applied only to the replies the stopping node waited
+for, so the other nodes had already drained by the time it was applied.
+Requests now name the node they apply to, and a daemon ignores one addressed
+to another node.
+
+Both ends have to be current for this to hold: the node that issues the drain
+sends the target, and the nodes that receive it must be new enough to honour
+it. A cluster mid-upgrade still drains cluster-wide from any node that has not
+yet been updated. **Upgrade every node before relying on single-node
+maintenance**, and until then stop guests deliberately rather than as a side
+effect of stopping a target.
+
+To check whether a cluster is exposed, confirm each node's `spx` is current:
+
+```bash
+spx version
+```
+
 ## Why Guests Outlive the Daemon

 `KillMode=process` is set on `spinifex-daemon.service` and
@@ -3,10 +3,13 @@ package daemon
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log/slog"
 	"os"
-	"os/exec"
 	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"

 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
@@ -28,6 +31,17 @@ type ShutdownRequest struct {
 	Phase   string `json:"phase"`
 	Force   bool   `json:"force"`
 	Timeout int    `json:"timeout_seconds"`
+	// Target names the only node that should act on this request. Empty means
+	// every node, which is what a whole-cluster shutdown sends. The phase
+	// subjects are fan-out, so without this a single node's stop drains all.
+	Target string `json:"target,omitempty"`
+}
+
+// forAnotherNode reports whether a targeted request belongs to a different
+// node. Such a request is dropped without an ACK: the sender is waiting for
+// its own node only, and an ACK from here would just be discarded.
+func (d *Daemon) forAnotherNode(req ShutdownRequest) bool {
+	return req.Target != "" && req.Target != d.node
 }

 // ShutdownACK is the response from a daemon after completing a shutdown phase.
@@ -55,6 +69,9 @@ func (d *Daemon) handleShutdownGate(msg *nats.Msg) {
 		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "gate", Error: err.Error()})
 		return
 	}
+	if d.forAnotherNode(req) {
+		return
+	}

 	slog.Info("Shutdown GATE phase starting", "node", d.node, "force", req.Force)

@@ -108,6 +125,9 @@ func (d *Daemon) handleShutdownDrain(msg *nats.Msg) {
 		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "drain", Error: err.Error()})
 		return
 	}
+	if d.forAnotherNode(req) {
+		return
+	}

 	slog.Info("Shutdown DRAIN phase starting", "node", d.node)

@@ -161,6 +181,9 @@ func (d *Daemon) handleShutdownStorage(msg *nats.Msg) {
 		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "storage", Error: err.Error()})
 		return
 	}
+	if d.forAnotherNode(req) {
+		return
+	}

 	slog.Info("Shutdown STORAGE phase starting", "node", d.node)

@@ -174,10 +197,7 @@ func (d *Daemon) handleShutdownStorage(msg *nats.Msg) {
 		}
 	}

-	// Best-effort cleanup of orphaned nbdkit processes
-	if err := exec.Command("pkill", "-f", "nbdkit").Run(); err != nil {
-		slog.Debug("pkill nbdkit (best-effort)", "result", err)
-	}
+	cleanupOrphanNBDKit(utils.RuntimeDir())

 	ack := ShutdownACK{
 		Node:    d.node,
@@ -196,6 +216,9 @@ func (d *Daemon) handleShutdownPersist(msg *nats.Msg) {
 		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "persist", Error: err.Error()})
 		return
 	}
+	if d.forAnotherNode(req) {
+		return
+	}

 	slog.Info("Shutdown PERSIST phase starting", "node", d.node)

@@ -227,6 +250,9 @@ func (d *Daemon) handleShutdownInfra(msg *nats.Msg) {
 		d.respondShutdownACK(msg, ShutdownACK{Node: d.node, Phase: "infra", Error: err.Error()})
 		return
 	}
+	if d.forAnotherNode(req) {
+		return
+	}

 	slog.Info("Shutdown INFRA phase starting", "node", d.node)

@@ -265,6 +291,79 @@ func (d *Daemon) handleShutdownInfra(msg *nats.Msg) {
 	os.Exit(0)
 }

+// nbdkitPidGlob matches the pidfiles viperblockd hands nbdkit, one per volume.
+const nbdkitPidGlob = "nbdkit-vol-*.pid"
+
+// procComm and signalProcess are overridable so the cleanup below can be
+// tested without a real nbdkit to point at.
+var (
+	procComm = func(pid int) (string, error) {
+		data, err := os.ReadFile(fmt.Sprintf("/proc/%d/comm", pid))
+		if err != nil {
+			return "", err
+		}
+		return strings.TrimSpace(string(data)), nil
+	}
+	signalProcess = func(pid int, sig syscall.Signal) error {
+		return syscall.Kill(pid, sig)
+	}
+)
+
+// cleanupOrphanNBDKit terminates the nbdkit backends this node started, found
+// through the pidfiles in its runtime dir. A pattern kill was used here before
+// and reached every nbdkit on the host, including other nodes' live backends.
+func cleanupOrphanNBDKit(runtimeDir string) {
+	if runtimeDir == "" {
+		return
+	}
+
+	pidFiles, err := filepath.Glob(filepath.Join(runtimeDir, nbdkitPidGlob))
+	if err != nil {
+		slog.Warn("nbdkit cleanup: cannot list pid files", "dir", runtimeDir, "error", err)
+		return
+	}
+
+	for _, path := range pidFiles {
+		pid, err := readPidFileAt(path)
+		if err != nil {
+			slog.Debug("nbdkit cleanup: unreadable pid file, removing", "path", path, "error", err)
+			removeStalePidFile(path)
+			continue
+		}
+
+		// A pidfile outlives the process it names, so the PID may since have
+		// been recycled onto something unrelated. Only signal it while the
+		// kernel still agrees it is nbdkit.
+		comm, err := procComm(pid)
+		if err != nil || comm != "nbdkit" {
+			slog.Debug("nbdkit cleanup: pid is no longer nbdkit, skipping", "pid", pid, "comm", comm, "error", err)
+			removeStalePidFile(path)
+			continue
+		}
+
+		if err := signalProcess(pid, syscall.SIGTERM); err != nil {
+			slog.Warn("nbdkit cleanup: failed to signal backend", "pid", pid, "path", path, "error", err)
+			continue
+		}
+		slog.Info("nbdkit cleanup: terminated backend", "pid", pid, "path", path)
+		removeStalePidFile(path)
+	}
+}
+
+func readPidFileAt(path string) (int, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return 0, err
+	}
+	return strconv.Atoi(strings.TrimSpace(string(data)))
+}
+
+func removeStalePidFile(path string) {
+	if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
+		slog.Debug("nbdkit cleanup: could not remove pid file", "path", path, "error", err)
+	}
+}
+
 // respondShutdownACK marshals and sends a ShutdownACK response.
 func (d *Daemon) respondShutdownACK(msg *nats.Msg, ack ShutdownACK) {
 	data, err := json.Marshal(ack)
@@ -55,6 +55,43 @@ func TestShutdownHelpersHaveASingleCallSite(t *testing.T) {
 	}
 }

+// A pattern kill matches by command line across the whole host, so it reaches
+// processes this node never started: another node's backends on a shared host,
+// and a live guest's backend that survived a daemon restart by design. Cleanup
+// has to work from this node's own pidfiles instead.
+func TestNoPatternKillsInDaemonPackage(t *testing.T) {
+	banned := []string{"pkill", "killall"}
+
+	entries, err := os.ReadDir(".")
+	if err != nil {
+		t.Fatalf("read package dir: %v", err)
+	}
+
+	var offences []string
+	for _, entry := range entries {
+		name := entry.Name()
+		if entry.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") {
+			continue
+		}
+		data, rerr := os.ReadFile(name)
+		if rerr != nil {
+			t.Fatalf("read %s: %v", name, rerr)
+		}
+		for i, line := range strings.Split(string(data), "\n") {
+			for _, cmd := range banned {
+				if strings.Contains(line, `"`+cmd+`"`) {
+					offences = append(offences, fmt.Sprintf("  %s:%d: %s", name, i+1, strings.TrimSpace(line)))
+				}
+			}
+		}
+	}
+
+	if len(offences) > 0 {
+		t.Fatalf("pattern kills are host-wide and must not appear in the daemon package:\n%s",
+			strings.Join(offences, "\n"))
+	}
+}
+
 type callSite struct {
 	file   string
 	line   int
@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"sync"
+	"syscall"
 	"testing"
 	"time"

@@ -315,6 +316,221 @@ func TestHandleShutdownGate(t *testing.T) {
 	})
 }

+// TestShutdownRequestTarget covers the node scoping that keeps one node's
+// local drain off every other node: the phase subjects are fan-out, so a
+// request naming another node must be dropped before any service is stopped.
+func TestShutdownRequestTarget(t *testing.T) {
+	t.Run("request for another node is ignored with no ack", func(t *testing.T) {
+		daemon := createTestDaemon(t, sharedNATSURL)
+		daemon.config.Services = []string{"awsgw", "ui", "vpcd"}
+		pidDir := configurePidDir(t, daemon)
+
+		pid := startSleepProcess(t)
+		require.NoError(t, utils.WritePidFileTo(pidDir, "awsgw", pid))
+
+		subject := "spinifex.cluster.shutdown.gate.targeted-elsewhere"
+		sub, err := daemon.natsConn.Subscribe(subject, daemon.handleShutdownGate)
+		require.NoError(t, err)
+		defer sub.Unsubscribe()
+		require.NoError(t, daemon.natsConn.Flush())
+
+		payload, err := json.Marshal(ShutdownRequest{Phase: "gate", Target: "some-other-node"})
+		require.NoError(t, err)
+
+		_, err = daemon.natsConn.Request(subject, payload, 2*time.Second)
+		require.Error(t, err, "a request for another node must not be answered")
+
+		assert.False(t, daemon.shuttingDown.Load(), "another node's drain must not gate this daemon")
+		_, statErr := os.Stat(filepath.Join(pidDir, "awsgw.pid"))
+		assert.NoError(t, statErr, "another node's drain must not stop this node's services")
+	})
+
+	t.Run("request for this node is honoured", func(t *testing.T) {
+		daemon := createTestDaemon(t, sharedNATSURL)
+		daemon.config.Services = []string{}
+		configurePidDir(t, daemon)
+
+		subject := "spinifex.cluster.shutdown.gate.targeted-here"
+		sub, err := daemon.natsConn.Subscribe(subject, daemon.handleShutdownGate)
+		require.NoError(t, err)
+		defer sub.Unsubscribe()
+		require.NoError(t, daemon.natsConn.Flush())
+
+		payload, err := json.Marshal(ShutdownRequest{Phase: "gate", Target: daemon.node})
+		require.NoError(t, err)
+
+		reply, err := daemon.natsConn.Request(subject, payload, 30*time.Second)
+		require.NoError(t, err)
+
+		var ack ShutdownACK
+		require.NoError(t, json.Unmarshal(reply.Data, &ack))
+		assert.Equal(t, daemon.node, ack.Node)
+		assert.Empty(t, ack.Error)
+		assert.True(t, daemon.shuttingDown.Load())
+	})
+
+	// Every phase is a fan-out subject, so each handler needs the check —
+	// STORAGE and PERSIST stop another node's storage, and INFRA exits its
+	// daemon outright.
+	t.Run("every phase handler ignores another node's request", func(t *testing.T) {
+		daemon := createTestDaemon(t, sharedNATSURL)
+		daemon.config.Services = []string{}
+		configurePidDir(t, daemon)
+
+		for phase, handler := range map[string]nats.MsgHandler{
+			"drain":   daemon.handleShutdownDrain,
+			"storage": daemon.handleShutdownStorage,
+			"persist": daemon.handleShutdownPersist,
+			"infra":   daemon.handleShutdownInfra,
+		} {
+			t.Run(phase, func(t *testing.T) {
+				subject := "spinifex.cluster.shutdown." + phase + ".elsewhere"
+				sub, err := daemon.natsConn.Subscribe(subject, handler)
+				require.NoError(t, err)
+				defer sub.Unsubscribe()
+				require.NoError(t, daemon.natsConn.Flush())
+
+				payload, err := json.Marshal(ShutdownRequest{Phase: phase, Target: "some-other-node"})
+				require.NoError(t, err)
+
+				_, err = daemon.natsConn.Request(subject, payload, 2*time.Second)
+				require.Error(t, err, "%s answered a request addressed to another node", phase)
+			})
+		}
+	})
+
+	t.Run("untargeted request still reaches every node", func(t *testing.T) {
+		daemon := createTestDaemon(t, sharedNATSURL)
+		daemon.config.Services = []string{}
+		configurePidDir(t, daemon)
+
+		subject := "spinifex.cluster.shutdown.gate.untargeted"
+		sub, err := daemon.natsConn.Subscribe(subject, daemon.handleShutdownGate)
+		require.NoError(t, err)
+		defer sub.Unsubscribe()
+		require.NoError(t, daemon.natsConn.Flush())
+
+		payload, err := json.Marshal(ShutdownRequest{Phase: "gate"})
+		require.NoError(t, err)
+
+		reply, err := daemon.natsConn.Request(subject, payload, 30*time.Second)
+		require.NoError(t, err)
+
+		var ack ShutdownACK
+		require.NoError(t, json.Unmarshal(reply.Data, &ack))
+		assert.Empty(t, ack.Error)
+		assert.True(t, daemon.shuttingDown.Load(), "a cluster-wide shutdown must still gate this daemon")
+	})
+}
+
+// TestCleanupOrphanNBDKit covers the scoped replacement for the host-wide
+// pattern kill: only PIDs named by this node's own pidfiles are signalled,
+// and only while the kernel still agrees the PID is nbdkit.
+func TestCleanupOrphanNBDKit(t *testing.T) {
+	writePid := func(t *testing.T, dir, name string, contents string) string {
+		t.Helper()
+		path := filepath.Join(dir, name)
+		require.NoError(t, os.WriteFile(path, []byte(contents), 0o600))
+		return path
+	}
+
+	// stubProcTable replaces the /proc lookup and the kill with in-memory
+	// equivalents, returning the slice that records what was signalled.
+	stubProcTable := func(t *testing.T, comms map[int]string) *[]int {
+		t.Helper()
+		origComm, origSignal := procComm, signalProcess
+		t.Cleanup(func() { procComm, signalProcess = origComm, origSignal })
+
+		var signalled []int
+		procComm = func(pid int) (string, error) {
+			comm, ok := comms[pid]
+			if !ok {
+				return "", os.ErrNotExist
+			}
+			return comm, nil
+		}
+		signalProcess = func(pid int, _ syscall.Signal) error {
+			signalled = append(signalled, pid)
+			return nil
+		}
+		return &signalled
+	}
+
+	t.Run("signals only this node's live nbdkit pids", func(t *testing.T) {
+		dir := t.TempDir()
+		writePid(t, dir, "nbdkit-vol-vol-a.pid", "1001")
+		writePid(t, dir, "nbdkit-vol-vol-b.pid", "1002")
+		writePid(t, dir, "qemu-i-0123.pid", "1003")
+
+		signalled := stubProcTable(t, map[int]string{1001: "nbdkit", 1002: "nbdkit", 1003: "qemu-system-x86_64"})
+
+		cleanupOrphanNBDKit(dir)
+
+		assert.ElementsMatch(t, []int{1001, 1002}, *signalled, "only nbdkit pidfiles should be swept")
+		assert.FileExists(t, filepath.Join(dir, "qemu-i-0123.pid"), "unrelated pid files must be left alone")
+	})
+
+	t.Run("skips a recycled pid that is no longer nbdkit", func(t *testing.T) {
+		dir := t.TempDir()
+		writePid(t, dir, "nbdkit-vol-recycled.pid", "2001")
+
+		signalled := stubProcTable(t, map[int]string{2001: "postgres"})
+
+		cleanupOrphanNBDKit(dir)
+
+		assert.Empty(t, *signalled, "a pid whose comm is not nbdkit must never be signalled")
+	})
+
+	t.Run("removes pid files for dead and unparsable entries", func(t *testing.T) {
+		dir := t.TempDir()
+		dead := writePid(t, dir, "nbdkit-vol-dead.pid", "3001")
+		garbage := writePid(t, dir, "nbdkit-vol-garbage.pid", "not-a-pid")
+
+		signalled := stubProcTable(t, map[int]string{})
+
+		cleanupOrphanNBDKit(dir)
+
+		assert.Empty(t, *signalled)
+		assert.NoFileExists(t, dead)
+		assert.NoFileExists(t, garbage)
+	})
+
+	t.Run("empty runtime dir is a no-op", func(t *testing.T) {
+		signalled := stubProcTable(t, map[int]string{})
+		cleanupOrphanNBDKit("")
+		assert.Empty(t, *signalled)
+	})
+}
+
+// TestProcCommAndSignalProcess exercises the real /proc and kill paths the
+// cleanup stubs out everywhere else, so a change to either is caught here
+// rather than only on a node.
+func TestProcCommAndSignalProcess(t *testing.T) {
+	t.Run("procComm reads this process's own name", func(t *testing.T) {
+		comm, err := procComm(os.Getpid())
+		require.NoError(t, err)
+		assert.NotEmpty(t, comm)
+		assert.NotContains(t, comm, "\n", "comm must be trimmed")
+	})
+
+	t.Run("procComm reports a pid that does not exist", func(t *testing.T) {
+		// Above the default pid_max, so it cannot collide with a live process.
+		_, err := procComm(1 << 30)
+		require.Error(t, err)
+	})
+
+	t.Run("signalProcess reaches a live process", func(t *testing.T) {
+		pid := startSleepProcess(t)
+		// Signal 0 runs every permission check and delivers nothing, which is
+		// the reachability probe without disturbing the process.
+		require.NoError(t, signalProcess(pid, syscall.Signal(0)))
+	})
+
+	t.Run("signalProcess reports a pid that does not exist", func(t *testing.T) {
+		require.Error(t, signalProcess(1<<30, syscall.Signal(0)))
+	})
+}
+
 // TestHandleShutdownDrain covers the DRAIN phase handler: graceful StopAll,
 // shutdown marker, state persistence, and progress publishing.
 func TestHandleShutdownDrain(t *testing.T) {