feat(eks): control-plane disaster recovery — etcd snapshot + restore-snapshot (#503)

* fix(eks): write etcd-snapshot.env at CP launch so snapshots actually run

* test(eks): cover resolveSystemPredastoreURL and buildEKSServiceDeps

Diff-coverage gate for 852f61d9 was unmet (48.8% < 60%) because the
mgmt-bridge branches and EKSServiceDeps assembly had no direct test.

* feat(eks): restore-snapshot DR path (fresh CP + cluster-reset restore + NLB re-point)

* fix(eks): make restore-snapshot fail-safe (validate snapshot, unwind on partial failure, honest status)

* fix(eks): sign etcd-snapshot uploads with deployment region (was 403ing)

* fix(eks): apply recovery directive on DR-seed first boot + single-node replacement host

* fix(daemon): log NATS handler failures at Error (were only on the trace span)

cmd/eks-gateway-fetch/main.go

@@ -156,9 +156,10 @@ func emitAddonsTSV(out io.Writer, body []byte) error {
 // recoveryDirective mirrors the gateway's RecoveryDirective wire shape; duplicated
 // here to keep this tiny VM binary free of the handler package's dependency graph.
 type recoveryDirective struct {
-	Epoch    int64  `json:"epoch"`
-	Action   string `json:"action"`
-	Snapshot string `json:"snapshot"`
+	Epoch            int64  `json:"epoch"`
+	Action           string `json:"action"`
+	Snapshot         string `json:"snapshot"`
+	SnapshotRequired bool   `json:"snapshotRequired"`
 }
 
 type internalRecoveryResponse struct {
@@ -166,8 +167,10 @@ type internalRecoveryResponse struct {
 }
 
 // emitRecoveryTSV decodes the recovery directive and writes a single
-// epoch\taction\tsnapshot line the on-VM k3s-recovery agent consumes without a
-// JSON parser. An unset action defaults to "none" (steady state).
+// epoch\taction\tsnapshot\tsnapshotRequired line the on-VM k3s-recovery agent
+// consumes without a JSON parser. An unset action defaults to "none" (steady
+// state); snapshotRequired is 1 when the snapshot MUST restore (fresh DR seed
+// aborts boot on fetch failure) else 0.
 func emitRecoveryTSV(out io.Writer, body []byte) error {
 	var resp internalRecoveryResponse
 	if err := json.Unmarshal(body, &resp); err != nil {
@@ -177,7 +180,11 @@ func emitRecoveryTSV(out io.Writer, body []byte) error {
 	if d.Action == "" {
 		d.Action = "none"
 	}
-	if _, err := fmt.Fprintf(out, "%d\t%s\t%s\n", d.Epoch, d.Action, d.Snapshot); err != nil {
+	required := "0"
+	if d.SnapshotRequired {
+		required = "1"
+	}
+	if _, err := fmt.Fprintf(out, "%d\t%s\t%s\t%s\n", d.Epoch, d.Action, d.Snapshot, required); err != nil {
 		return err
 	}
 	return nil

cmd/spinifex/cmd/admin_eks.go

@@ -0,0 +1,77 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/admin"
+	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
+	"github.com/spf13/cobra"
+)
+
+var adminEksCmd = &cobra.Command{
+	Use:   "eks",
+	Short: "Manage EKS control-plane disaster recovery",
+}
+
+var adminEksRestoreSnapshotCmd = &cobra.Command{
+	Use:   "restore-snapshot",
+	Short: "Rebuild a single-CP cluster's control plane from an etcd snapshot",
+	Long: `restore-snapshot drives the single-CP total-loss DR path: it launches a fresh
+control-plane VM as a cluster-init seed, directs its boot-time recovery agent to
+run k3s server --cluster-reset (restoring the given etcd snapshot, or the latest
+one found in predastore when --snapshot is omitted), and re-points the cluster
+NLB target groups at the new CP.
+
+Only single-CP clusters are supported. An HA cluster has a potentially
+surviving etcd quorum and should be recovered via quorum reformation instead
+of this destructive single-node rebuild.`,
+	Run: runAdminEksRestoreSnapshot,
+}
+
+func init() {
+	adminCmd.AddCommand(adminEksCmd)
+	adminEksCmd.AddCommand(adminEksRestoreSnapshotCmd)
+
+	adminEksRestoreSnapshotCmd.Flags().String("cluster", "", "Cluster name to restore (required)")
+	adminEksRestoreSnapshotCmd.Flags().String("snapshot", "", "Etcd snapshot key to restore; defaults to the latest snapshot found in predastore")
+	adminEksRestoreSnapshotCmd.Flags().String("account", "", "Account ID that owns the cluster; defaults to the bootstrap account")
+	_ = adminEksRestoreSnapshotCmd.MarkFlagRequired("cluster")
+}
+
+func runAdminEksRestoreSnapshot(cmd *cobra.Command, _ []string) {
+	clusterName, _ := cmd.Flags().GetString("cluster")
+	snapshot, _ := cmd.Flags().GetString("snapshot")
+	accountID, _ := cmd.Flags().GetString("account")
+	if accountID == "" {
+		accountID = admin.DefaultAccountID()
+	}
+
+	_, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	defer nc.Close()
+
+	svc := handlers_eks.NewNATSEKSService(nc)
+	// The server side launches a VM (30-60s) then fences the old CP with retries,
+	// so the RPC needs generous headroom over a bare launch.
+	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second)
+	defer cancel()
+
+	fmt.Printf("Restoring control plane for cluster %q (account %s)...\n", clusterName, accountID)
+	out, err := svc.RestoreSnapshot(ctx, &handlers_eks.RestoreSnapshotInput{
+		ClusterName: clusterName,
+		Snapshot:    snapshot,
+	}, accountID)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+
+	fmt.Printf("New control plane %s launched (snapshot %s)\n", out.NewInstanceID, out.Snapshot)
+	fmt.Printf("Status: %s\n", out.Status)
+}

docs/COMMANDS.md

@@ -171,6 +171,11 @@ read.
 | `spx admin gpu mig enable` | `--profile <name>` (required, e.g. `1g.10gb`), `--gpu <pci-addr>` (optional, default: all MIG-capable GPUs) | Checks no GPU instances running (NATS) → discovers MIG-capable GPUs via `gpu.Discover()` (filtered by `--gpu` if set) → enables MIG mode on each target (`gpu.EnableMIGMode`) → lists available profiles (`gpu.ListProfiles`) and validates requested profile name → destroys any existing instances (`gpu.DestroyAllInstances`) → creates new instances filling GPU capacity (`gpu.CreateInstances`) → writes `mig_profile` to `spinifex.toml` via `admin.SetMIGProfile` → sends SIGHUP to `spinifex-daemon`. Must be run directly on the target host. |
 | `spx admin gpu mig disable` | `--gpu <pci-addr>` (optional, default: all MIG-capable GPUs) | Checks no GPU instances running (NATS) → discovers MIG-capable GPUs via `gpu.Discover()` (filtered by `--gpu` if set) → destroys all GPU instances (`gpu.DestroyAllInstances`) → disables MIG mode (`gpu.DisableMIGMode`) → clears `mig_profile` in `spinifex.toml` via `admin.SetMIGProfile` → sends SIGHUP to `spinifex-daemon`. Must be run directly on the target host. |
 
+### EKS Control-Plane Disaster Recovery
+
+| Command | Flags | Description |
+|---------|-------|-------------|
+| `spx admin eks restore-snapshot` | `--cluster` (required), `--snapshot` (optional, defaults to the latest snapshot in predastore), `--account` (optional, defaults to the bootstrap account) | Single-CP total-loss DR path (fail-safe): validates the snapshot exists in predastore BEFORE any mutation (a typo'd/missing key hard-fails, never resets into an empty datastore) → launches a fresh control-plane VM as a cluster-init seed (replaying the persisted create-time launch template) → sets a required-snapshot `RecoveryDirective` (`cluster-reset`) so the boot-time recovery agent aborts rather than resets-into-empty if it cannot fetch the snapshot → persists the replacement in cluster meta BEFORE re-pointing the NLB (so an NLB failure is convergeable by the reconciler, returned as a provisional status, not a hard error) → re-points the cluster NLB's apiserver and konnectivity target groups from the old CP's ENI to the new one → fences the old CP with retries, failing loudly if it cannot be confirmed terminated (split-brain guard). Any failure before the meta commit unwinds the fresh CP (terminate + clear directive) so a re-run does not stack a second resetting control plane. The returned status is provisional — success means the sequence completed, not that etcd is restored and serving; verify cluster health. HA clusters (a spread with a potentially surviving quorum) are rejected — recover those via quorum reformation instead. |
 
 ## AWS-Compatible API
 

scripts/images/eks-node/eks-node-role.sh

@@ -48,14 +48,26 @@ if [ -z "${ROLE}" ] && [ -f "${AGENT_ENVFILE}" ]; then
     ROLE="agent"
 fi
 
+case "${ROLE}" in
+    server | server-join | agent) ;;
+    *)
+        die "SPINIFEX_K3S_ROLE missing or invalid: '${ROLE}'"
+        ;;
+esac
+
+# Written before the rc-service calls below: mulga-eks-k3s-recovery, started
+# inline for server/server-join, reads ROLE_FILE at the top of its own logic
+# and must find it already in place, not empty.
+mkdir -p "$(dirname "${ROLE_FILE}")"
+printf '%s\n' "${ROLE}" > "${ROLE_FILE}"
+
 case "${ROLE}" in
     server)
         log "configuring server role"
         rc-update add eks-token-webhook default
-        # Pre-k3s etcd recovery: added to the runlevel so it runs before k3s on
-        # every subsequent boot, but not started now — a fresh cluster has no
-        # directive and the role file it guards on is written at the end of this
-        # script. It first acts on the reboot the reconciler triggers.
+        # Pre-k3s etcd recovery: enabled for later boots and started inline
+        # now too, so a restore-snapshot DR seed's directive applies before
+        # k3s starts on this first boot. A plain create has no directive, so it's a no-op.
         rc-update add mulga-eks-k3s-recovery default
         rc-update add k3s default
         # konnectivity-server fronts apiserver egress; every server replica runs
@@ -69,6 +81,7 @@ case "${ROLE}" in
         # (mulga-siv-231.7); server-join nodes do not run it.
         rc-update add mulga-eks-addon-sync default
         rc-service eks-token-webhook start
+        rc-service mulga-eks-k3s-recovery start
         rc-service k3s start
         rc-service konnectivity-server start
         rc-service k3s-first-boot start
@@ -79,12 +92,14 @@ case "${ROLE}" in
         log "configuring server-join role"
         rc-update add eks-token-webhook default
         # Pre-k3s etcd recovery (see server role): server-join members are the
-        # wipe-rejoin followers of a cluster-reset, so they need it too.
+        # wipe-rejoin followers of a cluster-reset, so they need it too, and for
+        # the same first-boot reason must be started here, not deferred.
         rc-update add mulga-eks-k3s-recovery default
         rc-update add k3s default
         rc-update add konnectivity-server default
         rc-update add mulga-eks-state-report default
         rc-service eks-token-webhook start
+        rc-service mulga-eks-k3s-recovery start
         rc-service k3s start
         rc-service konnectivity-server start
         rc-service mulga-eks-state-report start
@@ -99,7 +114,5 @@ case "${ROLE}" in
         ;;
 esac
 
-mkdir -p "$(dirname "${ROLE_FILE}")"
-printf '%s\n' "${ROLE}" > "${ROLE_FILE}"
 rc-update del eks-node-role default 2>/dev/null || true
 log "role '${ROLE}' configured"

scripts/images/eks-node/eks-node-role_test.sh

@@ -65,8 +65,11 @@ check "server: enables k3s-first-boot" "yes" "$(grep -q 'rc-update add k3s-first
 check "server: starts k3s" "yes" "$(grep -q 'rc-service k3s start' "${CALLS}" && echo yes || echo no)"
 check "server: enables konnectivity-server" "yes" "$(grep -q 'rc-update add konnectivity-server default' "${CALLS}" && echo yes || echo no)"
 check "server: enables k3s-recovery" "yes" "$(grep -q 'rc-update add mulga-eks-k3s-recovery default' "${CALLS}" && echo yes || echo no)"
-# Recovery is added to the runlevel (pre-k3s on later boots) but not started now.
-check "server: does not start k3s-recovery" "no" "$(grep -q 'rc-service mulga-eks-k3s-recovery' "${CALLS}" && echo yes || echo no)"
+# Recovery is added to the runlevel (so it also runs pre-k3s on later boots)
+# AND started inline now — a restore-snapshot DR seed's directive must apply
+# on this very first boot, before k3s starts.
+check "server: starts k3s-recovery" "yes" "$(grep -q 'rc-service mulga-eks-k3s-recovery start' "${CALLS}" && echo yes || echo no)"
+check "server: starts k3s-recovery before k3s" "yes" "$(awk '/rc-service mulga-eks-k3s-recovery start/{r=NR} /rc-service k3s start/{k=NR} END{print (r>0 && k>0 && r<k) ? "yes" : "no"}' "${CALLS}")"
 check "server: no k3s-agent" "no" "$(grep -q 'k3s-agent' "${CALLS}" && echo yes || echo no)"
 check "server: self-disables" "yes" "$(grep -q 'rc-update del eks-node-role default' "${CALLS}" && echo yes || echo no)"
 
@@ -81,6 +84,7 @@ check "server-join: enables k3s" "yes" "$(grep -q 'rc-update add k3s default' "$
 check "server-join: enables state-report" "yes" "$(grep -q 'rc-update add mulga-eks-state-report default' "${CALLS}" && echo yes || echo no)"
 check "server-join: enables konnectivity-server" "yes" "$(grep -q 'rc-update add konnectivity-server default' "${CALLS}" && echo yes || echo no)"
 check "server-join: enables k3s-recovery" "yes" "$(grep -q 'rc-update add mulga-eks-k3s-recovery default' "${CALLS}" && echo yes || echo no)"
+check "server-join: starts k3s-recovery" "yes" "$(grep -q 'rc-service mulga-eks-k3s-recovery start' "${CALLS}" && echo yes || echo no)"
 # Join servers must NOT re-publish bootstrap; the first server already did.
 check "server-join: no k3s-first-boot" "no" "$(grep -q 'k3s-first-boot' "${CALLS}" && echo yes || echo no)"
 check "server-join: no k3s-agent" "no" "$(grep -q 'k3s-agent' "${CALLS}" && echo yes || echo no)"

scripts/images/eks-node/k3s.initd

@@ -28,6 +28,16 @@ depend() {
 }
 
 start_pre() {
+    # mulga-eks-k3s-recovery drops this marker when a required-snapshot restore
+    # could not fetch its etcd snapshot. Refuse to start: a fresh DR seed with no
+    # snapshot would cluster-init an EMPTY datastore (silent total data loss). The
+    # marker is in /run tmpfs, cleared on reboot, so recovery retries the fetch
+    # next boot and k3s starts once the snapshot restores.
+    if [ -f /run/spinifex-eks/k3s-restore-block ]; then
+        eerror "restore-snapshot could not fetch the required etcd snapshot; refusing to start k3s on an empty datastore (see mulga-eks-k3s-recovery log)"
+        return 1
+    fi
+
     # config.yaml is written by k3s-first-boot from the skeleton + cloud-init
     # data. Refuse to start if neither exists.
     if [ ! -f /etc/rancher/k3s/config.yaml ]; then

scripts/images/eks-node/mulga-eks-etcd-snapshot.sh

@@ -66,14 +66,24 @@ fi
 #
 # Note: --aws-sigv4 requires curl >= 7.75 (Alpine 3.21 ships 8.x).
 URL="${SPINIFEX_PREDASTORE_ENDPOINT%/}/${BUCKET}/${KEY}"
-curl -fsSL \
+# Capture curl's own exit: piping straight into logger masks it behind logger's
+# (near-always 0), so a failed upload would log "uploaded" and delete the only
+# copy — snapshots would silently never land. Keep the local snapshot and exit
+# non-zero on failure so cron surfaces it and the next run still has data on disk.
+if UPLOAD_OUT=$(curl -fsSL \
     --aws-sigv4 "aws:amz:${AWS_REGION:-au-mel-1}:s3" \
     --user "${SPINIFEX_PREDASTORE_AKID}:${SPINIFEX_PREDASTORE_SECRET}" \
     -H "x-amz-content-sha256: UNSIGNED-PAYLOAD" \
     --upload-file "${SNAPSHOT_FILE}" \
-    "${URL}" 2>&1 | logger -t mulga-eks-etcd-snapshot
-
-# Remove the local snapshot to bound disk usage; the canonical copy lives in
-# predastore. Restore path (`spx admin eks restore-snapshot`) pulls from there.
-rm -f "${SNAPSHOT_FILE}"
-logger -t mulga-eks-etcd-snapshot "uploaded ${KEY}"
+    "${URL}" 2>&1); then
+    [ -n "${UPLOAD_OUT}" ] && printf '%s\n' "${UPLOAD_OUT}" | logger -t mulga-eks-etcd-snapshot
+    # Remove the local snapshot only after a confirmed upload to bound disk usage;
+    # the canonical copy lives in predastore (`spx admin eks restore-snapshot`).
+    rm -f "${SNAPSHOT_FILE}"
+    logger -t mulga-eks-etcd-snapshot "uploaded ${KEY}"
+else
+    UPLOAD_RC=$?
+    [ -n "${UPLOAD_OUT}" ] && printf '%s\n' "${UPLOAD_OUT}" | logger -t mulga-eks-etcd-snapshot
+    logger -t mulga-eks-etcd-snapshot "ERROR: upload of ${KEY} failed (curl rc=${UPLOAD_RC}); keeping local snapshot ${SNAPSHOT_FILE}"
+    exit 1
+fi

scripts/images/eks-node/mulga-eks-etcd-snapshot_test.sh

@@ -29,10 +29,13 @@ done
 :
 EOF
 
-# curl stub: record the full argument line (carries --upload-file + URL).
+# curl stub: record the full argument line (carries --upload-file + URL). Exits
+# non-zero when CURL_FAIL=1 so the harness can exercise the failed-upload path.
 cat > "${STUBBIN}/curl" <<'EOF'
 #!/bin/sh
 echo "curl $*" >> "${CURL_CALLS}"
+[ "${CURL_FAIL:-0}" = "1" ] && { echo "curl: (7) connection refused" >&2; exit 7; }
+exit 0
 EOF
 
 # logger stub: swallow.
@@ -102,6 +105,24 @@ run server "${WORK}/15min/mulga-eks-etcd-snapshot"
 [ -s "${WORK}/curl.calls" ] && fail "no-env: must not upload" || pass "no-env: no upload without creds"
 ENVFILE="${ENVFILE_SAVE}"
 
+# --- Case 5: upload FAILS -> non-zero exit + local snapshot KEPT (not masked) ---
+# Set up like run() but expect a non-zero exit and assert the snapshot survives.
+K3S_CALLS="${WORK}/k3s.calls"; CURL_CALLS="${WORK}/curl.calls"
+SNAPSHOT_DIR="${WORK}/snapshots"; ROLE_FILE="${WORK}/role"; K3S_BIN=k3s
+: > "${K3S_CALLS}"; : > "${CURL_CALLS}"; rm -rf "${SNAPSHOT_DIR}"
+printf '%s' server > "${ROLE_FILE}"
+export K3S_CALLS CURL_CALLS SNAPSHOT_DIR ROLE_FILE K3S_BIN ENVFILE
+export TIER=frequent CURL_FAIL=1
+if sh "${WORK}/15min/mulga-eks-etcd-snapshot" </dev/null; then
+    fail "upload-fail: must exit non-zero when curl fails"
+else
+    pass "upload-fail: non-zero exit surfaces the failure"
+fi
+ls "${SNAPSHOT_DIR}"/etcd-frequent-* >/dev/null 2>&1 \
+    && pass "upload-fail: local snapshot kept for the next run" \
+    || fail "upload-fail: local snapshot deleted despite failed upload"
+unset TIER CURL_FAIL
+
 if [ "${FAILS}" -eq 0 ]; then
     echo "PASS: all mulga-eks-etcd-snapshot cases"
     exit 0

scripts/images/eks-node/mulga-eks-k3s-recovery.sh

@@ -7,10 +7,10 @@ set -eu
 # but embedded etcd never reformed quorum after a simultaneous restart).
 #
 # The reconciler cannot push into the guest, so recovery is pull-based: this
-# fetches a per-member directive epoch<TAB>action<TAB>snapshot through the AWS
-# gateway (eks-gateway-fetch -resource recovery), epoch-guarded so a directive
-# applies exactly once. Ordered `before k3s` so the etcd datastore is mutated
-# while the apiserver is stopped. Actions:
+# fetches a per-member directive epoch<TAB>action<TAB>snapshot<TAB>required
+# through the AWS gateway (eks-gateway-fetch -resource recovery), epoch-guarded
+# so a directive applies exactly once. Ordered `before k3s` so the etcd datastore
+# is mutated while the apiserver is stopped. Actions:
 #   none          steady state — nothing to do
 #   cluster-reset seed member: `k3s server --cluster-reset` reforms a
 #                 single-member etcd from this node's intact data (optionally
@@ -18,6 +18,13 @@ set -eu
 #   wipe-rejoin   follower: delete the local etcd datastore so k3s re-joins the
 #                 reset seed's quorum fresh on the next k3s start
 #
+# required=1 marks a snapshot that MUST restore (fresh DR seed via
+# `spx admin eks restore-snapshot`): a fresh seed has no intact local etcd, so a
+# cluster-reset without the snapshot would boot an EMPTY cluster. On a fetch
+# failure this drops BLOCK_MARKER and aborts boot; k3s.initd refuses to start on
+# the marker, so k3s never resets into an empty datastore. The fetch is retried
+# on the next boot (BLOCK_MARKER lives in /run tmpfs, cleared on reboot).
+#
 # Env knobs default to production paths; overridable so the logic is unit-testable
 # without root, IMDS, a gateway, or a real k3s.
 
@@ -25,6 +32,7 @@ ROLE_FILE=${ROLE_FILE:-/etc/spinifex-eks/role}
 ENVFILE=${ENVFILE:-/etc/spinifex-eks/first-boot.env}
 SNAPSHOT_ENVFILE=${SNAPSHOT_ENVFILE:-/etc/spinifex-eks/etcd-snapshot.env}
 EPOCH_FILE=${EPOCH_FILE:-/var/lib/spinifex-eks/recovery.epoch}
+BLOCK_MARKER=${BLOCK_MARKER:-/run/spinifex-eks/k3s-restore-block}
 K3S_BIN=${K3S_BIN:-/usr/local/bin/k3s}
 K3S_CONFIG=${K3S_CONFIG:-/etc/rancher/k3s/config.yaml}
 ETCD_DIR=${ETCD_DIR:-/var/lib/rancher/k3s/server/db/etcd}
@@ -72,12 +80,26 @@ fetch_snapshot() {
     [ -n "${SPINIFEX_PREDASTORE_SECRET:-}" ] || return 1
     _url="${SPINIFEX_PREDASTORE_ENDPOINT%/}/eks-backups-system/${EKS_ACCOUNT_ID}/${EKS_CLUSTER_NAME}/${_name}"
     mkdir -p "${SNAPSHOT_DIR}"
-    curl -fsSL \
+    # Capture curl's own exit — piping straight into logger would mask it behind
+    # logger's (near-always 0), so a failed download would look like success and
+    # the caller would wrongly cluster-reset from local etcd.
+    if _out=$(curl -fsSL \
         --aws-sigv4 "aws:amz:${AWS_REGION:-au-mel-1}:s3" \
         --user "${SPINIFEX_PREDASTORE_AKID}:${SPINIFEX_PREDASTORE_SECRET}" \
         -H "x-amz-content-sha256: UNSIGNED-PAYLOAD" \
         -o "${SNAPSHOT_DIR}/${_name}" \
-        "${_url}" 2>&1 | logger -t "${LOGTAG}"
+        "${_url}" 2>&1); then
+        [ -n "${_out}" ] && printf '%s\n' "${_out}" | logger -t "${LOGTAG}"
+        return 0
+    else
+        # Capture rc as the first else statement: $? read after `fi` would be the
+        # if's own 0, re-masking the failure we are here to surface.
+        _rc=$?
+        [ -n "${_out}" ] && printf '%s\n' "${_out}" | logger -t "${LOGTAG}"
+        rm -f "${SNAPSHOT_DIR}/${_name}"
+        log "snapshot ${_name} download failed (curl rc=${_rc})"
+        return 1
+    fi
 }
 
 # Instance ID from IMDS keys the per-member directive. IMDSv2 token if enforced,
@@ -109,8 +131,11 @@ fi
 epoch=$(printf '%s' "${line}" | cut -f1)
 action=$(printf '%s' "${line}" | cut -f2)
 snapshot=$(printf '%s' "${line}" | cut -f3)
+snapshot_required=$(printf '%s' "${line}" | cut -f4)
 [ -n "${action}" ] || action=none
 case "${epoch}" in '' | *[!0-9]*) epoch=0 ;; esac
+# Absent 4th field (older gateway) reads as not-required, preserving HA-reform semantics.
+case "${snapshot_required}" in 1) snapshot_required=1 ;; *) snapshot_required=0 ;; esac
 
 if [ "${action}" = "none" ]; then
     log "directive action=none (epoch ${epoch}); nothing to do"
@@ -136,6 +161,13 @@ case "${action}" in
             if fetch_snapshot "${snapshot}"; then
                 restore_arg="--cluster-reset-restore-path=${SNAPSHOT_DIR}/${snapshot}"
                 log "cluster-reset restoring snapshot ${snapshot}"
+            elif [ "${snapshot_required}" = "1" ]; then
+                # Fresh DR seed: no intact local etcd, so a reset without the snapshot
+                # would boot an EMPTY cluster. Block k3s and abort; retry next boot.
+                log "ERROR: required snapshot ${snapshot} could not be fetched; aborting boot to avoid resetting into an EMPTY datastore (retries next boot)"
+                mkdir -p "$(dirname "${BLOCK_MARKER}")"
+                : > "${BLOCK_MARKER}"
+                exit 1
             else
                 log "WARN: snapshot ${snapshot} fetch failed; cluster-reset from local etcd only"
             fi

scripts/images/eks-node/mulga-eks-k3s-recovery_test.sh

@@ -24,7 +24,9 @@ printf '%s\n' "${DIRECTIVE_LINE}"
 EOF
 
 # curl stub: answers the IMDS token PUT + instance-id GET, and materialises a
-# snapshot file for any `-o <path>` download. Records every call.
+# snapshot file for any `-o <path>` download. Records every call. A snapshot
+# download (an `-o` arg) fails with a non-zero exit when CURL_SNAP_FAIL=1, so the
+# harness can exercise the fetch-failure path.
 cat > "${STUBBIN}/curl" <<'EOF'
 #!/bin/sh
 echo "curl $*" >> "${CURL_CALLS}"
@@ -36,7 +38,10 @@ for a in "$@"; do
 done
 prev=""
 for a in "$@"; do
-    [ "$prev" = "-o" ] && echo snapdata > "$a"
+    if [ "$prev" = "-o" ]; then
+        [ "${CURL_SNAP_FAIL:-0}" = "1" ] && { echo "curl: (22) 404" >&2; exit 22; }
+        echo snapdata > "$a"
+    fi
     prev="$a"
 done
 exit 0
@@ -80,7 +85,7 @@ pass() { echo "ok: $*"; }
 
 # run <role> <directive-line>: reset capture files + state, then run the script
 # with every path knob pointed into the temp dir.
-run() {
+_setup_run() {
     _role=$1
     DIRECTIVE_LINE=$2
     K3S_CALLS="${WORK}/k3s.calls"
@@ -89,13 +94,29 @@ run() {
     EPOCH_FILE="${WORK}/recovery.epoch"
     ETCD_DIR="${WORK}/etcd"
     SNAPSHOT_DIR="${WORK}/snapshots"
+    BLOCK_MARKER="${WORK}/run/k3s-restore-block"
     : > "${K3S_CALLS}"
     : > "${CURL_CALLS}"
+    rm -f "${BLOCK_MARKER}"
     printf '%s' "${_role}" > "${ROLE_FILE}"
-    export DIRECTIVE_LINE K3S_CALLS CURL_CALLS ROLE_FILE EPOCH_FILE ETCD_DIR SNAPSHOT_DIR
+    export DIRECTIVE_LINE K3S_CALLS CURL_CALLS ROLE_FILE EPOCH_FILE ETCD_DIR SNAPSHOT_DIR BLOCK_MARKER
     export ENVFILE SNAPSHOT_ENVFILE
     export K3S_BIN=k3s FETCH_BIN=eks-gateway-fetch
-    sh "${SCRIPT}" </dev/null || fail "role=${_role}: non-zero exit"
+}
+
+# run <role> <directive-line>: expects a zero exit (the steady-state path).
+run() {
+    _setup_run "$1" "$2"
+    sh "${SCRIPT}" </dev/null || fail "role=${1}: non-zero exit"
+}
+
+# run_abort <role> <directive-line>: expects a NON-zero exit (the required-snapshot
+# fetch-failure abort). CURL_SNAP_FAIL must be set by the caller.
+run_abort() {
+    _setup_run "$1" "$2"
+    if sh "${SCRIPT}" </dev/null; then
+        fail "role=${1}: expected non-zero exit (required-snapshot abort)"
+    fi
 }
 
 TAB=$(printf '\t')
@@ -149,6 +170,36 @@ grep -q 'aws-sigv4' "${WORK}/curl.calls" \
 [ -f "${WORK}/snapshots/etcd-daily-20260706T000000Z.snap" ] \
     && pass "snapshot: fetched into snapshot dir" || fail "snapshot: file not materialised"
 
+# --- Case 7: REQUIRED snapshot, fetch OK -> restore-path + no block marker ---
+rm -f "${WORK}/recovery.epoch"
+CURL_SNAP_FAIL=0; export CURL_SNAP_FAIL
+run server "5${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}1"
+grep -q 'cluster-reset-restore-path=.*etcd-frequent-20260709T010000Z.snap' "${WORK}/k3s.calls" \
+    && pass "required-ok: restore-path passed" || fail "required-ok: no restore-path: $(cat "${WORK}/k3s.calls")"
+[ -f "${WORK}/run/k3s-restore-block" ] && fail "required-ok: must not block k3s on success" || pass "required-ok: no block marker"
+
+# --- Case 8: REQUIRED snapshot, fetch FAILS -> abort: block marker, no k3s, no epoch ---
+rm -f "${WORK}/recovery.epoch"
+CURL_SNAP_FAIL=1; export CURL_SNAP_FAIL
+run_abort server "6${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}1"
+[ -f "${WORK}/run/k3s-restore-block" ] \
+    && pass "required-fail: block marker dropped to stop k3s" || fail "required-fail: no block marker"
+[ -s "${WORK}/k3s.calls" ] && fail "required-fail: must NOT cluster-reset into empty datastore" || pass "required-fail: k3s not reset"
+[ -f "${WORK}/recovery.epoch" ] && fail "required-fail: epoch must not record (retry next boot)" || pass "required-fail: epoch not recorded"
+
+# --- Case 9: NON-required snapshot, fetch FAILS -> local reset, no block, epoch recorded ---
+rm -f "${WORK}/recovery.epoch"
+CURL_SNAP_FAIL=1; export CURL_SNAP_FAIL
+run server "7${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}0"
+grep -q 'server --config .* --cluster-reset' "${WORK}/k3s.calls" \
+    && pass "optional-fail: falls back to local cluster-reset" || fail "optional-fail: no reset: $(cat "${WORK}/k3s.calls")"
+grep -q 'cluster-reset-restore-path' "${WORK}/k3s.calls" \
+    && fail "optional-fail: must not pass a restore-path for a failed fetch" || pass "optional-fail: local-data reset only"
+[ -f "${WORK}/run/k3s-restore-block" ] && fail "optional-fail: must not block k3s (HA-reform semantics)" || pass "optional-fail: no block marker"
+[ "$(cat "${WORK}/recovery.epoch" 2>/dev/null)" = 7 ] \
+    && pass "optional-fail: epoch recorded" || fail "optional-fail: epoch not recorded"
+CURL_SNAP_FAIL=0; export CURL_SNAP_FAIL
+
 if [ "${FAILS}" -eq 0 ]; then
     echo "PASS: all mulga-eks-k3s-recovery cases"
     exit 0

spinifex/daemon/daemon.go

@@ -975,6 +975,7 @@ func (d *Daemon) subscribeAll() error {
 			natsSub{"eks.ListStagedAddonManifests", d.handleEKSListStagedAddonManifests, "spinifex-workers"},
 			natsSub{"eks.GetRecoveryDirective", d.handleEKSGetRecoveryDirective, "spinifex-workers"},
 			natsSub{"eks.SetRecoveryDirective", d.handleEKSSetRecoveryDirective, "spinifex-workers"},
+			natsSub{"eks.RestoreSnapshot", d.handleEKSRestoreSnapshot, "spinifex-workers"},
 			natsSub{"eks.AssociateIdentityProviderConfig", d.handleEKSAssociateIdentityProviderConfig, "spinifex-workers"},
 			natsSub{"eks.DescribeIdentityProviderConfig", d.handleEKSDescribeIdentityProviderConfig, "spinifex-workers"},
 			natsSub{"eks.ListIdentityProviderConfigs", d.handleEKSListIdentityProviderConfigs, "spinifex-workers"},

spinifex/daemon/daemon_handlers.go

@@ -84,6 +84,10 @@ func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(context.Conte
 	}
 	output, err := serviceFn(ctx, input, accountID)
 	if err != nil {
+		// The error was otherwise only recorded on the OTel span, invisible
+		// without a trace backend. Log at Error so handler failures show up
+		// in the journal too.
+		slog.ErrorContext(ctx, "handleNATSRequest: service call failed", "subject", msg.Subject, "err", err)
 		utils.MarkSpanError(span, err)
 		respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
 		return

spinifex/daemon/daemon_handlers_eks.go

@@ -130,6 +130,10 @@ func (d *Daemon) handleEKSSetRecoveryDirective(msg *nats.Msg) {
 	handleNATSRequest(msg, d.eksService.SetRecoveryDirective)
 }
 
+func (d *Daemon) handleEKSRestoreSnapshot(msg *nats.Msg) {
+	handleNATSRequest(msg, d.eksService.RestoreSnapshot)
+}
+
 // --- OIDC identity-provider configs ---
 
 func (d *Daemon) handleEKSAssociateIdentityProviderConfig(msg *nats.Msg) {

spinifex/daemon/eks_deps.go

@@ -8,6 +8,7 @@ import (
 	handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
 	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"log/slog"
 )
 
@@ -74,33 +75,37 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps {
 	}
 
 	deps := handlers_eks.EKSServiceDeps{
-		Config:           d.config,
-		NATSConn:         d.natsConn,
-		MasterKey:        masterKey,
-		GatewayBaseURL:   d.resolveGatewayBaseURL(),
-		Region:           d.config.Region,
-		HolderID:         d.node,
-		ClusterSize:      clusterSize,
-		InternalSuffix:   internalSuffix,
-		SystemGatewayURL: d.resolveSystemGatewayBaseURL(),
-		SystemAccessKey:  d.config.Predastore.AccessKey,
-		SystemSecretKey:  d.config.Predastore.SecretKey,
-		GatewayCACert:    gatewayCA,
-		VPCSG:            d.vpcService,
-		VPCK3s:           d.vpcService,
-		VPCSubnet:        &daemonEKSSubnetResolver{d: d},
-		NLB:              d.elbv2Service,
-		Instance:         d,
-		Image:            d.imageService,
-		EIP:              d.eipService,
-		IGW:              d.igwService,
-		Worker:           d,
-		VPCMgr:           d.vpcService,
-		NATGW:            d.natGatewayService,
-		RouteTable:       d.routeTableService,
-		PlacementGroup:   handlers_ec2_placementgroup.NewNATSPlacementGroupService(d.natsConn),
-		Scheduler:        handlers_eks.NewNATSHostScheduler(d.natsConn),
-		CPControl:        d.newEKSCPControl(),
+		Config:              d.config,
+		NATSConn:            d.natsConn,
+		MasterKey:           masterKey,
+		GatewayBaseURL:      d.resolveGatewayBaseURL(),
+		Region:              d.config.Region,
+		HolderID:            d.node,
+		ClusterSize:         clusterSize,
+		InternalSuffix:      internalSuffix,
+		SystemGatewayURL:    d.resolveSystemGatewayBaseURL(),
+		SystemAccessKey:     d.config.Predastore.AccessKey,
+		SystemSecretKey:     d.config.Predastore.SecretKey,
+		GatewayCACert:       gatewayCA,
+		SystemPredastoreURL: d.resolveSystemPredastoreURL(),
+		SnapshotStore: objectstore.NewS3ObjectStoreFromConfig(
+			d.config.Predastore.Host, d.config.Predastore.Region,
+			d.config.Predastore.AccessKey, d.config.Predastore.SecretKey),
+		VPCSG:          d.vpcService,
+		VPCK3s:         d.vpcService,
+		VPCSubnet:      &daemonEKSSubnetResolver{d: d},
+		NLB:            d.elbv2Service,
+		Instance:       d,
+		Image:          d.imageService,
+		EIP:            d.eipService,
+		IGW:            d.igwService,
+		Worker:         d,
+		VPCMgr:         d.vpcService,
+		NATGW:          d.natGatewayService,
+		RouteTable:     d.routeTableService,
+		PlacementGroup: handlers_ec2_placementgroup.NewNATSPlacementGroupService(d.natsConn),
+		Scheduler:      handlers_eks.NewNATSHostScheduler(d.natsConn),
+		CPControl:      d.newEKSCPControl(),
 	}
 
 	// A KV-backed IAM service (sharing the gateway's buckets over NATS) lets EKS
@@ -176,3 +181,25 @@ func (d *Daemon) resolveSystemGatewayBaseURL() string {
 	}
 	return "https://" + net.JoinHostPort(d.mgmtBridgeIP, gatewayPort)
 }
+
+// resolveSystemPredastoreURL derives the predastore URL a system instance (the
+// K3s server VM) fetches/PUTs etcd snapshots to. Same mgmt-bridge-only
+// reachability constraint as resolveSystemGatewayBaseURL: config.Predastore.Host
+// is rewritten to 127.0.0.1 for the local daemon, which a guest cannot dial, so
+// the bridge IP is substituted with the configured port. Falls back to the
+// configured Predastore.Host (scheme-prefixed) when no mgmt bridge exists.
+func (d *Daemon) resolveSystemPredastoreURL() string {
+	predastorePort := "8443"
+	if d.config.Predastore.Host != "" {
+		if _, port, splitErr := net.SplitHostPort(d.config.Predastore.Host); splitErr == nil && port != "" {
+			predastorePort = port
+		}
+	}
+	if d.mgmtBridgeIP == "" {
+		if d.config.Predastore.Host == "" {
+			return ""
+		}
+		return "https://" + d.config.Predastore.Host
+	}
+	return "https://" + net.JoinHostPort(d.mgmtBridgeIP, predastorePort)
+}

spinifex/daemon/eks_deps_test.go

@@ -84,3 +84,74 @@ func TestResolveGatewayHost(t *testing.T) {
 		})
 	}
 }
+
+// TestResolveSystemPredastoreURL covers the mgmt-bridge-present,
+// mgmt-bridge-absent, and no-config branches: a guest system VM (the K3s
+// server) can only reach predastore via the mgmt bridge, so the local
+// loopback-rewritten Predastore.Host must never leak into the URL handed to
+// a guest.
+func TestResolveSystemPredastoreURL(t *testing.T) {
+	tests := []struct {
+		name           string
+		predastoreHost string
+		mgmtBridge     string
+		want           string
+	}{
+		{
+			name:           "mgmt_bridge_present_uses_bridge_ip_and_configured_port",
+			predastoreHost: "127.0.0.1:9443",
+			mgmtBridge:     "10.15.8.1",
+			want:           "https://10.15.8.1:9443",
+		},
+		{
+			name:           "mgmt_bridge_present_defaults_port_when_host_has_none",
+			predastoreHost: "predastore-host-no-port",
+			mgmtBridge:     "10.15.8.1",
+			want:           "https://10.15.8.1:8443",
+		},
+		{
+			name:           "no_mgmt_bridge_falls_back_to_configured_host",
+			predastoreHost: "predastore.internal:8443",
+			want:           "https://predastore.internal:8443",
+		},
+		{
+			name: "no_mgmt_bridge_no_host_returns_empty",
+			want: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			d := &Daemon{
+				mgmtBridgeIP: tc.mgmtBridge,
+				config: &config.Config{
+					Predastore: config.PredastoreConfig{Host: tc.predastoreHost},
+				},
+			}
+			assert.Equal(t, tc.want, d.resolveSystemPredastoreURL())
+		})
+	}
+}
+
+// TestBuildEKSServiceDeps pins that the mgmt-bridge-derived predastore URL and
+// a non-nil snapshot object store actually reach EKSServiceDeps — RestoreSnapshot
+// silently can't resolve "latest snapshot" without SnapshotStore, and CP launch
+// silently can't write etcd-snapshot.env without SystemPredastoreURL.
+func TestBuildEKSServiceDeps(t *testing.T) {
+	d := &Daemon{
+		mgmtBridgeIP: "10.15.8.1",
+		config: &config.Config{
+			Predastore: config.PredastoreConfig{
+				Host:      "127.0.0.1:8443",
+				AccessKey: "AKIAPREDASTORE",
+				SecretKey: "pred-s3cr3t",
+				Region:    "us-east-1",
+			},
+		},
+	}
+
+	deps := d.buildEKSServiceDeps()
+
+	assert.Equal(t, "https://10.15.8.1:8443", deps.SystemPredastoreURL)
+	assert.NotNil(t, deps.SnapshotStore)
+}

spinifex/handlers/eks/cluster_reconciler.go

@@ -684,7 +684,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu
 	// consuming an attempt (stopping without the cluster-reset directive would just
 	// reproduce the wedge); a follower failure logs and continues (that member stays
 	// wedged, retried next attempt).
-	if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, seed.InstanceID, RecoveryActionClusterReset, ""); err != nil {
+	if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, seed.InstanceID, RecoveryActionClusterReset, "", false); err != nil {
 		slog.Warn("ClusterReconciler: set cluster-reset directive failed",
 			"cluster", r.clusterName, "seed", seed.InstanceID, "err", err)
 		return
@@ -692,7 +692,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu
 	r.resetAttempts++
 	r.resetIssued = true
 	for _, n := range members[1:] {
-		if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionWipeRejoin, ""); err != nil {
+		if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionWipeRejoin, "", false); err != nil {
 			slog.Warn("ClusterReconciler: set wipe-rejoin directive failed",
 				"cluster", r.clusterName, "instanceId", n.InstanceID, "err", err)
 		}
@@ -727,7 +727,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu
 // still prevents re-application of the already-applied directive.
 func (r *ClusterReconciler) clearRecoveryDirectives(meta *ClusterMeta) {
 	for _, n := range cpMemberNodes(meta) {
-		if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionNone, ""); err != nil {
+		if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionNone, "", false); err != nil {
 			slog.Warn("ClusterReconciler: clear recovery directive failed",
 				"cluster", r.clusterName, "instanceId", n.InstanceID, "err", err)
 		}

spinifex/handlers/eks/delete_cluster_test.go

@@ -73,30 +73,31 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture {
 	rt := newFakeRouteTableProvisioner()
 
 	svc, err := NewEKSServiceImpl(EKSServiceDeps{
-		NATSConn:         nc,
-		MasterKey:        bootstrapTestMasterKey,
-		GatewayBaseURL:   "https://gw.local:9999",
-		Region:           "us-east-1",
-		HolderID:         "node-1",
-		SystemGatewayURL: "https://gw.local:9999",
-		SystemAccessKey:  "AKIAEXAMPLE",
-		SystemSecretKey:  "s3cr3t-key",
-		GatewayCACert:    "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n",
-		VPCSG:            sg,
-		VPCK3s:           vpc,
-		VPCSubnet:        fakeSubnetResolver{},
-		IAM:              newFakeEnsurer(),
-		NLB:              nlb,
-		Instance:         inst,
-		Image:            ami,
-		EIP:              eip,
-		Worker:           worker,
-		VPCMgr:           vpcMgr,
-		IGW:              igw,
-		NATGW:            ngw,
-		RouteTable:       rt,
-		PlacementGroup:   &fakePlacer{},
-		Scheduler:        &fakeHostScheduler{},
+		NATSConn:            nc,
+		MasterKey:           bootstrapTestMasterKey,
+		GatewayBaseURL:      "https://gw.local:9999",
+		Region:              "us-east-1",
+		HolderID:            "node-1",
+		SystemGatewayURL:    "https://gw.local:9999",
+		SystemAccessKey:     "AKIAEXAMPLE",
+		SystemSecretKey:     "s3cr3t-key",
+		SystemPredastoreURL: "https://10.15.8.1:8443",
+		GatewayCACert:       "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n",
+		VPCSG:               sg,
+		VPCK3s:              vpc,
+		VPCSubnet:           fakeSubnetResolver{},
+		IAM:                 newFakeEnsurer(),
+		NLB:                 nlb,
+		Instance:            inst,
+		Image:               ami,
+		EIP:                 eip,
+		Worker:              worker,
+		VPCMgr:              vpcMgr,
+		IGW:                 igw,
+		NATGW:               ngw,
+		RouteTable:          rt,
+		PlacementGroup:      &fakePlacer{},
+		Scheduler:           &fakeHostScheduler{},
 	})
 	require.NoError(t, err)
 	// Tight Ready-gating knobs so nodegroup launches resolve in tests without the

spinifex/handlers/eks/k3s_ha_control_plane.go

@@ -162,6 +162,8 @@ func (s *EKSServiceImpl) ProvisionReplacementCP(ctx context.Context, req Replace
 		in.AccessKey = s.deps.SystemAccessKey
 		in.SecretKey = s.deps.SystemSecretKey
 	}
+	in.PredastoreAccessKey = s.deps.SystemAccessKey
+	in.PredastoreSecretKey = s.deps.SystemSecretKey
 
 	instanceType := in.InstanceType
 	if instanceType == "" {
@@ -182,6 +184,65 @@ func (s *EKSServiceImpl) ProvisionReplacementCP(ctx context.Context, req Replace
 	return controlPlaneNode(target, out), nil
 }
 
+// FreshCPRequest carries everything ProvisionFreshControlPlane needs to launch a
+// brand-new single-node control plane from a persisted template — the no-survivor
+// case, where there is no live quorum member to join.
+type FreshCPRequest struct {
+	AccountID    string
+	ClusterName  string
+	Template     *K3sServerInput
+	ExcludeHosts []string
+}
+
+// ProvisionFreshControlPlane launches a single control-plane VM as a fresh
+// cluster-init seed (ServerURL empty), replaying the create-time template with
+// rotating creds re-derived. It is the single-node analogue of
+// ProvisionReplacementCP for the case where the entire control plane (VM +
+// volume) is gone and there is no surviving etcd quorum to join — the caller
+// (restore-snapshot) drives the etcd restore via a RecoveryDirective set after
+// launch, not via any state this function knows about.
+func (s *EKSServiceImpl) ProvisionFreshControlPlane(ctx context.Context, req FreshCPRequest) (ControlPlaneNode, error) {
+	if req.Template == nil {
+		return ControlPlaneNode{}, errors.New("eks: ProvisionFreshControlPlane nil template")
+	}
+
+	in := *req.Template
+	in.ServerURL = ""
+	in.KonnServerCount = 1
+	in.PrunePeerIP = ""
+
+	sysAcct := admin.SystemAccountID()
+	in.IamInstanceProfileArn = ""
+	in.AccessKey = ""
+	in.SecretKey = ""
+	if profileARN := s.ensureCPInstanceProfile(sysAcct); profileARN != "" {
+		in.IamInstanceProfileArn = profileARN
+	} else {
+		in.AccessKey = s.deps.SystemAccessKey
+		in.SecretKey = s.deps.SystemSecretKey
+	}
+	in.PredastoreAccessKey = s.deps.SystemAccessKey
+	in.PredastoreSecretKey = s.deps.SystemSecretKey
+
+	instanceType := in.InstanceType
+	if instanceType == "" {
+		instanceType = defaultK3sServerInstanceType
+	}
+	target, err := s.pickReplacementHost(ctx, instanceType, req.ExcludeHosts)
+	if err != nil {
+		return ControlPlaneNode{}, err
+	}
+	in.TargetNodeID = target
+
+	out, err := LaunchK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, s.deps.Image, in)
+	if err != nil {
+		return ControlPlaneNode{}, err
+	}
+	slog.Info("ProvisionFreshControlPlane: fresh control plane launched",
+		"cluster", req.ClusterName, "instanceId", out.InstanceID, "host", target)
+	return controlPlaneNode(target, out), nil
+}
+
 // pickReplacementHost returns a schedulable host for the given instance type that
 // is not already holding a live control-plane member, preserving one-member-per-host
 // spread. Errors when every schedulable host is excluded (host permanently down),
@@ -191,11 +252,18 @@ func (s *EKSServiceImpl) pickReplacementHost(ctx context.Context, instanceType s
 	for _, h := range exclude {
 		ex[h] = true
 	}
-	for _, h := range s.deps.Scheduler.SchedulableHosts(ctx, instanceType) {
+	hosts := s.deps.Scheduler.SchedulableHosts(ctx, instanceType)
+	for _, h := range hosts {
 		if !ex[h] {
 			return h, nil
 		}
 	}
+	// On a single-node deployment the sole schedulable host is the old CP's
+	// own host. placeControlPlane already supports a <2-host single-CP
+	// topology, so fall back to it when that host is the only one available.
+	if len(hosts) == 1 {
+		return hosts[0], nil
+	}
 	return "", fmt.Errorf("eks: no schedulable host for replacement control plane (excluding %d live-member host(s))", len(exclude))
 }
 

spinifex/handlers/eks/k3s_ha_control_plane_test.go

@@ -583,3 +583,75 @@ func TestSpecForSystemType_RejectsNonSystem(t *testing.T) {
 	_, _, ok = instancetypes.SpecForSystemType("sys.nonexistent")
 	assert.False(t, ok, "unknown system size has no spec")
 }
+
+func TestProvisionFreshControlPlane_ClusterInitNoJoin(t *testing.T) {
+	sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b"}}
+	inst := &seqK3sInst{}
+	svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, inst)
+	// IAM is unwired here, so the launch re-derives static system creds from
+	// deps (as ProvisionReplacementCP/CreateCluster do).
+	svc.deps.SystemAccessKey = "AKIATEST"
+	svc.deps.SystemSecretKey = "secret"
+	tmpl := validK3sInput()
+
+	node, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{
+		ClusterName:  "alpha",
+		Template:     &tmpl,
+		ExcludeHosts: []string{"node-a"},
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, "node-b", node.NodeID, "placed on the one non-excluded host")
+	assert.Equal(t, "i-node-b", node.InstanceID)
+	assert.NotEmpty(t, node.ENIID, "fresh CP carries a fresh ENI")
+	ud := inst.userData["node-b"]
+	assert.Contains(t, ud, "cluster-init: true", "fresh CP seeds a new single-member etcd, not a join")
+	assert.NotContains(t, ud, "server: https://", "must not join a (nonexistent) survivor")
+	assert.Contains(t, ud, "EKS_ACCESS_KEY=AKIATEST", "re-derives system creds like ProvisionReplacementCP")
+	assert.Contains(t, ud, "SPINIFEX_PREDASTORE_AKID=AKIATEST", "re-derives predastore creds too")
+}
+
+func TestProvisionFreshControlPlane_NoFreeHostErrors(t *testing.T) {
+	// Two hosts, both excluded: genuinely nothing schedulable. A single
+	// excluded host is covered separately since that falls back to it
+	// (single-node topology).
+	sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b"}}
+	svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{})
+	svc.deps.SystemAccessKey = "AKIATEST"
+	svc.deps.SystemSecretKey = "secret"
+	tmpl := validK3sInput()
+
+	_, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{
+		ClusterName:  "alpha",
+		Template:     &tmpl,
+		ExcludeHosts: []string{"node-a", "node-b"},
+	})
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "no schedulable host")
+}
+
+func TestProvisionFreshControlPlane_SingleHostFallsBack(t *testing.T) {
+	// Single-node deployment: the only schedulable host is also the excluded
+	// (old CP) host. pickReplacementHost falls back to it rather than erroring.
+	sched := &fakeHostScheduler{hosts: []string{"node-a"}}
+	svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{})
+	svc.deps.SystemAccessKey = "AKIATEST"
+	svc.deps.SystemSecretKey = "secret"
+	tmpl := validK3sInput()
+
+	node, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{
+		ClusterName:  "alpha",
+		Template:     &tmpl,
+		ExcludeHosts: []string{"node-a"},
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, "node-a", node.NodeID)
+}
+
+func TestProvisionFreshControlPlane_NilTemplateRejected(t *testing.T) {
+	svc := newPlacerService(&fakeHostScheduler{}, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{})
+	_, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{ClusterName: "alpha"})
+	require.Error(t, err, "nil template rejected")
+}

spinifex/handlers/eks/k3s_server_vm.go

@@ -67,6 +67,10 @@ const (
 	// k3sFirstBootEnvPath is the env file k3s-first-boot.sh sources at boot.
 	k3sFirstBootEnvPath = "/etc/spinifex-eks/first-boot.env"
 
+	// k3sSnapshotEnvPath is the env file mulga-eks-etcd-snapshot.sh (cron) and
+	// mulga-eks-k3s-recovery.sh (boot-time restore) source for predastore creds.
+	k3sSnapshotEnvPath = "/etc/spinifex-eks/etcd-snapshot.env"
+
 	// agentEnvPath is the env file the k3s-agent OpenRC service sources for
 	// K3S_URL/K3S_TOKEN/K3S_NODE_NAME/K3S_NODE_LABEL.
 	agentEnvPath = "/etc/spinifex-eks/agent.env"
@@ -170,6 +174,13 @@ type K3sServerInput struct {
 	// EKS_KONNECTIVITY_SERVER_COUNT so the konnectivity-server advertises
 	// --server-count=N and every agent holds a tunnel to every replica (HA-correct).
 	KonnServerCount int `json:"konnServerCount,omitempty"`
+	// PredastoreEndpoint/PredastoreAccessKey/PredastoreSecretKey are the
+	// mgmt-bridge-reachable predastore SigV4 creds baked into etcd-snapshot.env.
+	// The guest snapshot cron and boot-time recovery script have no IMDS-role
+	// support, so these are always static (unlike AccessKey/SecretKey above).
+	PredastoreEndpoint  string `json:"predastoreEndpoint,omitempty"`
+	PredastoreAccessKey string `json:"predastoreAccessKey,omitempty"`
+	PredastoreSecretKey string `json:"predastoreSecretKey,omitempty"`
 }
 
 // K3sServerOutput carries identifiers to persist in ClusterMeta and register with the NLB.
@@ -447,6 +458,12 @@ func validateK3sServerInput(in K3sServerInput) error {
 		return errors.New("eks: K3sServerInput AccessKey and SecretKey must both be set or both empty")
 	case strings.TrimSpace(in.GatewayCACert) == "":
 		return errors.New("eks: K3sServerInput empty GatewayCACert")
+	case in.PredastoreEndpoint == "":
+		return errors.New("eks: K3sServerInput empty PredastoreEndpoint")
+	case in.PredastoreAccessKey == "":
+		return errors.New("eks: K3sServerInput empty PredastoreAccessKey")
+	case in.PredastoreSecretKey == "":
+		return errors.New("eks: K3sServerInput empty PredastoreSecretKey")
 	case in.ServerURL != "" && in.JoinToken == "":
 		return errors.New("eks: K3sServerInput join server (ServerURL set) requires JoinToken")
 	}
@@ -638,9 +655,22 @@ func buildK3sUserData(in K3sServerInput) string {
 	)
 	k3sConfig := strings.Join(configLines, "\n")
 
+	snapshotEnvBody := strings.Join([]string{
+		"EKS_ACCOUNT_ID=" + in.ClusterAccountID,
+		"EKS_CLUSTER_NAME=" + in.ClusterName,
+		"SPINIFEX_PREDASTORE_ENDPOINT=" + in.PredastoreEndpoint,
+		"SPINIFEX_PREDASTORE_AKID=" + in.PredastoreAccessKey,
+		"SPINIFEX_PREDASTORE_SECRET=" + in.PredastoreSecretKey,
+		// Without AWS_REGION the snapshot SigV4 PUT signs with the fallback
+		// region and predastore rejects it with 403.
+		"AWS_REGION=" + in.Region,
+	}, "\n")
+
 	files := []userDataFile{
 		// 0600: may contain a system SigV4 secret key (static-cred mode).
 		{Path: k3sFirstBootEnvPath, Perms: "0600", Body: envBody},
+		// 0600: carries predastore SigV4 static creds for the snapshot cron + boot-time recovery.
+		{Path: k3sSnapshotEnvPath, Perms: "0600", Body: snapshotEnvBody},
 		{Path: k3sOIDCSigningKeyPath, Perms: "0600", Body: strings.TrimRight(in.OIDCPrivateKeyPEM, "\n")},
 		{Path: k3sOIDCPublicKeyPath, Perms: "0644", Body: strings.TrimRight(in.OIDCPublicKeyPEM, "\n")},
 		{Path: k3sConfigPath, Perms: "0644", Body: k3sConfig},

spinifex/handlers/eks/k3s_server_vm_test.go

@@ -178,6 +178,10 @@ func validK3sInput() K3sServerInput {
 		SecretKey:         "s3cr3t-key",
 		GatewayCACert:     "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n",
 		JoinToken:         "clustertok-deadbeef",
+
+		PredastoreEndpoint:  "https://10.15.8.1:8443",
+		PredastoreAccessKey: "AKIAPREDASTORE",
+		PredastoreSecretKey: "pred-s3cr3t-key",
 	}
 }
 
@@ -205,6 +209,9 @@ func TestLaunchK3sServerVM_EmptyInputsRejected(t *testing.T) {
 		{"empty AccessKey", mk(func(in *K3sServerInput) { in.AccessKey = "" })},
 		{"empty SecretKey", mk(func(in *K3sServerInput) { in.SecretKey = "" })},
 		{"empty GatewayCACert", mk(func(in *K3sServerInput) { in.GatewayCACert = "  \n" })},
+		{"empty PredastoreEndpoint", mk(func(in *K3sServerInput) { in.PredastoreEndpoint = "" })},
+		{"empty PredastoreAccessKey", mk(func(in *K3sServerInput) { in.PredastoreAccessKey = "" })},
+		{"empty PredastoreSecretKey", mk(func(in *K3sServerInput) { in.PredastoreSecretKey = "" })},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -239,6 +246,16 @@ func TestBuildK3sUserData_StaticCredsBakeKeys(t *testing.T) {
 	assert.Contains(t, ud, "EKS_SECRET_KEY=s3cr3t-key")
 }
 
+func TestBuildK3sUserData_WritesEtcdSnapshotEnv(t *testing.T) {
+	ud := buildK3sUserData(validK3sInput())
+	require.Contains(t, ud, "path: "+k3sSnapshotEnvPath)
+	assert.Contains(t, ud, "EKS_ACCOUNT_ID=111122223333")
+	assert.Contains(t, ud, "EKS_CLUSTER_NAME=alpha")
+	assert.Contains(t, ud, "SPINIFEX_PREDASTORE_ENDPOINT=https://10.15.8.1:8443")
+	assert.Contains(t, ud, "SPINIFEX_PREDASTORE_AKID=AKIAPREDASTORE")
+	assert.Contains(t, ud, "SPINIFEX_PREDASTORE_SECRET=pred-s3cr3t-key")
+}
+
 func TestBuildK3sUserData_ProfileModeOmitsKeys(t *testing.T) {
 	in := validK3sInput()
 	in.AccessKey = ""

spinifex/handlers/eks/recovery_directive.go

@@ -30,10 +30,15 @@ const (
 // RecoveryDirective is the per-member recovery instruction. Epoch increases on
 // each set so the guest applies a directive at most once: it records the
 // last-applied epoch and ignores an equal-or-older one across reboots.
+// SnapshotRequired marks a snapshot the guest MUST restore (fresh DR seed): if
+// the fetch fails the guest aborts boot rather than cluster-reset into an empty
+// datastore. It stays false for the HA-reform path, whose seed cluster-resets
+// from intact local etcd when no snapshot is present.
 type RecoveryDirective struct {
-	Epoch    int64          `json:"epoch"`
-	Action   RecoveryAction `json:"action"`
-	Snapshot string         `json:"snapshot,omitempty"`
+	Epoch            int64          `json:"epoch"`
+	Action           RecoveryAction `json:"action"`
+	Snapshot         string         `json:"snapshot,omitempty"`
+	SnapshotRequired bool           `json:"snapshotRequired,omitempty"`
 }
 
 // GetRecoveryDirectiveInput names the cluster + member whose directive to read.
@@ -50,10 +55,11 @@ type GetRecoveryDirectiveOutput struct {
 // SetRecoveryDirectiveInput sets a member's directive; Epoch is assigned by the
 // store (previous+1), so callers supply only the action + optional snapshot key.
 type SetRecoveryDirectiveInput struct {
-	ClusterName string         `json:"clusterName"`
-	InstanceID  string         `json:"instanceId"`
-	Action      RecoveryAction `json:"action"`
-	Snapshot    string         `json:"snapshot,omitempty"`
+	ClusterName      string         `json:"clusterName"`
+	InstanceID       string         `json:"instanceId"`
+	Action           RecoveryAction `json:"action"`
+	Snapshot         string         `json:"snapshot,omitempty"`
+	SnapshotRequired bool           `json:"snapshotRequired,omitempty"`
 }
 
 // SetRecoveryDirectiveOutput returns the stored directive incl. the assigned epoch.
@@ -83,12 +89,12 @@ func LoadRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string) (Re
 
 // StoreRecoveryDirective writes a member's directive with epoch = previous+1 and
 // returns the stored value. Used by the reconciler escalation and the operator CLI.
-func StoreRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string, action RecoveryAction, snapshot string) (RecoveryDirective, error) {
+func StoreRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string, action RecoveryAction, snapshot string, snapshotRequired bool) (RecoveryDirective, error) {
 	prev, err := LoadRecoveryDirective(acctKV, cluster, instanceID)
 	if err != nil {
 		return RecoveryDirective{}, err
 	}
-	next := RecoveryDirective{Epoch: prev.Epoch + 1, Action: action, Snapshot: snapshot}
+	next := RecoveryDirective{Epoch: prev.Epoch + 1, Action: action, Snapshot: snapshot, SnapshotRequired: snapshotRequired}
 	data, err := json.Marshal(&next)
 	if err != nil {
 		return RecoveryDirective{}, fmt.Errorf("marshal recovery directive: %w", err)
@@ -130,7 +136,7 @@ func (s *EKSServiceImpl) SetRecoveryDirective(_ context.Context, input *SetRecov
 	if err != nil {
 		return nil, err
 	}
-	d, err := StoreRecoveryDirective(acctKV, input.ClusterName, input.InstanceID, input.Action, input.Snapshot)
+	d, err := StoreRecoveryDirective(acctKV, input.ClusterName, input.InstanceID, input.Action, input.Snapshot, input.SnapshotRequired)
 	if err != nil {
 		return nil, err
 	}

spinifex/handlers/eks/recovery_directive_test.go

@@ -26,11 +26,11 @@ func TestStoreRecoveryDirective_EpochIncrements(t *testing.T) {
 	acctKV, err := GetOrCreateAccountBucket(js, testAccountID, 1)
 	require.NoError(t, err)
 
-	first, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionClusterReset, "")
+	first, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionClusterReset, "", false)
 	require.NoError(t, err)
 	assert.Equal(t, int64(1), first.Epoch, "first set starts at epoch 1")
 
-	second, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionWipeRejoin, "snap.snap")
+	second, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionWipeRejoin, "snap.snap", false)
 	require.NoError(t, err)
 	assert.Equal(t, int64(2), second.Epoch, "each set advances the epoch")
 
@@ -41,6 +41,21 @@ func TestStoreRecoveryDirective_EpochIncrements(t *testing.T) {
 	assert.Equal(t, int64(2), got.Epoch)
 }
 
+func TestStoreRecoveryDirective_SnapshotRequiredRoundTrips(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	acctKV, err := GetOrCreateAccountBucket(js, testAccountID, 1)
+	require.NoError(t, err)
+
+	stored, err := StoreRecoveryDirective(acctKV, "alpha", "i-seed", RecoveryActionClusterReset, "etcd-frequent-20260709T010000Z.snap", true)
+	require.NoError(t, err)
+	assert.True(t, stored.SnapshotRequired, "a required snapshot is recorded so the guest aborts on fetch failure")
+
+	got, err := LoadRecoveryDirective(acctKV, "alpha", "i-seed")
+	require.NoError(t, err)
+	assert.True(t, got.SnapshotRequired, "SnapshotRequired survives the KV round-trip")
+	assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", got.Snapshot)
+}
+
 func TestSetRecoveryDirective_RejectsBadInput(t *testing.T) {
 	s := &EKSServiceImpl{}
 

spinifex/handlers/eks/restore_snapshot.go

@@ -0,0 +1,351 @@
+package handlers_eks
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/mulgadc/spinifex/spinifex/admin"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/nats-io/nats.go"
+)
+
+// eksBackupsBucket is the predastore bucket the guest etcd-snapshot cron PUTs
+// to and this restore path lists/reads from. Keys are
+// {accountID}/{clusterName}/etcd-{tier}-{timestamp}.snap.
+const eksBackupsBucket = "eks-backups-system"
+
+// Old-CP fence retry budget. A wedged-but-alive old CP left running while the
+// new one takes over is split-brain (two live etcds), so terminate is retried
+// before the restore is declared complete.
+const (
+	restoreFenceAttempts = 3
+	restoreFenceDelay    = 2 * time.Second
+)
+
+// RestoreSnapshotInput names the cluster to restore and, optionally, the exact
+// snapshot key (basename) to apply. Empty Snapshot resolves the latest
+// frequent-tier snapshot, falling back to the latest of any tier.
+type RestoreSnapshotInput struct {
+	ClusterName string `json:"clusterName"`
+	Snapshot    string `json:"snapshot,omitempty"`
+}
+
+// RestoreSnapshotOutput reports the fresh control-plane instance and the
+// snapshot key its boot-time recovery agent will apply. Status is deliberately
+// provisional: a successful return means the new CP was launched, its restore
+// directive queued, meta re-pointed and the old CP fenced — NOT that etcd is
+// restored and the apiserver is serving. That is observed asynchronously by the
+// reconciler, so the caller must verify cluster health before relying on it.
+type RestoreSnapshotOutput struct {
+	ClusterName   string `json:"clusterName"`
+	NewInstanceID string `json:"newInstanceId"`
+	Snapshot      string `json:"snapshot"`
+	Status        string `json:"status"`
+}
+
+// etcdSnapshotKey is a parsed "etcd-{tier}-{timestamp}.snap" object basename.
+type etcdSnapshotKey struct {
+	tier      string
+	timestamp string
+	basename  string
+}
+
+// parseEtcdSnapshotKey parses a snapshot object basename (not the full S3 key)
+// into tier + timestamp. The timestamp (YYYYMMDDTHHMMSSZ) is fixed-width, so
+// lexicographic comparison of the timestamp alone is chronological — tier names
+// vary in length, so comparing the whole basename would not be.
+func parseEtcdSnapshotKey(basename string) (etcdSnapshotKey, bool) {
+	const prefix, suffix = "etcd-", ".snap"
+	if !strings.HasPrefix(basename, prefix) || !strings.HasSuffix(basename, suffix) {
+		return etcdSnapshotKey{}, false
+	}
+	mid := strings.TrimSuffix(strings.TrimPrefix(basename, prefix), suffix)
+	i := strings.LastIndex(mid, "-")
+	if i <= 0 || i == len(mid)-1 {
+		return etcdSnapshotKey{}, false
+	}
+	return etcdSnapshotKey{tier: mid[:i], timestamp: mid[i+1:], basename: basename}, true
+}
+
+// resolveLatestSnapshot lists the cluster's snapshot prefix and returns the
+// newest object's basename, preferring the frequent tier (tightest RPO) and
+// falling back to the newest snapshot of any tier.
+func resolveLatestSnapshot(ctx context.Context, store objectstore.ObjectStore, accountID, clusterName string) (string, error) {
+	if store == nil {
+		return "", errors.New("eks: no snapshot store configured; pass --snapshot explicitly")
+	}
+	prefix := accountID + "/" + clusterName + "/"
+	out, err := store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+		Bucket: aws.String(eksBackupsBucket),
+		Prefix: aws.String(prefix),
+	})
+	if err != nil {
+		return "", fmt.Errorf("list etcd snapshots: %w", err)
+	}
+	var bestFrequent, bestAny etcdSnapshotKey
+	haveFrequent, haveAny := false, false
+	for _, obj := range out.Contents {
+		base := strings.TrimPrefix(aws.StringValue(obj.Key), prefix)
+		parsed, ok := parseEtcdSnapshotKey(base)
+		if !ok {
+			continue
+		}
+		if !haveAny || parsed.timestamp > bestAny.timestamp {
+			bestAny, haveAny = parsed, true
+		}
+		if parsed.tier == "frequent" && (!haveFrequent || parsed.timestamp > bestFrequent.timestamp) {
+			bestFrequent, haveFrequent = parsed, true
+		}
+	}
+	switch {
+	case haveFrequent:
+		return bestFrequent.basename, nil
+	case haveAny:
+		return bestAny.basename, nil
+	default:
+		return "", fmt.Errorf("eks: no etcd snapshots found under %s/%s", eksBackupsBucket, prefix)
+	}
+}
+
+// validateSnapshotExists HEADs the snapshot object before any mutation. An
+// explicit --snapshot that is typo'd or expired would otherwise be handed to the
+// guest verbatim, its fetch fail, and — without the required-snapshot guard — a
+// fresh seed cluster-reset into an EMPTY datastore reported as success. A nil
+// store cannot verify, so it is a hard error rather than an optimistic pass.
+func validateSnapshotExists(ctx context.Context, store objectstore.ObjectStore, accountID, clusterName, basename string) error {
+	if store == nil {
+		return errors.New("eks: no snapshot store configured; cannot verify the snapshot exists before a destructive restore")
+	}
+	key := accountID + "/" + clusterName + "/" + basename
+	if _, err := store.HeadObject(ctx, &s3.HeadObjectInput{
+		Bucket: aws.String(eksBackupsBucket),
+		Key:    aws.String(key),
+	}); err != nil {
+		if objectstore.IsNoSuchKeyError(err) {
+			return fmt.Errorf("eks: snapshot %q not found under %s/%s/%s/; refusing to restore (a cluster-reset with no snapshot boots an EMPTY cluster)",
+				basename, eksBackupsBucket, accountID, clusterName)
+		}
+		return fmt.Errorf("eks: verify snapshot %q exists: %w", basename, err)
+	}
+	return nil
+}
+
+// replaceControlPlaneForRestore overwrites meta's single control-plane member
+// with the freshly provisioned node, mirroring the create-path scalar fields
+// (service_impl.go CreateCluster). restore-snapshot only ever targets a
+// single-CP cluster (isHAControlPlane guards the caller), so this replaces
+// wholesale rather than matching a dead instance ID like SwapControlPlaneMember.
+func replaceControlPlaneForRestore(kv nats.KeyValue, name string, newNode ControlPlaneNode) error {
+	return casUpdateMeta(kv, name, func(m *ClusterMeta) bool {
+		m.ControlPlaneNodes = []ControlPlaneNode{newNode}
+		m.ControlPlaneInstanceID = newNode.InstanceID
+		m.ControlPlaneENIID = newNode.ENIID
+		m.ControlPlaneENIIP = newNode.ENIIP
+		m.ControlPlaneMgmtIP = newNode.MgmtIP
+		return true
+	})
+}
+
+// unwindFreshCP rolls back a just-launched fresh CP when a step before the meta
+// commit fails. It supersedes the recovery directive with a no-op (so a VM that
+// boots before terminate lands does not destructively reset) then terminates the
+// VM. Both are best-effort and logged: leaving an orphan boot a resetting CP,
+// which the operator would then compound by re-running restore into a second one.
+func (s *EKSServiceImpl) unwindFreshCP(ctx context.Context, accountID, clusterName string, node ControlPlaneNode) {
+	if _, err := s.SetRecoveryDirective(ctx, &SetRecoveryDirectiveInput{
+		ClusterName: clusterName,
+		InstanceID:  node.InstanceID,
+		Action:      RecoveryActionNone,
+	}, accountID); err != nil {
+		slog.WarnContext(ctx, "RestoreSnapshot: clear directive during unwind failed; guest may still reset if it boots",
+			"cluster", clusterName, "instanceId", node.InstanceID, "err", err)
+	}
+	if err := TerminateK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, admin.SystemAccountID(), node.InstanceID, node.ENIID); err != nil {
+		slog.WarnContext(ctx, "RestoreSnapshot: terminate fresh CP during unwind failed; may orphan a resetting control plane",
+			"cluster", clusterName, "instanceId", node.InstanceID, "err", err)
+	}
+}
+
+// confirmOldCPTerminated fences the old control plane: it retries terminate until
+// the instance is confirmed gone (TerminateK3sServerVM treats an already-absent
+// instance as success, the primary DR case). A persistent failure means a
+// wedged-but-alive old CP that would run its etcd alongside the new one —
+// split-brain — so the caller must surface it loudly rather than declare success.
+func (s *EKSServiceImpl) confirmOldCPTerminated(ctx context.Context, accountID string, node ControlPlaneNode, attempts int, delay time.Duration) error {
+	if node.InstanceID == "" && node.ENIID == "" {
+		return nil
+	}
+	var lastErr error
+	for i := range attempts {
+		if i > 0 {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(delay):
+			}
+		}
+		if err := TerminateK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, accountID, node.InstanceID, node.ENIID); err != nil {
+			lastErr = err
+			slog.WarnContext(ctx, "RestoreSnapshot: old control plane terminate attempt failed",
+				"instanceId", node.InstanceID, "attempt", i+1, "err", err)
+			continue
+		}
+		return nil
+	}
+	return lastErr
+}
+
+// RestoreSnapshot drives the single-CP total-loss DR path: it validates the
+// snapshot exists, launches a fresh control-plane VM as a cluster-init seed,
+// directs its boot-time recovery agent to `k3s server --cluster-reset` restoring
+// that snapshot, persists the new CP into meta, re-points the cluster NLB target
+// groups, and fences the old CP. HA clusters are out of scope — they have a
+// potentially surviving quorum and should be recovered via quorum reformation,
+// not a destructive single-node rebuild. The returned status is provisional:
+// success means the sequence completed, not that etcd is restored and serving.
+func (s *EKSServiceImpl) RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error) {
+	if input == nil || input.ClusterName == "" {
+		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	acctKV, err := s.acctKVForCluster(accountID, input.ClusterName)
+	if err != nil {
+		return nil, err
+	}
+	meta, err := GetClusterMeta(acctKV, input.ClusterName)
+	if err != nil {
+		if errors.Is(err, ErrClusterNotFound) {
+			return nil, errors.New(awserrors.ErrorEKSResourceNotFound)
+		}
+		return nil, err
+	}
+	if isHAControlPlane(meta) {
+		return nil, errors.New("eks: restore-snapshot only supports a single control-plane cluster; " +
+			"this cluster runs an HA spread with a potentially surviving quorum — use quorum reformation instead")
+	}
+	if meta.ControlPlaneTemplate == nil {
+		return nil, errors.New("eks: cluster has no persisted control-plane template; cannot restore")
+	}
+
+	// Resolve + validate the snapshot BEFORE any mutation. resolveLatestSnapshot
+	// already guarantees existence (it lists then picks); an explicit key is HEADed
+	// so a typo/expired key hard-fails here rather than booting an empty cluster.
+	snapshot := input.Snapshot
+	if snapshot == "" {
+		snapshot, err = resolveLatestSnapshot(ctx, s.deps.SnapshotStore, accountID, input.ClusterName)
+		if err != nil {
+			return nil, err
+		}
+	} else if verr := validateSnapshotExists(ctx, s.deps.SnapshotStore, accountID, input.ClusterName, snapshot); verr != nil {
+		return nil, verr
+	}
+
+	var oldNode ControlPlaneNode
+	if oldNodes := controlPlaneTeardownNodes(meta); len(oldNodes) > 0 {
+		oldNode = oldNodes[0]
+	}
+	var excludeHosts []string
+	if oldNode.NodeID != "" {
+		excludeHosts = []string{oldNode.NodeID}
+	}
+
+	newNode, err := s.ProvisionFreshControlPlane(ctx, FreshCPRequest{
+		AccountID:    accountID,
+		ClusterName:  input.ClusterName,
+		Template:     meta.ControlPlaneTemplate,
+		ExcludeHosts: excludeHosts,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("eks: provision fresh control plane: %w", err)
+	}
+
+	// Wire the recovery directive immediately after launch: the guest applies it
+	// before k3s starts on this very first boot, and it is keyed by InstanceID,
+	// which only exists once launched. SnapshotRequired makes the guest abort
+	// (not reset into an empty datastore) if it cannot fetch the snapshot.
+	// Any failure up to and including the meta commit unwinds the fresh CP, so a
+	// re-run does not stack a second resetting control plane.
+	if _, derr := s.SetRecoveryDirective(ctx, &SetRecoveryDirectiveInput{
+		ClusterName:      input.ClusterName,
+		InstanceID:       newNode.InstanceID,
+		Action:           RecoveryActionClusterReset,
+		Snapshot:         snapshot,
+		SnapshotRequired: true,
+	}, accountID); derr != nil {
+		s.unwindFreshCP(ctx, accountID, input.ClusterName, newNode)
+		return nil, fmt.Errorf("eks: set recovery directive: %w", derr)
+	}
+
+	// Persist the new CP into meta BEFORE re-pointing the NLB. Once committed the
+	// new CP is canonical, so a later NLB failure is convergeable by the reconciler
+	// (provisional success) rather than a bare error implying nothing happened.
+	if perr := replaceControlPlaneForRestore(acctKV, input.ClusterName, newNode); perr != nil {
+		s.unwindFreshCP(ctx, accountID, input.ClusterName, newNode)
+		return nil, fmt.Errorf("eks: persist replacement control plane: %w", perr)
+	}
+
+	sysAcct := admin.SystemAccountID()
+	// Re-point the cluster NLB: deregister the old (likely already-dead) CP,
+	// register the new one, on both the apiserver and konnectivity TGs. Deregister
+	// is best-effort (the old target is likely already unhealthy/absent). A failed
+	// register leaves meta already naming the new CP, so the reconciler converges —
+	// surfaced as a provisional status, not an error.
+	if oldNode.ENIIP != "" {
+		if meta.NLBTargetGroupArn != "" {
+			if derr := DeregisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.NLBTargetGroupArn, oldNode.ENIIP, k3sAPIServerPort); derr != nil {
+				slog.WarnContext(ctx, "RestoreSnapshot: deregister old apiserver target failed", "cluster", input.ClusterName, "err", derr)
+			}
+		}
+		if meta.KonnTargetGroupArn != "" {
+			if derr := DeregisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.KonnTargetGroupArn, oldNode.ENIIP, konnectivityAgentPort); derr != nil {
+				slog.WarnContext(ctx, "RestoreSnapshot: deregister old konnectivity target failed", "cluster", input.ClusterName, "err", derr)
+			}
+		}
+	}
+	nlbComplete := true
+	if meta.NLBTargetGroupArn != "" {
+		if rerr := RegisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.NLBTargetGroupArn, newNode.ENIIP, k3sAPIServerPort); rerr != nil {
+			nlbComplete = false
+			slog.WarnContext(ctx, "RestoreSnapshot: register new apiserver target failed; reconciler will converge",
+				"cluster", input.ClusterName, "err", rerr)
+		}
+	}
+	if meta.KonnTargetGroupArn != "" {
+		if rerr := RegisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.KonnTargetGroupArn, newNode.ENIIP, konnectivityAgentPort); rerr != nil {
+			nlbComplete = false
+			slog.WarnContext(ctx, "RestoreSnapshot: register new konnectivity target failed; reconciler will converge",
+				"cluster", input.ClusterName, "err", rerr)
+		}
+	}
+
+	// Fence the old CP before declaring success: a wedged-but-alive old CP running
+	// its etcd next to the new one is split-brain. Terminate is retried; if the old
+	// instance cannot be confirmed gone, fail loudly (the new CP is up and canonical
+	// — the operator must fence the old one manually, NOT re-run restore).
+	if ferr := s.confirmOldCPTerminated(ctx, sysAcct, oldNode, restoreFenceAttempts, restoreFenceDelay); ferr != nil {
+		return nil, fmt.Errorf("eks: restore of %q brought up new control plane %s but the old control plane %s could not be confirmed terminated after %d attempts (split-brain risk) — fence it manually, do NOT re-run restore: %w",
+			input.ClusterName, newNode.InstanceID, oldNode.InstanceID, restoreFenceAttempts, ferr)
+	}
+
+	status := fmt.Sprintf("restore initiated: new control plane %s is booting and will restore etcd from snapshot %s — verify cluster health before relying on it",
+		newNode.InstanceID, snapshot)
+	if !nlbComplete {
+		status = fmt.Sprintf("control plane %s restored and recorded, but NLB re-point is incomplete; the reconciler will converge — verify cluster health",
+			newNode.InstanceID)
+	}
+	slog.InfoContext(ctx, "RestoreSnapshot: control-plane restore sequence complete",
+		"cluster", input.ClusterName, "newInstanceId", newNode.InstanceID, "snapshot", snapshot, "nlbComplete", nlbComplete)
+
+	return &RestoreSnapshotOutput{
+		ClusterName:   input.ClusterName,
+		NewInstanceID: newNode.InstanceID,
+		Snapshot:      snapshot,
+		Status:        status,
+	}, nil
+}

spinifex/handlers/eks/restore_snapshot_test.go

@@ -0,0 +1,317 @@
+package handlers_eks
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseEtcdSnapshotKey(t *testing.T) {
+	cases := []struct {
+		name     string
+		basename string
+		wantOK   bool
+		wantTier string
+		wantTS   string
+	}{
+		{"frequent tier", "etcd-frequent-20260709T010000Z.snap", true, "frequent", "20260709T010000Z"},
+		{"daily tier", "etcd-daily-20260601T000000Z.snap", true, "daily", "20260601T000000Z"},
+		{"missing prefix", "frequent-20260709T010000Z.snap", false, "", ""},
+		{"missing suffix", "etcd-frequent-20260709T010000Z.tar", false, "", ""},
+		{"no tier separator", "etcd-20260709T010000Z.snap", false, "", ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			parsed, ok := parseEtcdSnapshotKey(tc.basename)
+			require.Equal(t, tc.wantOK, ok)
+			if tc.wantOK {
+				assert.Equal(t, tc.wantTier, parsed.tier)
+				assert.Equal(t, tc.wantTS, parsed.timestamp)
+			}
+		})
+	}
+}
+
+func seedSnapshot(t *testing.T, store objectstore.ObjectStore, accountID, cluster, basename string) {
+	t.Helper()
+	key := accountID + "/" + cluster + "/" + basename
+	_, err := store.PutObject(context.Background(), &s3.PutObjectInput{
+		Bucket: aws.String(eksBackupsBucket),
+		Key:    aws.String(key),
+		Body:   strings.NewReader("fake-snapshot-data"),
+	})
+	require.NoError(t, err)
+}
+
+func TestResolveLatestSnapshot_PrefersFrequentTier(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260709T230000Z.snap")
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap")
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T020000Z.snap")
+	// A different cluster's snapshot must not be considered.
+	seedSnapshot(t, store, testAccountID, "beta", "etcd-frequent-20260710T230000Z.snap")
+
+	got, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, "etcd-frequent-20260709T020000Z.snap", got, "newest frequent-tier snapshot wins even though a newer daily exists")
+}
+
+func TestResolveLatestSnapshot_FallsBackToAnyTierWhenNoFrequent(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260601T000000Z.snap")
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260602T000000Z.snap")
+
+	got, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, "etcd-daily-20260602T000000Z.snap", got)
+}
+
+func TestResolveLatestSnapshot_NoStoreConfigured(t *testing.T) {
+	_, err := resolveLatestSnapshot(context.Background(), nil, testAccountID, "alpha")
+	require.Error(t, err)
+}
+
+func TestResolveLatestSnapshot_NoSnapshotsFound(t *testing.T) {
+	store := objectstore.NewMemoryObjectStore()
+	_, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha")
+	require.Error(t, err)
+}
+
+// restoreSnapshotFixtureMeta builds a single-CP ClusterMeta with everything
+// RestoreSnapshot needs: a persisted launch template and NLB target-group ARNs.
+func restoreSnapshotFixtureMeta(name string) *ClusterMeta {
+	tmpl := validK3sInput()
+	tmpl.AccessKey = ""
+	tmpl.SecretKey = ""
+	tmpl.PredastoreAccessKey = ""
+	tmpl.PredastoreSecretKey = ""
+
+	meta := sampleClusterMeta(name)
+	meta.NLBTargetGroupArn = "arn:aws:elasticloadbalancing:us-east-1:111122223333:targetgroup/" + ClusterTargetGroupName(name) + "/tg-001"
+	meta.KonnTargetGroupArn = "arn:aws:elasticloadbalancing:us-east-1:111122223333:targetgroup/" + ClusterTargetGroupName(name) + "-konn/tg-002"
+	meta.ControlPlaneTemplate = &tmpl
+	meta.ControlPlaneNodes = []ControlPlaneNode{{
+		NodeID:     "node-old",
+		InstanceID: "i-old111",
+		ENIID:      "eni-old111",
+		ENIIP:      "10.0.1.9",
+		MgmtIP:     "10.255.0.5",
+	}}
+	meta.ControlPlaneInstanceID = "i-old111"
+	meta.ControlPlaneENIID = "eni-old111"
+	meta.ControlPlaneENIIP = "10.0.1.9"
+	return meta
+}
+
+func TestRestoreSnapshot_HappyPathLaunchesDirectsAndRepoints(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}}
+	store := objectstore.NewMemoryObjectStore()
+	f.svc.deps.SnapshotStore = store
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap")
+
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	out, err := f.svc.RestoreSnapshot(context.Background(), &RestoreSnapshotInput{ClusterName: "alpha"}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", out.Snapshot, "empty --snapshot resolves the latest one")
+	assert.NotEmpty(t, out.NewInstanceID)
+
+	// The recovery directive must be wired for the new instance.
+	dirOut, err := f.svc.GetRecoveryDirective(context.Background(),
+		&GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: out.NewInstanceID}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, RecoveryActionClusterReset, dirOut.Directive.Action)
+	assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", dirOut.Directive.Snapshot)
+	assert.True(t, dirOut.Directive.SnapshotRequired, "fresh DR seed must require the snapshot (abort, not reset-into-empty)")
+	assert.NotEmpty(t, out.Status, "success carries a provisional status, never a bare 'restored'")
+
+	// NLB re-pointed: old CP deregistered, new CP registered, on both TGs.
+	require.Len(t, f.nlb.deregisterCalls, 2, "apiserver + konnectivity TGs")
+	require.Len(t, f.nlb.registerCalls, 2)
+	for _, dc := range f.nlb.deregisterCalls {
+		require.Len(t, dc.Targets, 1)
+		assert.Equal(t, "10.0.1.9", aws.StringValue(dc.Targets[0].Id), "deregisters the OLD CP's ENI IP")
+	}
+	for _, rc := range f.nlb.registerCalls {
+		require.Len(t, rc.Targets, 1)
+		assert.NotEqual(t, "10.0.1.9", aws.StringValue(rc.Targets[0].Id), "registers the NEW CP's ENI IP, not the old one")
+	}
+
+	// Cluster meta reflects the replacement.
+	got, err := GetClusterMeta(f.kv, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, out.NewInstanceID, got.ControlPlaneInstanceID)
+	require.Len(t, got.ControlPlaneNodes, 1)
+	assert.Equal(t, out.NewInstanceID, got.ControlPlaneNodes[0].InstanceID)
+
+	// The old CP is torn down best-effort.
+	assert.Contains(t, f.inst.terminateCalls, "i-old111")
+}
+
+func TestRestoreSnapshot_ExplicitSnapshotUsedVerbatim(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}}
+	store := objectstore.NewMemoryObjectStore()
+	f.svc.deps.SnapshotStore = store
+	// A different (newer) snapshot exists; the explicit one must be used verbatim.
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260710T010000Z.snap")
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260601T000000Z.snap")
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	out, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, "etcd-daily-20260601T000000Z.snap", out.Snapshot, "explicit snapshot is used verbatim, no latest-resolution")
+
+	// The directive must mark the snapshot as required so the guest aborts on a
+	// fetch failure rather than resetting into an empty datastore.
+	dirOut, err := f.svc.GetRecoveryDirective(context.Background(),
+		&GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: out.NewInstanceID}, testAccountID)
+	require.NoError(t, err)
+	assert.True(t, dirOut.Directive.SnapshotRequired)
+	assert.NotEmpty(t, out.Status, "a provisional status is always returned")
+}
+
+func TestRestoreSnapshot_ExplicitSnapshotMissingHardFailsBeforeLaunch(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}}
+	store := objectstore.NewMemoryObjectStore()
+	f.svc.deps.SnapshotStore = store
+	// Only an unrelated snapshot exists — the requested key is absent.
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap")
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	_, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "not found")
+	assert.Empty(t, f.inst.launchCalls, "a missing explicit snapshot must hard-fail before launching anything")
+}
+
+func TestRestoreSnapshot_ExplicitSnapshotWithNoStoreHardFails(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}}
+	f.svc.deps.SnapshotStore = nil
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	_, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.Error(t, err)
+	assert.Empty(t, f.inst.launchCalls, "cannot verify snapshot exists without a store — must not launch")
+}
+
+func TestRestoreSnapshot_HARejected(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	meta := restoreSnapshotFixtureMeta("alpha")
+	meta.ControlPlaneSpreadGroup = "eks-cp-111122223333-alpha"
+	meta.ControlPlaneNodes = append(meta.ControlPlaneNodes, ControlPlaneNode{NodeID: "node-2", InstanceID: "i-cp2"})
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	_, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "HA spread")
+	assert.Empty(t, f.inst.launchCalls, "must not launch anything for an HA cluster")
+}
+
+func TestRestoreSnapshot_NoTemplateRejected(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	meta := restoreSnapshotFixtureMeta("alpha")
+	meta.ControlPlaneTemplate = nil
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	_, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "template")
+}
+
+func TestRestoreSnapshot_ClusterNotFound(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	_, err := f.svc.RestoreSnapshot(context.Background(),
+		&RestoreSnapshotInput{ClusterName: "ghost", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID)
+	require.Error(t, err)
+}
+
+func TestRestoreSnapshot_NLBFailureIsProvisionalNotError(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}}
+	store := objectstore.NewMemoryObjectStore()
+	f.svc.deps.SnapshotStore = store
+	seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap")
+	// NLB register fails after meta is persisted: the new CP is already canonical.
+	f.nlb.registerErr = errors.New("TargetGroupNotFound")
+
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	out, err := f.svc.RestoreSnapshot(context.Background(), &RestoreSnapshotInput{ClusterName: "alpha"}, testAccountID)
+	require.NoError(t, err, "an NLB re-point failure after meta commit is provisional, not a hard error")
+	assert.Contains(t, out.Status, "NLB re-point is incomplete")
+
+	// Meta was persisted BEFORE the NLB step, so it already names the new CP and
+	// the reconciler can converge the target group.
+	got, err := GetClusterMeta(f.kv, "alpha")
+	require.NoError(t, err)
+	assert.Equal(t, out.NewInstanceID, got.ControlPlaneInstanceID, "meta persisted before NLB re-point")
+
+	// The new CP must NOT be torn down on an NLB failure (it is canonical now).
+	assert.NotContains(t, f.inst.terminateCalls, out.NewInstanceID)
+}
+
+func TestRestoreSnapshot_FenceFailureIsLoudError(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	// The old CP cannot be terminated (still alive) — split-brain risk.
+	f.inst.terminateErr = errors.New("InstanceStillRunning")
+
+	oldNode := ControlPlaneNode{NodeID: "node-old", InstanceID: "i-old111", ENIID: "eni-old111", ENIIP: "10.0.1.9"}
+	// Drive the fence helper directly with no delay so the retry budget is exercised fast.
+	err := f.svc.confirmOldCPTerminated(context.Background(), testAccountID, oldNode, 3, 0)
+	require.Error(t, err, "a persistently un-terminable old CP is a fence failure")
+	require.Len(t, f.inst.terminateCalls, 3, "terminate is retried up to the attempt budget")
+}
+
+func TestConfirmOldCPTerminated_NoOldNodeSucceeds(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	err := f.svc.confirmOldCPTerminated(context.Background(), testAccountID, ControlPlaneNode{}, 3, 0)
+	require.NoError(t, err, "no old CP to fence is trivially confirmed")
+	assert.Empty(t, f.inst.terminateCalls)
+}
+
+func TestUnwindFreshCP_TerminatesAndClearsDirective(t *testing.T) {
+	f := newEKSServiceFixture(t)
+	meta := restoreSnapshotFixtureMeta("alpha")
+	require.NoError(t, PutClusterMeta(f.kv, meta))
+
+	// Pretend a directive was already set for the fresh CP (as the real flow does
+	// just before a meta-commit failure), then unwind.
+	node := ControlPlaneNode{NodeID: "node-new", InstanceID: "i-new999", ENIID: "eni-new999", ENIIP: "10.0.1.42"}
+	_, err := f.svc.SetRecoveryDirective(context.Background(), &SetRecoveryDirectiveInput{
+		ClusterName: "alpha", InstanceID: node.InstanceID,
+		Action: RecoveryActionClusterReset, Snapshot: "etcd-frequent-20260709T010000Z.snap", SnapshotRequired: true,
+	}, testAccountID)
+	require.NoError(t, err)
+
+	f.svc.unwindFreshCP(context.Background(), testAccountID, "alpha", node)
+
+	// The fresh CP is terminated so it cannot boot and destructively reset.
+	assert.Contains(t, f.inst.terminateCalls, "i-new999")
+
+	// The directive is superseded with a no-op so a VM that boots first does not reset.
+	dirOut, err := f.svc.GetRecoveryDirective(context.Background(),
+		&GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: node.InstanceID}, testAccountID)
+	require.NoError(t, err)
+	assert.Equal(t, RecoveryActionNone, dirOut.Directive.Action, "directive cleared to none on unwind")
+}

spinifex/handlers/eks/service.go

@@ -54,6 +54,11 @@ type EKSService interface {
 	// reconciler / operator CLI set it to drive an etcd cluster-reset recovery.
 	GetRecoveryDirective(ctx context.Context, input *GetRecoveryDirectiveInput, accountID string) (*GetRecoveryDirectiveOutput, error)
 	SetRecoveryDirective(ctx context.Context, input *SetRecoveryDirectiveInput, accountID string) (*SetRecoveryDirectiveOutput, error)
+	// RestoreSnapshot is an internal control-plane method (not an AWS-SDK
+	// action): the operator CLI (spx admin eks restore-snapshot) drives the
+	// single-CP total-loss DR path — fresh CP launch, recovery-directive wiring,
+	// and NLB re-point.
+	RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error)
 
 	// OIDC identity-provider configs.
 	AssociateIdentityProviderConfig(ctx context.Context, input *eks.AssociateIdentityProviderConfigInput, accountID string) (*eks.AssociateIdentityProviderConfigOutput, error)

spinifex/handlers/eks/service_impl.go

@@ -24,6 +24,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/config"
 	handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
 	"github.com/mulgadc/spinifex/spinifex/handlers/sysinstance"
+	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
 )
@@ -60,6 +61,16 @@ type EKSServiceDeps struct {
 	SystemSecretKey  string
 	GatewayCACert    string
 
+	// SystemPredastoreURL is the mgmt-bridge-reachable predastore endpoint baked
+	// into the CP VM's etcd-snapshot.env (SystemAccessKey/SystemSecretKey double
+	// as the predastore SigV4 creds — same system credential used for the gateway).
+	SystemPredastoreURL string
+
+	// SnapshotStore lists/reads etcd snapshots from the eks-backups-system bucket
+	// for restore-snapshot's latest-snapshot resolution. Nil disables that lookup
+	// (callers must pass --snapshot explicitly).
+	SnapshotStore objectstore.ObjectStore
+
 	VPCSG     sgProvisioner
 	VPCK3s    k3sVPCProvisioner
 	VPCSubnet SubnetVPCResolver
@@ -707,24 +718,27 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc
 	elbSubnets := dedupSubnetsByAZ(ctx, s.deps.VPCSubnet, accountID, lc.subnetIDs)
 
 	serverIn := K3sServerInput{
-		AccountID:         sysAcct,
-		ClusterAccountID:  accountID,
-		ClusterName:       name,
-		Region:            region,
-		SubnetID:          cpRefs.PrivateSubnetIDs[0],
-		VpcID:             meta.ResourcesVpcConfig.VpcId,
-		ELBSubnetIDs:      elbSubnets,
-		ControlPlaneSGID:  cpSG,
-		NLBDNS:            nlb.DNSName,
-		EndpointIP:        nlb.FrontendIP,
-		PrivateEndpointIP: meta.PrivateEndpointIP,
-		OIDCIssuer:        oidcIssuer,
-		OIDCPrivateKeyPEM: privPEM,
-		OIDCPublicKeyPEM:  pubPEM,
-		GatewayURL:        s.deps.SystemGatewayURL,
-		AddonGatewayURL:   s.deps.GatewayBaseURL,
-		GatewayCACert:     s.deps.GatewayCACert,
-		JoinToken:         joinToken,
+		AccountID:           sysAcct,
+		ClusterAccountID:    accountID,
+		ClusterName:         name,
+		Region:              region,
+		SubnetID:            cpRefs.PrivateSubnetIDs[0],
+		VpcID:               meta.ResourcesVpcConfig.VpcId,
+		ELBSubnetIDs:        elbSubnets,
+		ControlPlaneSGID:    cpSG,
+		NLBDNS:              nlb.DNSName,
+		EndpointIP:          nlb.FrontendIP,
+		PrivateEndpointIP:   meta.PrivateEndpointIP,
+		OIDCIssuer:          oidcIssuer,
+		OIDCPrivateKeyPEM:   privPEM,
+		OIDCPublicKeyPEM:    pubPEM,
+		GatewayURL:          s.deps.SystemGatewayURL,
+		AddonGatewayURL:     s.deps.GatewayBaseURL,
+		GatewayCACert:       s.deps.GatewayCACert,
+		JoinToken:           joinToken,
+		PredastoreEndpoint:  s.deps.SystemPredastoreURL,
+		PredastoreAccessKey: s.deps.SystemAccessKey,
+		PredastoreSecretKey: s.deps.SystemSecretKey,
 	}
 
 	// Prefer IMDS instance-role creds: attach a system instance profile so the
@@ -767,6 +781,8 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc
 	tmpl.AccessKey = ""
 	tmpl.SecretKey = ""
 	tmpl.IamInstanceProfileArn = ""
+	tmpl.PredastoreAccessKey = ""
+	tmpl.PredastoreSecretKey = ""
 	meta.ControlPlaneTemplate = &tmpl
 
 	// Persist the CP VM + ENI + spread-group refs now, before any further fallible

spinifex/handlers/eks/service_nats.go

@@ -153,6 +153,10 @@ func (s *NATSEKSService) SetRecoveryDirective(ctx context.Context, input *SetRec
 	return utils.NATSRequest[SetRecoveryDirectiveOutput](ctx, s.natsConn, "eks.SetRecoveryDirective", input, defaultTimeout, accountID)
 }
 
+func (s *NATSEKSService) RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error) {
+	return utils.NATSRequest[RestoreSnapshotOutput](ctx, s.natsConn, "eks.RestoreSnapshot", input, defaultTimeout, accountID)
+}
+
 // --- OIDC identity-provider configs ---
 
 func (s *NATSEKSService) AssociateIdentityProviderConfig(ctx context.Context, input *eks.AssociateIdentityProviderConfigInput, accountID string) (*eks.AssociateIdentityProviderConfigOutput, error) {