feat: probe real qemu state (#839)

spinifex/gateway/ec2/instance/RebootInstances_test.go

@@ -113,19 +113,24 @@ func TestRebootInstances_StoppedInstance(t *testing.T) {
 }
 
 func TestRebootInstances_DaemonError(t *testing.T) {
-	_, nc := startTestNATSServer(t)
+	for _, code := range []string{
+		awserrors.ErrorIncorrectInstanceState,
+		awserrors.ErrorServerInternal,
+	} {
+		t.Run(code, func(t *testing.T) {
+			_, nc := startTestNATSServer(t)
+			instanceID := "i-error"
 
-	instanceID := "i-error"
+			nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
+				msg.Respond(utils.GenerateErrorPayload(code))
+			})
 
-	nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
-		msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorIncorrectInstanceState))
-	})
-
-	input := &ec2.RebootInstancesInput{
-		InstanceIds: []*string{aws.String(instanceID)},
+			input := &ec2.RebootInstancesInput{
+				InstanceIds: []*string{aws.String(instanceID)},
+			}
+			_, err := RebootInstances(context.Background(), input, nc, "123456789012")
+			require.Error(t, err)
+			assert.Equal(t, code, err.Error())
+		})
 	}
-
-	_, err := RebootInstances(context.Background(), input, nc, "123456789012")
-	require.Error(t, err)
-	assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
 }

spinifex/handlers/ec2/instance/service_impl_test.go

@@ -3582,6 +3582,17 @@ func TestRebootInstance_NotFound(t *testing.T) {
 	assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
 }
 
+func TestRebootInstance_QMPFailureReturnsInternalError(t *testing.T) {
+	id := "i-qmp-failure"
+	instance := &vm.VM{ID: id, Status: vm.StateRunning}
+	mgr := mgrWith(map[string]*vm.VM{id: instance})
+	svc := &InstanceServiceImpl{vmMgr: mgr}
+
+	err := svc.RebootInstance(context.Background(), instance, spxtypes.EC2InstanceCommand{ID: id})
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+}
+
 // TestStartInstance_NotFound verifies that a missing instance returns
 // InvalidInstanceID.NotFound rather than a generic internal error.
 func TestStartInstance_NotFound(t *testing.T) {

spinifex/vm/lifecycle.go

@@ -97,8 +97,13 @@ func (m *Manager) Start(ctx context.Context, id string) error {
 	return m.launch(ctx, instance)
 }
 
-// Reboot issues a QMP system_reset; the VM stays in StateRunning while QEMU
-// re-runs firmware. Returns ErrInstanceNotFound or ErrInvalidTransition as appropriate.
+const (
+	rebootRunningTimeout     = 4 * time.Second
+	rebootStatusPollInterval = 100 * time.Millisecond
+)
+
+// Reboot resets QEMU and verifies its vCPUs resume. The VM remains in
+// StateRunning while firmware re-runs.
 func (m *Manager) Reboot(ctx context.Context, id string) error {
 	instance, ok := m.Get(id)
 	if !ok {
@@ -108,10 +113,72 @@ func (m *Manager) Reboot(ctx context.Context, id string) error {
 		return fmt.Errorf("%w: cannot reboot instance %s in state %s",
 			ErrInvalidTransition, id, status)
 	}
-	if _, err := sendQMPCommand(ctx, instance.QMPClient, qmp.QMPCommand{Execute: "system_reset"}, id); err != nil {
+
+	rebootCtx, cancel := context.WithTimeout(ctx, rebootRunningTimeout)
+	defer cancel()
+
+	if _, err := sendQMPCommandWithTimeout(rebootCtx, instance.QMPClient,
+		qmp.QMPCommand{Execute: "system_reset"}, id, contextTimeRemaining(rebootCtx)); err != nil {
 		return fmt.Errorf("QMP system_reset: %w", err)
 	}
-	return nil
+	return m.waitForQMPRunning(rebootCtx, instance, rebootStatusPollInterval)
+}
+
+// waitForQMPRunning polls until QEMU reports running or ctx expires. Paused
+// and prelaunch guests receive cont so a reset cannot leave their vCPUs parked.
+func (m *Manager) waitForQMPRunning(ctx context.Context, instance *VM, pollInterval time.Duration) error {
+	var lastStatus qmp.Status
+	for {
+		if err := ctx.Err(); err != nil {
+			return fmt.Errorf("wait for QMP running (last status %q): %w", lastStatus.Status, err)
+		}
+		if status := m.Status(instance); status != StateRunning {
+			return fmt.Errorf("%w: reboot interrupted for instance %s in state %s",
+				ErrInvalidTransition, instance.ID, status)
+		}
+
+		qmpStatus, err := queryQMPStatus(ctx, instance, contextTimeRemaining(ctx))
+		if err != nil {
+			return fmt.Errorf("verify QMP running: %w", err)
+		}
+		lastStatus = qmpStatus
+		if qmpStatus.Running {
+			return nil
+		}
+
+		if qmpStatus.Status == "paused" || qmpStatus.Status == "prelaunch" {
+			if status := m.Status(instance); status != StateRunning {
+				return fmt.Errorf("%w: cannot resume instance %s in state %s",
+					ErrInvalidTransition, instance.ID, status)
+			}
+			if _, err := sendQMPCommandWithTimeout(ctx, instance.QMPClient,
+				qmp.QMPCommand{Execute: "cont"}, instance.ID, contextTimeRemaining(ctx)); err != nil {
+				return fmt.Errorf("QMP cont from status %q: %w", qmpStatus.Status, err)
+			}
+		}
+
+		timer := time.NewTimer(pollInterval)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return fmt.Errorf("wait for QMP running (last status %q): %w", lastStatus.Status, ctx.Err())
+		case <-timer.C:
+		}
+	}
+}
+
+// contextTimeRemaining returns the active context deadline budget. Reboot
+// always installs a deadline before issuing QMP commands.
+func contextTimeRemaining(ctx context.Context) time.Duration {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		return qmpCommandTimeout
+	}
+	remaining := time.Until(deadline)
+	if remaining <= 0 {
+		return time.Nanosecond
+	}
+	return remaining
 }
 
 // launchStillValid returns true when status is still pending/stopped/provisioning.
@@ -204,7 +271,7 @@ func (m *Manager) launch(ctx context.Context, instance *VM) (err error) {
 		return err
 	}
 	instance.QMPClient = qmpClient
-	go m.qmpHeartbeat(instance) //nolint:gosec // heartbeat outlives the launch request; must not inherit its cancellation
+	go m.qmpHeartbeat(instance)
 
 	m.Insert(instance)
 
@@ -811,17 +878,13 @@ const (
 	QMPMaxConsecutiveFailures = 3
 )
 
-// qmpHeartbeat polls query-status every qmpHeartbeatInterval and acts on
-// failures: it tracks consecutive failures, and once they reach
-// QMPMaxConsecutiveFailures it checks process liveness. A dead process triggers
-// crash recovery; an alive-but-wedged QEMU stays impaired for an operator. The
-// goroutine exits and closes the QMP connection on any terminal/transitional state.
+// qmpHeartbeat polls query-status every qmpHeartbeatInterval. A dead process
+// triggers recovery; a live but non-running QEMU remains impaired for operators.
 func (m *Manager) qmpHeartbeat(instance *VM) {
 	for {
 		time.Sleep(qmpHeartbeatInterval)
 
 		status := m.Status(instance)
-
 		if status == StateStopping || status == StateStopped ||
 			status == StateShuttingDown || status == StateTerminated || status == StateError {
 			slog.Info("QMP heartbeat exiting - instance not running", "instance", instance.ID, "status", status)
@@ -833,30 +896,62 @@ func (m *Manager) qmpHeartbeat(instance *VM) {
 			return
 		}
 
-		slog.Debug("QMP heartbeat", "instance", instance.ID)
-		qmpStatus, err := sendQMPCommand(context.WithValue(context.Background(), noTraceKey{}, true),
-			instance.QMPClient, qmp.QMPCommand{Execute: "query-status"}, instance.ID)
-		if err != nil {
-			failures := m.recordQMPFailure(instance)
-			slog.Warn("QMP heartbeat failed", "instance", instance.ID, "consecutiveFailures", failures, "err", err)
-			if failures < QMPMaxConsecutiveFailures {
-				continue
-			}
-			if !isInstanceProcessRunning(instance) {
-				slog.Error("QEMU process dead and QMP unresponsive, triggering crash recovery",
-					"instance", instance.ID, "consecutiveFailures", failures)
-				m.HandleCrash(instance, fmt.Errorf("qmp unresponsive (%d failures), process dead", failures))
-				return
-			}
-			slog.Error("QEMU process alive but QMP unresponsive", "instance", instance.ID, "consecutiveFailures", failures)
-			continue
+		if !m.qmpHeartbeatPoll(instance) {
+			return
 		}
-
-		m.recordQMPSuccess(instance)
-		slog.Debug("QMP status", "instance", instance.ID, "status", string(qmpStatus.Return))
 	}
 }
 
+// qmpHeartbeatPoll performs one status check and returns false after crash
+// recovery takes ownership.
+func (m *Manager) qmpHeartbeatPoll(instance *VM) bool {
+	slog.Debug("QMP heartbeat", "instance", instance.ID)
+	ctx := context.WithValue(context.Background(), noTraceKey{}, true)
+	qmpStatus, err := queryQMPStatus(ctx, instance, qmpCommandTimeout)
+	if err == nil && !qmpStatus.Running && m.Status(instance) == StateRunning {
+		err = fmt.Errorf("QMP reported non-running status %q", qmpStatus.Status)
+	}
+	if err != nil {
+		failures := m.recordQMPFailure(instance)
+		slog.Warn("QMP heartbeat failed", "instance", instance.ID, "consecutiveFailures", failures, "err", err)
+		if failures < QMPMaxConsecutiveFailures {
+			return true
+		}
+		if !isInstanceProcessRunning(instance) {
+			slog.Error("QEMU process dead and QMP unhealthy, triggering crash recovery",
+				"instance", instance.ID, "consecutiveFailures", failures)
+			m.HandleCrash(instance, fmt.Errorf("qmp unhealthy (%d failures), process dead: %w", failures, err))
+			return false
+		}
+		slog.Error("QEMU process alive but QMP unhealthy", "instance", instance.ID, "consecutiveFailures", failures)
+		return true
+	}
+
+	if qmpStatus.Running {
+		m.recordQMPSuccess(instance)
+	}
+	slog.Debug("QMP status", "instance", instance.ID, "status", qmpStatus.Status, "running", qmpStatus.Running)
+	return true
+}
+
+// queryQMPStatus decodes query-status so monitor responsiveness is not
+// mistaken for a running guest.
+func queryQMPStatus(ctx context.Context, instance *VM, timeout time.Duration) (qmp.Status, error) {
+	resp, err := sendQMPCommandWithTimeout(ctx, instance.QMPClient,
+		qmp.QMPCommand{Execute: "query-status"}, instance.ID, timeout)
+	if err != nil {
+		return qmp.Status{}, err
+	}
+	var status qmp.Status
+	if err := json.Unmarshal(resp.Return, &status); err != nil {
+		return qmp.Status{}, fmt.Errorf("decode query-status response: %w", err)
+	}
+	if status.Status == "" {
+		return qmp.Status{}, fmt.Errorf("decode query-status response: missing status")
+	}
+	return status, nil
+}
+
 // recordQMPFailure increments the consecutive QMP failure counter, stamping
 // ImpairedSince when the count first reaches QMPMaxConsecutiveFailures. Returns
 // the new count.

spinifex/vm/vm.go

@@ -25,9 +25,8 @@ type InstanceHealthState struct {
 	RestartCount    int       `json:"restart_count"`
 	FirstCrashTime  time.Time `json:"first_crash_time"`
 
-	// QMP health monitoring. QMPConsecutiveFailures counts back-to-back
-	// query-status failures; at QMPMaxConsecutiveFailures the instance is
-	// impaired and ImpairedSince is stamped. A successful poll clears all three.
+	// QMP health monitoring. Consecutive failures include transport errors and
+	// non-running runstates; only a running poll clears impairment.
 	QMPConsecutiveFailures int       `json:"qmp_consecutive_failures"`
 	LastQMPSuccess         time.Time `json:"last_qmp_success,omitzero"`
 	ImpairedSince          time.Time `json:"impaired_since,omitzero"`

spinifex/vm/volumes_test.go

@@ -2,11 +2,13 @@ package vm
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
 	"net"
+	"os"
 	"path/filepath"
 	"sync"
 	"sync/atomic"
@@ -204,6 +206,12 @@ func (r *qmpRecorder) executes() []string {
 	return out
 }
 
+func qmpStatusResponse(status string, running bool) map[string]any {
+	return map[string]any{"return": map[string]any{
+		"status": status, "running": running, "singlestep": false,
+	}}
+}
+
 func TestNextAvailableDevice(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -596,7 +604,10 @@ func TestReboot_DoesNotFireHooks(t *testing.T) {
 	recorder := &qmpRecorder{}
 	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
 		recorder.record(cmd)
-		return nil // success
+		if cmd.Execute == "query-status" {
+			return qmpStatusResponse("running", true)
+		}
+		return nil
 	})
 	defer cancel()
 
@@ -612,15 +623,20 @@ func TestReboot_DoesNotFireHooks(t *testing.T) {
 
 	assert.Equal(t, 0, upCalls, "Reboot must not fire OnInstanceUp")
 	assert.Equal(t, 0, downCalls, "Reboot must not fire OnInstanceDown")
-	assert.Equal(t, []string{"system_reset"}, recorder.executes(),
-		"Reboot must issue exactly one system_reset QMP command")
+	assert.Equal(t, []string{"system_reset", "query-status"}, recorder.executes(),
+		"Reboot must verify QEMU resumed after system_reset")
 }
 
 // TestReboot_DoesNotChangeStatus asserts the VM stays in StateRunning across
 // reboot. Pairs with TestReboot_DoesNotFireHooks to lock down the hook
 // contract: status doesn't transition, so hooks shouldn't fire.
 func TestReboot_DoesNotChangeStatus(t *testing.T) {
-	qmpClient, cancel := newMockQMPClient(t, nil)
+	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+		if cmd.Execute == "query-status" {
+			return qmpStatusResponse("running", true)
+		}
+		return nil
+	})
 	defer cancel()
 
 	m := NewManager()
@@ -652,6 +668,144 @@ func TestReboot_QMPFailureSurfacesError(t *testing.T) {
 	assert.Contains(t, err.Error(), "QMP system_reset")
 }
 
+func TestReboot_ResumesPausedRunstates(t *testing.T) {
+	for _, runstate := range []string{"paused", "prelaunch"} {
+		t.Run(runstate, func(t *testing.T) {
+			recorder := &qmpRecorder{}
+			statusQueries := 0
+			qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+				recorder.record(cmd)
+				if cmd.Execute != "query-status" {
+					return nil
+				}
+				statusQueries++
+				if statusQueries == 1 {
+					return qmpStatusResponse(runstate, false)
+				}
+				return qmpStatusResponse("running", true)
+			})
+			defer cancel()
+
+			m := NewManager()
+			m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
+
+			require.NoError(t, m.Reboot(t.Context(), "i-1"))
+			assert.Equal(t, []string{"system_reset", "query-status", "cont", "query-status"}, recorder.executes())
+		})
+	}
+}
+
+func TestReboot_NonRunningTimesOut(t *testing.T) {
+	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+		if cmd.Execute == "query-status" {
+			return qmpStatusResponse("shutdown", false)
+		}
+		return nil
+	})
+	defer cancel()
+
+	m := NewManager()
+	m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
+	ctx, cancelContext := context.WithTimeout(t.Context(), 25*time.Millisecond)
+	defer cancelContext()
+
+	err := m.Reboot(ctx, "i-1")
+	require.Error(t, err)
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.Contains(t, err.Error(), `last status "shutdown"`)
+}
+
+func TestReboot_MalformedStatusSurfacesError(t *testing.T) {
+	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+		if cmd.Execute == "query-status" {
+			return map[string]any{"return": map[string]any{}}
+		}
+		return nil
+	})
+	defer cancel()
+
+	m := NewManager()
+	m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
+
+	err := m.Reboot(t.Context(), "i-1")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing status")
+}
+
+func TestReboot_ContFailureSurfacesError(t *testing.T) {
+	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+		if cmd.Execute == "query-status" {
+			return qmpStatusResponse("paused", false)
+		}
+		if cmd.Execute == "cont" {
+			return map[string]any{"error": map[string]any{
+				"class": "GenericError", "desc": "cannot resume",
+			}}
+		}
+		return nil
+	})
+	defer cancel()
+
+	m := NewManager()
+	m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
+
+	err := m.Reboot(t.Context(), "i-1")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), `QMP cont from status "paused"`)
+}
+
+func TestReboot_StateChangePreventsCont(t *testing.T) {
+	var m *Manager
+	recorder := &qmpRecorder{}
+	qmpClient, cancel := newMockQMPClient(t, func(cmd qmp.QMPCommand) map[string]any {
+		recorder.record(cmd)
+		if cmd.Execute == "query-status" {
+			m.UpdateState("i-1", func(v *VM) { v.Status = StateStopping })
+			return qmpStatusResponse("paused", false)
+		}
+		return nil
+	})
+	defer cancel()
+
+	m = NewManager()
+	m.Insert(&VM{ID: "i-1", Status: StateRunning, QMPClient: qmpClient})
+
+	err := m.Reboot(t.Context(), "i-1")
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrInvalidTransition)
+	assert.Equal(t, []string{"system_reset", "query-status"}, recorder.executes())
+}
+
+func TestQMPHeartbeatPoll_UsesRunstate(t *testing.T) {
+	running := false
+	qmpClient, cancel := newMockQMPClient(t, func(qmp.QMPCommand) map[string]any {
+		if running {
+			return qmpStatusResponse("running", true)
+		}
+		return qmpStatusResponse("paused", false)
+	})
+	defer cancel()
+
+	m := NewManager()
+	instance := &VM{ID: "i-heartbeat-runstate", Status: StateRunning, QMPClient: qmpClient}
+	m.Insert(instance)
+	require.NoError(t, utils.WritePidFile(instance.ID, os.Getpid()))
+	t.Cleanup(func() { _ = utils.RemovePidFile(instance.ID) })
+
+	for range QMPMaxConsecutiveFailures {
+		assert.True(t, m.qmpHeartbeatPoll(instance))
+	}
+	assert.Equal(t, QMPMaxConsecutiveFailures, instance.Health.QMPConsecutiveFailures)
+	assert.False(t, instance.Health.ImpairedSince.IsZero())
+	assert.True(t, instance.Health.LastQMPSuccess.IsZero())
+
+	running = true
+	assert.True(t, m.qmpHeartbeatPoll(instance))
+	assert.Zero(t, instance.Health.QMPConsecutiveFailures)
+	assert.True(t, instance.Health.ImpairedSince.IsZero())
+	assert.False(t, instance.Health.LastQMPSuccess.IsZero())
+}
+
 // attachVolumeRunningInstance returns a manager with a single running VM
 // wired with the supplied QMP client. Used by AttachVolume rollback tests.
 func attachVolumeRunningInstance(t *testing.T, qmpClient *qmp.QMPClient, mounter VolumeMounter) (*Manager, *VM) {

tests/e2e/single/rebootliveness_test.go

@@ -0,0 +1,90 @@
+//go:build e2e
+
+package single
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/mulgadc/spinifex/tests/e2e/harness"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const earlyRebootRecoveryTimeout = 3 * time.Minute
+
+// runEarlyRebootLiveness reboots a fresh guest as soon as cloud-init.target is
+// observable. The control plane must either reject a stuck reset or boot again.
+func runEarlyRebootLiveness(t *testing.T, fix *Fixture) {
+	harness.Phase(t, "Single — early reboot resumes or reports an API error")
+
+	instType, _ := needInstanceTypeArch(t, fix)
+	keyName, keyPath := needKeyPair(t, fix)
+	amiID := needAMI(t, fix)
+	vpc := harness.EnsureDefaultVPC(t, fix.Harness)
+	require.NotEmpty(t, vpc.SGID, "default SG ID required")
+	harness.AuthorizeSSHIngress(t, fix.AWS, vpc.SGID)
+
+	harness.Step(t, "launch fresh guest for early reboot")
+	instanceID := launchBaselineInstance(t, fix, amiID, instType, keyName, vpc.SubnetID, []string{vpc.SGID})
+	inst := describeSingletonInstance(t, fix, instanceID)
+	host, port := harness.InstancePublicSSHHost(t, inst)
+	waitForSSHReady(t, host, port, keyPath)
+	tgt := harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath}
+
+	bootID := strings.TrimSpace(runSSH(t, tgt, "cat /proc/sys/kernel/random/boot_id"))
+	require.NotEmpty(t, bootID, "fresh guest returned an empty boot ID")
+	waitForCloudInitTarget(t, tgt)
+
+	harness.Step(t, "reboot-instances %s immediately after cloud-init.target", instanceID)
+	_, err := fix.AWS.EC2.RebootInstances(&ec2.RebootInstancesInput{
+		InstanceIds: []*string{aws.String(instanceID)},
+	})
+	if err != nil {
+		var apiErr awserr.Error
+		require.ErrorAs(t, err, &apiErr, "reboot failure must be an AWS API error")
+		assert.Equal(t, "InternalError", apiErr.Code(), "a liveness failure must surface as InternalError")
+		harness.Detail(t, "reboot_outcome", "api_error", "code", apiErr.Code())
+		return
+	}
+
+	deadline := time.Now().Add(earlyRebootRecoveryTimeout)
+	for {
+		ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+		out, sshErr := harness.RunGuestSSH(ctx, tgt, "cat /proc/sys/kernel/random/boot_id")
+		cancel()
+		if sshErr == nil {
+			newBootID := strings.TrimSpace(string(out))
+			if newBootID != "" && newBootID != bootID {
+				harness.Detail(t, "reboot_outcome", "resumed", "boot_id", newBootID)
+				return
+			}
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("reboot API succeeded but guest boot ID did not change within %s", earlyRebootRecoveryTimeout)
+		}
+		time.Sleep(2 * time.Second)
+	}
+}
+
+func waitForCloudInitTarget(t *testing.T, tgt harness.SSHTarget) {
+	t.Helper()
+	deadline := time.Now().Add(60 * time.Second)
+	for {
+		ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+		out, err := harness.RunGuestSSH(ctx, tgt, "systemctl is-active cloud-init.target")
+		cancel()
+		if err == nil && strings.TrimSpace(string(out)) == "active" {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("cloud-init.target did not become active within 1m (last output %q, err: %v)", strings.TrimSpace(string(out)), err)
+		}
+		time.Sleep(500 * time.Millisecond)
+	}
+}

tests/e2e/single/tests_test.go

@@ -76,12 +76,14 @@ func TestVolumeStatus(t *testing.T) {
 	runVolumeStatus(t, requireSingleNodeFixture(t))
 }
 
-// TestGuestChurnDurability merges the former TestSSHProbe, TestENIHotplug,
-// TestENIHotplugReconcile, TestVolumeDurability, TestModifyInstanceAttribute,
-// and TestRebootInstance around one shared guest and one shared
-// data-integrity sentinel (see runGuestChurnDurability for the stage
-// breakdown and gating rationale). Sequential: it stops/starts and reboots
-// the singleton, leaving it running.
+// TestEarlyRebootLiveness owns a fresh guest so the reset lands as soon as
+// cloud-init.target is observable rather than against the old singleton.
+func TestEarlyRebootLiveness(t *testing.T) {
+	runEarlyRebootLiveness(t, requireSingleNodeFixture(t))
+}
+
+// TestGuestChurnDurability merges guest churn around one shared instance and
+// data-integrity sentinel. Sequential: it leaves the singleton running.
 func TestGuestChurnDurability(t *testing.T) {
 	runGuestChurnDurability(t, requireSingleNodeFixture(t))
 }