MIR-857: Back off every sandbox that fails before reaching RUNNING

api/compute/schema.yml

@@ -239,6 +239,11 @@ kinds:
       doc: The status of the pod
       choices: [pending, not_ready, running, stopped, dead]
 
+    startup_outcome:
+      type: enum
+      choices: [startup_running, startup_failed]
+      doc: Durable distinction between a sandbox that reached RUNNING and one that failed before then
+
     last_activity:
       type: time
       doc: Last lease activity (throttled updates, ~30s granularity for scale-down)
@@ -503,11 +508,11 @@ kinds:
     # Crash loop detection and cooldown (managed by SandboxPoolManager)
     consecutive_crash_count:
       type: int
-      doc: Number of consecutive quick crashes (sandboxes that died within 60s of creation)
+      doc: Number of consecutive sandbox failures, including quick crashes after RUNNING
 
     last_crash_time:
       type: time
-      doc: Timestamp of the most recent quick crash
+      doc: Timestamp of the most recently counted sandbox failure
 
     cooldown_until:
       type: time

controllers/nodehealth/controller.go

@@ -242,12 +242,21 @@ func (c *Controller) markNodeSandboxesDead(ctx context.Context, nodeID entity.Id
 			"node", nodeID,
 			"previous_status", sb.Status)
 
+		dead := &compute_v1alpha.Sandbox{Status: compute_v1alpha.DEAD}
+		if sb.StartupOutcome == "" {
+			switch sb.Status {
+			case compute_v1alpha.PENDING, compute_v1alpha.NOT_READY:
+				dead.StartupOutcome = compute_v1alpha.STARTUP_FAILED
+			case compute_v1alpha.RUNNING:
+				dead.StartupOutcome = compute_v1alpha.STARTUP_RUNNING
+			case compute_v1alpha.STOPPED, compute_v1alpha.DEAD:
+				// STOPPED may be an intentional retirement; DEAD was skipped above.
+			}
+		}
 		_, err := c.eac.Patch(ctx, entity.New(
 			entity.DBId, sb.ID,
-			(&compute_v1alpha.Sandbox{
-				Status: compute_v1alpha.DEAD,
-			}).Encode,
-		).Attrs(), 0)
+			dead.Encode,
+		).Attrs(), e.Revision())
 		if err != nil {
 			c.log.Error("failed to mark sandbox dead", "sandbox", sb.ID, "error", err)
 			patchErr = errors.Join(patchErr, err)

controllers/nodehealth/controller_test.go

@@ -156,7 +156,9 @@ func TestNonReadyNodeGracePeriodExpired(t *testing.T) {
 	require.NoError(t, err)
 
 	sbRunning := createScheduledSandbox(t, ctx, server, "sb-running", nodeID, compute_v1alpha.RUNNING)
+	server.Store.NowFunc = func() time.Time { return time.Now().Add(-6 * time.Minute) }
 	sbPending := createScheduledSandbox(t, ctx, server, "sb-pending", nodeID, compute_v1alpha.PENDING)
+	server.Store.NowFunc = nil
 
 	ctrl := NewController(testutils.TestLogger(t), server.EAC)
 	ctrl.gracePeriod = 5 * time.Minute
@@ -175,6 +177,19 @@ func TestNonReadyNodeGracePeriodExpired(t *testing.T) {
 		"RUNNING sandbox should be marked DEAD after grace period")
 	assert.Equal(t, compute_v1alpha.DEAD, getSandboxStatus(t, ctx, server, sbPending),
 		"PENDING sandbox should be marked DEAD after grace period")
+	for _, tc := range []struct {
+		id   entity.Id
+		want compute_v1alpha.SandboxStartupOutcome
+	}{
+		{sbRunning, compute_v1alpha.STARTUP_RUNNING},
+		{sbPending, compute_v1alpha.STARTUP_FAILED},
+	} {
+		resp, err := server.EAC.Get(ctx, tc.id.String())
+		require.NoError(t, err)
+		var sb compute_v1alpha.Sandbox
+		sb.Decode(resp.Entity().Entity())
+		assert.Equal(t, tc.want, sb.StartupOutcome)
+	}
 }
 
 func TestNodeRecoversWithinGracePeriod(t *testing.T) {

controllers/sandbox/create_saga.go

@@ -2,6 +2,7 @@ package sandbox
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net/netip"
@@ -493,6 +494,9 @@ func waitPorts(ctx context.Context, in waitPortsIn) (waitPortsOut, error) {
 		if err == nil {
 			continue // configured port bound — the normal case
 		}
+		if errors.Is(err, errProcessExited) {
+			return waitPortsOut{}, err
+		}
 		if ctx.Err() != nil {
 			// We're shutting down, not looking at a port mismatch — skip
 			// diagnosis so we don't emit misleading "listening elsewhere" events.
@@ -616,6 +620,7 @@ func setRunning(ctx context.Context, in setRunningIn) (setRunningOut, error) {
 		func() []entity.Attr {
 			attrs := []entity.Attr{
 				entity.Ref(compute.SandboxStatusId, compute.SandboxStatusRunningId),
+				entity.Ref(compute.SandboxStartupOutcomeId, compute.SandboxStartupOutcomeStartupRunningId),
 			}
 			for _, op := range in.ObservedPorts {
 				bp := compute.BoundPort{Port: int64(op.Port), Address: op.Address}

controllers/sandbox/create_saga_test.go

@@ -556,6 +556,15 @@ func TestCreateSandboxSaga_HappyPath(t *testing.T) {
 	assert.Equal(t, 1, h.runtime.waitForPortCalls)
 	assert.Equal(t, 1, h.obs.addMetricsCalls)
 	assert.Equal(t, 1, h.obs.updateSvcsCalls)
+	runningOutcome := false
+	for _, attrs := range h.entities.patchCalls {
+		for _, attr := range attrs {
+			if attr.ID == compute.SandboxStartupOutcomeId && attr.Value.Id() == compute.SandboxStartupOutcomeStartupRunningId {
+				runningOutcome = true
+			}
+		}
+	}
+	assert.True(t, runningOutcome, "reaching RUNNING must persist the startup outcome")
 
 	// No undo actions called
 	assert.Equal(t, 0, h.networking.releaseCalls)
@@ -717,6 +726,20 @@ func TestCreateSandboxSaga_WaitPortsFails(t *testing.T) {
 	assert.Equal(t, 1, h.networking.releaseCalls)
 }
 
+func TestCreateSandboxSaga_WaitPortsProcessExit(t *testing.T) {
+	h := newTestHarness(t)
+	h.runtime.waitForPortErr = errProcessExited
+	h.runtime.mockContainer.taskFn = func(ctx context.Context, attach cio.Attach) (containerd.Task, error) {
+		return h.runtime.mockTask, nil
+	}
+
+	err := h.execute(t)
+	require.ErrorContains(t, err, errProcessExited.Error())
+	assert.Equal(t, 0, h.runtime.diagnoseListeningCalls,
+		"an exited process cannot be auto-routed to a different port")
+	assert.Equal(t, saga.StatusFailed, h.execution(t).Status)
+}
+
 func TestSingleAlternativePort(t *testing.T) {
 	// Exactly one routable port that isn't the configured one -> auto-routable.
 	alt, ok := singleAlternativePort([]int{3000}, 8080)

controllers/sandbox/saga_controller.go

@@ -89,11 +89,25 @@ func (c *SandboxController) createSandboxViaSaga(ctx context.Context, co *comput
 		// NOTE: this runs at the call site, so a crash between saga completion
 		// and this patch leaves the entity PENDING (retried by reconciler).
 		// Durable saga outcome declaration is future work.
+		current, meta, getErr := c.ops.GetSandbox(ctx, co.ID.String())
+		failure := &compute.Sandbox{Status: compute.DEAD}
+		var revision int64
+		if getErr != nil {
+			c.Log.Warn("failed to fetch sandbox after saga failure; leaving startup outcome unchanged", "id", co.ID, "error", getErr)
+		} else {
+			if current.Status == compute.DEAD {
+				return fmt.Errorf("saga sandbox creation failed: %w", err)
+			}
+			revision = meta.GetRevision()
+			if current.StartupOutcome != compute.STARTUP_RUNNING && current.Status != compute.RUNNING {
+				failure.StartupOutcome = compute.STARTUP_FAILED
+			}
+		}
 		patchAttrs := entity.New(
 			entity.Ref(entity.DBId, co.ID),
-			(&compute.Sandbox{Status: compute.DEAD}).Encode,
+			failure.Encode,
 		)
-		if _, patchErr := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), 0); patchErr != nil {
+		if _, patchErr := c.ops.PatchSandbox(ctx, patchAttrs.Attrs(), revision); patchErr != nil {
 			c.Log.Error("failed to mark sandbox DEAD after saga failure", "id", co.ID, "error", patchErr)
 		}
 

controllers/sandbox/saga_controller_test.go

@@ -2,6 +2,7 @@ package sandbox
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"testing"
 	"time"
@@ -14,6 +15,51 @@ import (
 	"miren.dev/runtime/pkg/saga"
 )
 
+type sagaFailureOps struct {
+	SandboxEntityStore
+	SandboxNetworking
+	SandboxContainerRuntime
+	SandboxObservability
+}
+
+func TestSagaFailureMarksDeadWhenFinalFetchFails(t *testing.T) {
+	h := newTestHarness(t)
+	h.networking.allocateErr = errors.New("no IPs available")
+	h.entities.getSandboxFunc = func(_ context.Context, _ string) (*compute.Sandbox, *entity.Meta, error) {
+		if h.entities.getCalls > 1 {
+			return nil, nil, errors.New("transient fetch failure")
+		}
+		return h.entities.sandbox, h.entities.meta, nil
+	}
+	c := &SandboxController{
+		Log: slog.Default(), ops: sagaFailureOps{SandboxEntityStore: h.entities},
+		executor: h.executor, sagaStorage: h.storage,
+	}
+	err := c.createSandboxViaSaga(context.Background(), h.entities.sandbox, false)
+	require.ErrorContains(t, err, "no IPs available")
+	require.Len(t, h.entities.patchCalls, 1)
+	patch := entity.New(h.entities.patchCalls[0])
+	status, ok := patch.Get(compute.SandboxStatusId)
+	require.True(t, ok)
+	assert.Equal(t, compute.SandboxStatusDeadId, status.Value.Id())
+	_, outcomeSet := patch.Get(compute.SandboxStartupOutcomeId)
+	assert.False(t, outcomeSet, "a failed fetch must not derive an outcome from the stale snapshot")
+}
+
+func TestSagaFailureDoesNotRewriteDead(t *testing.T) {
+	h := newTestHarness(t)
+	h.networking.allocateErr = errors.New("no IPs available")
+	h.entities.sandbox.Status = compute.DEAD
+	h.entities.sandbox.StartupOutcome = compute.STARTUP_FAILED
+	c := &SandboxController{
+		Log: slog.Default(), ops: sagaFailureOps{SandboxEntityStore: h.entities},
+		executor: h.executor, sagaStorage: h.storage,
+	}
+	err := c.createSandboxViaSaga(context.Background(), h.entities.sandbox, false)
+	require.ErrorContains(t, err, "no IPs available")
+	require.Empty(t, h.entities.patchCalls, "already-DEAD sandbox must retain its failure timestamp")
+}
+
 // newSagaControllerForResume wires up only what sagaResumeNeeded reads
 // (storage + log), so no live containerd client is needed.
 func newSagaControllerForResume(t *testing.T) *SandboxController {