feat: graceful guest drain on systemctl stop / host shutdown (#412)

* fix: drop removed predastore s3.Config.Compaction from admin_test

The predastore submodule bump removed the Compaction field from s3.Config.
Parse the rendered [compaction] block into a local struct so the template
round-trip test no longer references the deleted field.

* feat: graceful guest drain on systemctl stop / host shutdown

systemctl stop spinifex.target tore down nbdkit/viperblock while guest QEMU
was still writing, corrupting guest filesystems. The ordered drain existed
only over NATS via cluster shutdown, never on the systemd/host path.

Add spx admin node drain --local (GATE+DRAIN against the local node, leaving
STORAGE/PERSIST/INFRA to systemd), wired as the ExecStop of a new
spinifex-shutdown.service oneshot ordered After= all services so it stops
first and drains guests while every service is still up. DDIL preserved: the
oneshot fires only on target/host shutdown, not on a daemon restart.

Harden the viperblockd SIGTERM handler to flush the WAL but skip reaping an
nbdkit a guest is still attached to. Add a units ordering invariant.

* fix: ExecStop drain timeout needs duration unit (120 -> 120s)

The spinifex-shutdown.service ExecStop passed --timeout=120, but spx admin
node drain's --timeout is a cobra Duration flag and rejects a bare integer
(time: missing unit in duration "120"). The drain exited 1 without running,
so a systemctl stop tore nbdkit out from under the running guest and corrupted
its ext4 (the exact failure this feature prevents). Pass 120s.

* fix: nbdkitInUse read ss state from correct column + cover it

ss -H rows are '<netid> <state> ...', so the state is field[1], not field[0]
(field[0] is 'u_str'). The old index never matched ESTAB, so an in-use nbdkit
read as idle and the SIGTERM path would reap it under a live guest. Add tests
exercising a real connected unix socket, an absent socket, and TCP transport.

cmd/spinifex/cmd/admin.go

@@ -115,6 +115,23 @@ Each phase waits for all nodes to ACK before proceeding to the next.`,
 	Run: runClusterShutdown,
 }
 
+var nodeCmd = &cobra.Command{
+	Use:   "node",
+	Short: "Node-local operations",
+	Long:  `Node-local administrative operations such as a graceful local guest drain.`,
+}
+
+var nodeDrainCmd = &cobra.Command{
+	Use:   "drain",
+	Short: "Gracefully drain guests on the local node",
+	Long: `Run the GATE and DRAIN shutdown phases against the local node only: power down
+its guests via QMP and unmount their volumes (flushing the viperblock WAL) while
+every service is still running. STORAGE/PERSIST/INFRA are left to systemd's
+ordered unit teardown. This is the ExecStop of spinifex-shutdown.service, so a
+systemctl stop or host reboot drains guests before any storage service stops.`,
+	Run: runNodeDrainLocal,
+}
+
 var clusterDrainDHCPCmd = &cobra.Command{
 	Use:   "drain-dhcp",
 	Short: "Release all upstream DHCP leases held by vpcd",
@@ -258,6 +275,11 @@ func init() {
 	clusterCmd.AddCommand(clusterDrainDHCPCmd)
 	clusterDrainDHCPCmd.Flags().Duration("timeout", 30*time.Second, "Reply-collection window for vpcd drain responders")
 
+	adminCmd.AddCommand(nodeCmd)
+	nodeCmd.AddCommand(nodeDrainCmd)
+	nodeDrainCmd.Flags().Bool("local", false, "Drain the local node only (required)")
+	nodeDrainCmd.Flags().Duration("timeout", 120*time.Second, "Maximum time to wait per phase")
+
 	adminCmd.AddCommand(imagesCmd)
 	imagesCmd.AddCommand(imagesImportCmd)
 	imagesCmd.AddCommand(imagesListCmd)

cmd/spinifex/cmd/admin_test.go

@@ -9,7 +9,6 @@ import (
 	"strings"
 	"testing"
 
-	s3 "github.com/mulgadc/predastore/s3"
 	"github.com/mulgadc/spinifex/spinifex/admin"
 	"github.com/mulgadc/spinifex/spinifex/formation"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -503,7 +502,7 @@ func TestPredastoreTemplates_SetCompactionEmitsParseableBlock(t *testing.T) {
 		AccessKey: "AK", SecretKey: "SK", NatsToken: "tok", CompactionIntervalSeconds: interval,
 	})
 	assert.Contains(t, single, "[compaction]")
-	var singleCfg s3.Config
+	var singleCfg compactionConfig
 	require.NoError(t, toml.Unmarshal([]byte(single), &singleCfg))
 	assert.Equal(t, interval, singleCfg.Compaction.IntervalSeconds)
 
@@ -511,11 +510,20 @@ func TestPredastoreTemplates_SetCompactionEmitsParseableBlock(t *testing.T) {
 		predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", interval,
 	)
 	require.NoError(t, err)
-	var multiCfg s3.Config
+	var multiCfg compactionConfig
 	require.NoError(t, toml.Unmarshal([]byte(multi), &multiCfg))
 	assert.Equal(t, interval, multiCfg.Compaction.IntervalSeconds)
 }
 
+// compactionConfig parses just the rendered [compaction] block. Predastore's
+// s3.Config dropped its Compaction field, so this asserts the spinifex template
+// still emits a valid, parseable block carrying the interval.
+type compactionConfig struct {
+	Compaction struct {
+		IntervalSeconds int `toml:"interval_seconds"`
+	} `toml:"compaction"`
+}
+
 // The init/join flag must register with an unset (0) default and flow into the
 // rendered config via ConfigSettings.
 func TestAdminInitFlag_CompactionIntervalReachesRenderedConfig(t *testing.T) {

cmd/spinifex/cmd/cluster.go

@@ -161,6 +161,93 @@ func runClusterShutdown(cmd *cobra.Command, args []string) {
 	fmt.Printf("Cluster shutdown complete (%s)\n", time.Since(start).Round(time.Millisecond))
 }
 
+// runNodeDrainLocal runs the GATE and DRAIN shutdown phases against the local
+// node only, leaving STORAGE/PERSIST/INFRA to systemd's ordered unit teardown.
+// It is the ExecStop of spinifex-shutdown.service: guests are powered down and
+// their volumes unmounted while every service is still up, so the subsequent
+// service stops never tear storage out from under a running guest.
+func runNodeDrainLocal(cmd *cobra.Command, args []string) {
+	local, _ := cmd.Flags().GetBool("local")
+	timeout, _ := cmd.Flags().GetDuration("timeout")
+	if !local {
+		fmt.Fprintln(os.Stderr, "Error: node drain currently supports only --local")
+		os.Exit(1)
+	}
+
+	cfg, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	defer nc.Close()
+
+	node := cfg.Node
+	fmt.Printf("Draining local node %q (gate -> drain), timeout %s/phase\n", node, timeout)
+
+	for _, phase := range []string{"gate", "drain"} {
+		topic := "spinifex.cluster.shutdown." + phase
+		req := daemon.ShutdownRequest{Phase: phase, Timeout: int(timeout.Seconds())}
+		reqData, err := json.Marshal(req)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "Error marshaling request: %v\n", err)
+			os.Exit(1)
+		}
+
+		ack, err := collectLocalShutdownACK(nc, topic, reqData, node, timeout)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "[%s] %v\n", strings.ToUpper(phase), err)
+			os.Exit(1)
+		}
+		if ack.Error != "" {
+			fmt.Fprintf(os.Stderr, "[%s] %s: ERROR - %s\n", strings.ToUpper(phase), ack.Node, ack.Error)
+			os.Exit(1)
+		}
+		if len(ack.Stopped) > 0 {
+			fmt.Printf("[%s] %s: stopped %s\n", strings.ToUpper(phase), ack.Node, strings.Join(ack.Stopped, ", "))
+		} else {
+			fmt.Printf("[%s] %s: OK\n", strings.ToUpper(phase), ack.Node)
+		}
+	}
+
+	fmt.Printf("Local node %q drained; systemd may now stop storage services\n", node)
+}
+
+// collectLocalShutdownACK publishes a shutdown-phase request and returns the ACK
+// from the local node, ignoring ACKs from any other node. Returns an error if
+// the local node does not respond within timeout.
+func collectLocalShutdownACK(nc *nats.Conn, topic string, reqData []byte, node string, timeout time.Duration) (daemon.ShutdownACK, error) {
+	inbox := nats.NewInbox()
+	sub, err := nc.SubscribeSync(inbox)
+	if err != nil {
+		return daemon.ShutdownACK{}, fmt.Errorf("failed to subscribe to inbox: %w", err)
+	}
+	defer sub.Unsubscribe()
+
+	if err := nc.PublishRequest(topic, inbox, reqData); err != nil {
+		return daemon.ShutdownACK{}, fmt.Errorf("failed to publish request: %w", err)
+	}
+	nc.Flush()
+
+	deadline := time.Now().Add(timeout)
+	for {
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			return daemon.ShutdownACK{}, fmt.Errorf("timeout waiting for local node %q ACK", node)
+		}
+		msg, err := sub.NextMsg(remaining)
+		if err != nil {
+			return daemon.ShutdownACK{}, fmt.Errorf("timeout waiting for local node %q ACK", node)
+		}
+		var ack daemon.ShutdownACK
+		if err := json.Unmarshal(msg.Data, &ack); err != nil {
+			continue
+		}
+		if ack.Node == node {
+			return ack, nil
+		}
+	}
+}
+
 // runClusterDrainDHCP asks every vpcd to DHCPRELEASE all external-pool leases
 // it currently holds, returning them to the upstream DHCP server. Intended for
 // the teardown path: an env reset otherwise strands held leases until TTL