fix: keep the ecs-agent dialling containerd and brake a deployment that cannot start (#1090)

The agent resolved the container runtime once, in New. A containerd socket that was not accepting during the first second of boot left the agent runtime-less for the life of the process, and nothing retried. It registered and heartbeated normally, so the scheduler saw a healthy instance and kept placing tasks that died immediately at the nil runtime check.

The runtime now lives behind a gate that a watch loop keeps current: it dials while the gate is empty, and probes with a container list while it is full, clearing the gate when the probe fails so the next tick re-dials. That covers a socket that is not ready yet at boot and a containerd restarted underneath a running agent alike, and everything that used the puller or the runner reads through the gate, so a runtime arriving late is picked up without a restart.

An agent with no runtime advertises no capacity. Registration and the heartbeat report zero CPU and zero memory while the gate is empty, so placement skips the host, and the control plane reads that pair of zeros as an involuntary drain: the instance becomes DRAINING with agentConnected false and zero registered resources, which is visible through DescribeContainerInstances rather than only in the node's log. Capacity returning clears it through the branch a reaped instance already used. A host with no CPU cannot exist, so the zero needs no new wire field. An operator drain is no longer cleared by a bus registration either, which it previously was.

A deployment whose tasks keep failing to start now backs off. Deployment gains NextLaunchAt, and past the failure count the circuit breaker already tracks, the reconciler widens the gap between launch passes up to a cap. It is keyed on that counter but applies whether or not the breaker is enabled, because AWS leaves it off by default and an unbraked service relaunches a task that cannot start on every pass, forever.

cmd/ecs-agent/agent.go

@@ -30,10 +30,13 @@ type Agent struct {
 	cfg      config
 	id       identity
 	cp       controlPlane
-	puller   ctrruntime.ImagePuller
-	runner   ctrruntime.Runner
 	resolver ctrruntime.Resolver
 
+	// rt holds the container runtime, which is resolved by a loop rather than
+	// once at boot. Empty until the containerd socket answers, and emptied again
+	// if it stops answering.
+	rt *runtimeGate
+
 	reg *registrar
 	hb  *heartbeater
 
@@ -48,26 +51,34 @@ type Agent struct {
 	// nvidia-smi discovery.
 	gpu *gpuLedger
 
+	// dialRuntime connects to the container runtime. nil in unit tests, which
+	// drive the gate directly.
+	dialRuntime func() (ctrruntime.Runtime, error)
+
 	closers []func() error
 }
 
 // newAgent assembles an Agent from already-built seams. Tests use this directly
-// with fakes; New builds the production seams and delegates here. runner may be
-// nil when containerd is unavailable; the assign path then reports the task
-// STOPPED rather than crashing.
-func newAgent(cfg config, id identity, cp controlPlane, puller ctrruntime.ImagePuller, runner ctrruntime.Runner, resolver ctrruntime.Resolver) *Agent {
-	return &Agent{
+// with fakes; New builds the production seams and delegates here. rt may be nil
+// when containerd is unavailable; the assign path then reports the task STOPPED
+// rather than crashing, and the watch loop keeps dialling for it.
+func newAgent(cfg config, id identity, cp controlPlane, rt ctrruntime.Runtime, resolver ctrruntime.Resolver) *Agent {
+	gate := newRuntimeGate(rt)
+	a := &Agent{
 		cfg:      cfg,
 		id:       id,
 		cp:       cp,
-		puller:   puller,
-		runner:   runner,
 		resolver: resolver,
-		reg:      newRegistrar(cp, id),
-		hb:       newHeartbeater(cp, id, cfg.Heartbeat),
+		rt:       gate,
 		netns:    newTaskNetns(execNetRunner{}),
 		gpu:      newGPULedger(id.Capacity.GPUIDs),
 	}
+	// Registration reports the host's capacity only while a runtime is there to
+	// use it, so an agent that cannot run a container advertises none and takes
+	// no placement.
+	a.reg = newRegistrar(cp, id, gate.available)
+	a.hb = newHeartbeater(cp, id, cfg.Heartbeat, gate.available)
+	return a
 }
 
 // New builds an Agent from config: it resolves the host identity from IMDS,
@@ -98,13 +109,12 @@ func New(cfg config) (*Agent, error) {
 	creds := credentials.NewIMDSProvider(imdsClient, cfg.IMDSBase)
 	resolver := newLazyECRResolver(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA)
 
-	var puller ctrruntime.ImagePuller
-	var runner ctrruntime.Runner
-	if rt, perr := ctrruntime.New(cfg.ContainerdSocket); perr != nil {
-		slog.Warn("ecs-agent: containerd unavailable at boot, image pulls disabled", "err", perr)
-	} else {
-		puller = rt
-		runner = rt
+	// A dial that fails here is not fatal and not final: the node starts the
+	// container daemon and the agent back to back, so the socket is often not
+	// accepting yet. Run's watch loop keeps dialling.
+	rt, perr := ctrruntime.New(cfg.ContainerdSocket)
+	if perr != nil {
+		slog.Warn("ecs-agent: containerd unavailable at boot, will keep retrying", "err", perr)
 	}
 
 	cp, err := newGatewayControlPlane(cfg, creds)
@@ -112,12 +122,10 @@ func New(cfg config) (*Agent, error) {
 		return nil, fmt.Errorf("build gateway control-plane: %w", err)
 	}
 
-	a := newAgent(cfg, id, cp, puller, runner, resolver)
+	a := newAgent(cfg, id, cp, rt, resolver)
 	a.cred = newCredEndpoint(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA,
 		cfg.CredEndpointIP, cfg.CredEndpointPort, execNetRunner{})
-	if puller != nil {
-		a.closers = append(a.closers, puller.Close)
-	}
+	a.dialRuntime = func() (ctrruntime.Runtime, error) { return ctrruntime.New(cfg.ContainerdSocket) }
 	return a, nil
 }
 
@@ -138,6 +146,12 @@ func (a *Agent) Run(ctx context.Context) error {
 
 	go a.hb.Run(ctx)
 
+	// Keeps the runtime current for the life of the process: dials while there
+	// is none, and drops one that stops answering.
+	if a.dialRuntime != nil {
+		go a.rt.watch(ctx, a.dialRuntime, a.cfg.RuntimeRetry)
+	}
+
 	// Re-adopt containers still running from before this restart, then poll with
 	// their tasks pre-seeded so re-delivered assignments are acked, not re-run.
 	go a.pollAssignments(ctx, a.reconcile(ctx))
@@ -157,6 +171,7 @@ func (a *Agent) Stop() error {
 			firstErr = err
 		}
 	}
+	a.rt.clear()
 	return firstErr
 }
 

cmd/ecs-agent/agent_test.go

@@ -12,7 +12,7 @@ func TestAgent_RunRegistersThenStopsOnContext(t *testing.T) {
 	cp := &fakeCP{}
 	cfg := config{Heartbeat: 5 * time.Millisecond, PollInterval: 5 * time.Millisecond}
 	puller := &ctrruntime.FakePuller{}
-	a := newAgent(cfg, testIdentity(), cp, puller, puller, nil)
+	a := newAgent(cfg, testIdentity(), cp, puller, nil)
 	a.closers = append(a.closers, puller.Close)
 
 	ctx, cancel := context.WithCancel(context.Background())

cmd/ecs-agent/config.go

@@ -33,6 +33,7 @@ type config struct {
 	ContainerdSocket string
 	Heartbeat        time.Duration
 	PollInterval     time.Duration
+	RuntimeRetry     time.Duration
 	CredEndpointIP   string
 	CredEndpointPort int
 }
@@ -75,6 +76,12 @@ func loadConfig(envFile string) config {
 			cfg.PollInterval = d
 		}
 	}
+	cfg.RuntimeRetry = defaultRuntimeRetry
+	if v := get("ECS_RUNTIME_RETRY"); v != "" {
+		if d, err := time.ParseDuration(v); err == nil && d > 0 {
+			cfg.RuntimeRetry = d
+		}
+	}
 	cfg.CredEndpointIP = get("ECS_CRED_ENDPOINT_IP")
 	if cfg.CredEndpointIP == "" {
 		cfg.CredEndpointIP = defaultCredEndpointIP

cmd/ecs-agent/heartbeat.go

@@ -14,18 +14,23 @@ type heartbeater struct {
 	cp       controlPlane
 	id       identity
 	interval time.Duration
+	// runtimeReady reports whether a container runtime is resolved, so a beat
+	// advertises capacity only while there is something to run it with. nil
+	// means always ready.
+	runtimeReady func() bool
 }
 
-func newHeartbeater(cp controlPlane, id identity, interval time.Duration) *heartbeater {
+func newHeartbeater(cp controlPlane, id identity, interval time.Duration, runtimeReady func() bool) *heartbeater {
 	if interval <= 0 {
 		interval = defaultHeartbeat
 	}
-	return &heartbeater{cp: cp, id: id, interval: interval}
+	return &heartbeater{cp: cp, id: id, interval: interval, runtimeReady: runtimeReady}
 }
 
 // beat re-registers the instance once, refreshing its LastSeen.
 func (h *heartbeater) beat() error {
-	return h.cp.Register(h.id)
+	ready := h.runtimeReady == nil || h.runtimeReady()
+	return h.cp.Register(advertisedIdentity(h.id, ready))
 }
 
 // Run beats every interval until ctx is cancelled. A failed beat is logged and

cmd/ecs-agent/reconcile.go

@@ -24,10 +24,11 @@ import (
 // agent falls back to its pre-reconcile behaviour (poll + run).
 func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 	adopted := map[string]bool{}
-	if a.runner == nil {
+	runner := a.rt.get()
+	if runner == nil {
 		return adopted
 	}
-	containers, err := a.runner.List(ctx)
+	containers, err := runner.List(ctx)
 	if err != nil {
 		slog.Warn("ecs-agent: reconcile list failed", "err", err)
 		return adopted
@@ -96,9 +97,13 @@ func (a *Agent) reconcile(ctx context.Context) map[string]bool {
 // is STOPPED, so the scheduler stops counting a task this instance is not
 // running and its service can place a replacement.
 func (a *Agent) reportStoppedTask(ctx context.Context, as *bus.Assign, dead []ctrruntime.Container) {
+	runner := a.rt.get()
 	statuses := make([]bus.ContainerStatus, 0, len(dead))
 	for _, c := range dead {
-		if err := a.runner.Remove(ctx, c.ID); err != nil {
+		if runner == nil {
+			break
+		}
+		if err := runner.Remove(ctx, c.ID); err != nil {
 			slog.Warn("ecs-agent: reconcile remove failed", "task", as.TaskID, "container", c.ID, "err", err)
 		}
 		statuses = append(statuses, bus.ContainerStatus{

cmd/ecs-agent/reconcile_test.go

@@ -55,7 +55,7 @@ func TestReconcile_AdoptsRunningLabeledContainer(t *testing.T) {
 			adoptedContainer("t-001", "web", "default", "t-001", testTaskRole, "52:54:00:de:ad:01", true),
 		},
 	}
-	a := newAgent(config{}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentity(), cp, rt, nil)
 	a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
 
 	adopted := a.reconcile(context.Background())
@@ -84,7 +84,7 @@ func TestReconcile_ReportsTaskWhoseContainerDied(t *testing.T) {
 			adoptedContainer("t-dead", "web", "default", "t-dead", testTaskRole, "52:54:00:de:ad:02", false),
 		},
 	}
-	a := newAgent(config{}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentity(), cp, rt, nil)
 	a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
 
 	adopted := a.reconcile(context.Background())
@@ -116,7 +116,7 @@ func TestReconcile_PartiallyDeadTaskStaysRunning(t *testing.T) {
 			adoptedContainer("t-mixed", "sidecar", "default", "t-mixed", testTaskRole, "52:54:00:de:ad:03", false),
 		},
 	}
-	a := newAgent(config{}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentity(), cp, rt, nil)
 	a.cred = newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil)
 
 	a.reconcile(context.Background())
@@ -145,7 +145,7 @@ func TestPollAssignments_SeededTaskNotRerun(t *testing.T) {
 		nil,
 	}}
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go a.pollAssignments(ctx, map[string]bool{"t-001": true})
@@ -197,7 +197,7 @@ func TestReconcile_IgnoresOtherClusterContainers(t *testing.T) {
 			adoptedContainer("t-ok", "web", "default", "", "", "", true),
 		},
 	}
-	a := newAgent(config{}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentity(), cp, rt, nil)
 
 	adopted := a.reconcile(context.Background())
 
@@ -213,13 +213,13 @@ func TestReconcile_IgnoresOtherClusterContainers(t *testing.T) {
 
 // A nil runner or a List error degrades to an empty set without panicking.
 func TestReconcile_DegradesGracefully(t *testing.T) {
-	noRunner := newAgent(config{}, testIdentity(), &fakeCP{}, nil, nil, nil)
+	noRunner := newAgent(config{}, testIdentity(), &fakeCP{}, nil, nil)
 	if got := noRunner.reconcile(context.Background()); len(got) != 0 {
 		t.Errorf("nil runner: adopted = %v, want empty", got)
 	}
 
 	rt := &ctrruntime.FakePuller{ListErr: errors.New("boom")}
-	listErr := newAgent(config{}, testIdentity(), &fakeCP{}, rt, rt, nil)
+	listErr := newAgent(config{}, testIdentity(), &fakeCP{}, rt, nil)
 	if got := listErr.reconcile(context.Background()); len(got) != 0 {
 		t.Errorf("list error: adopted = %v, want empty", got)
 	}

cmd/ecs-agent/registration.go

@@ -39,20 +39,39 @@ func instanceAttributes(id identity) []*ecs.Attribute {
 	return attrs
 }
 
+// advertisedIdentity is the identity to register with, given whether a
+// container runtime is available. A host with no runtime can run nothing, so it
+// advertises no capacity: the scheduler places nothing on it, and the control
+// plane reads the zero as an involuntary drain. The identity is otherwise
+// unchanged, so the instance stays visible rather than disappearing.
+func advertisedIdentity(id identity, runtimeReady bool) identity {
+	if runtimeReady {
+		return id
+	}
+	id.Capacity = bus.InstanceCapacity{}
+	return id
+}
+
 // registrar registers the host as a container instance through the gateway.
 type registrar struct {
 	cp controlPlane
 	id identity
+	// runtimeReady reports whether a container runtime is resolved. nil means
+	// always ready, which is what a test with no gate wants.
+	runtimeReady func() bool
 }
 
-func newRegistrar(cp controlPlane, id identity) *registrar {
-	return &registrar{cp: cp, id: id}
+func newRegistrar(cp controlPlane, id identity, runtimeReady func() bool) *registrar {
+	return &registrar{cp: cp, id: id, runtimeReady: runtimeReady}
 }
 
+// ready reports the runtime state, defaulting to ready when no gate is wired.
+func (r *registrar) ready() bool { return r.runtimeReady == nil || r.runtimeReady() }
+
 // Register calls the gateway's RegisterContainerInstance. The scheduler records
 // (or refreshes) the container instance on receipt.
 func (r *registrar) Register() error {
-	if err := r.cp.Register(r.id); err != nil {
+	if err := r.cp.Register(advertisedIdentity(r.id, r.ready())); err != nil {
 		return fmt.Errorf("register instance %s: %w", r.id.InstanceID, err)
 	}
 	return nil

cmd/ecs-agent/registration_test.go

@@ -18,15 +18,16 @@ import (
 type fakeCP struct {
 	mu sync.Mutex
 
-	registers    int
-	states       []bus.TaskState
-	gpuReports   []gpuReport
-	pollAcks     [][]string
-	pollStopAcks [][]string
-	pollReplies  [][]bus.Assign
-	stopReplies  [][]bus.StopDirective
-	stopPending  []bus.StopDirective
-	pollCalls    int
+	registers      int
+	lastRegistered identity
+	states         []bus.TaskState
+	gpuReports     []gpuReport
+	pollAcks       [][]string
+	pollStopAcks   [][]string
+	pollReplies    [][]bus.Assign
+	stopReplies    [][]bus.StopDirective
+	stopPending    []bus.StopDirective
+	pollCalls      int
 
 	registerErr bool
 	submitErr   bool
@@ -41,16 +42,24 @@ type gpuReport struct {
 
 var _ controlPlane = (*fakeCP)(nil)
 
-func (f *fakeCP) Register(_ identity) error {
+func (f *fakeCP) Register(id identity) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	if f.registerErr {
 		return errors.New("register boom")
 	}
 	f.registers++
+	f.lastRegistered = id
 	return nil
 }
 
+// registered returns the identity of the most recent Register call.
+func (f *fakeCP) registered() identity {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.lastRegistered
+}
+
 func (f *fakeCP) SubmitTaskState(st bus.TaskState) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -150,7 +159,7 @@ func testIdentityWithGPUs(uuids ...string) identity {
 
 func TestRegistrar_Register(t *testing.T) {
 	cp := &fakeCP{}
-	if err := newRegistrar(cp, testIdentity()).Register(); err != nil {
+	if err := newRegistrar(cp, testIdentity(), nil).Register(); err != nil {
 		t.Fatalf("Register: %v", err)
 	}
 	if cp.registerCount() != 1 {
@@ -160,14 +169,14 @@ func TestRegistrar_Register(t *testing.T) {
 
 func TestRegistrar_RegisterError(t *testing.T) {
 	cp := &fakeCP{registerErr: true}
-	if err := newRegistrar(cp, testIdentity()).Register(); err == nil {
+	if err := newRegistrar(cp, testIdentity(), nil).Register(); err == nil {
 		t.Fatal("expected register error")
 	}
 }
 
 func TestHeartbeater_Beat(t *testing.T) {
 	cp := &fakeCP{}
-	if err := newHeartbeater(cp, testIdentity(), time.Second).beat(); err != nil {
+	if err := newHeartbeater(cp, testIdentity(), time.Second, nil).beat(); err != nil {
 		t.Fatalf("beat: %v", err)
 	}
 	if cp.registerCount() != 1 {
@@ -176,7 +185,7 @@ func TestHeartbeater_Beat(t *testing.T) {
 }
 
 func TestHeartbeater_DefaultInterval(t *testing.T) {
-	h := newHeartbeater(&fakeCP{}, testIdentity(), 0)
+	h := newHeartbeater(&fakeCP{}, testIdentity(), 0, nil)
 	if h.interval != defaultHeartbeat {
 		t.Errorf("interval = %v, want default", h.interval)
 	}
@@ -184,7 +193,7 @@ func TestHeartbeater_DefaultInterval(t *testing.T) {
 
 func TestHeartbeater_RunStopsOnContext(t *testing.T) {
 	cp := &fakeCP{}
-	h := newHeartbeater(cp, testIdentity(), 5*time.Millisecond)
+	h := newHeartbeater(cp, testIdentity(), 5*time.Millisecond, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	done := make(chan struct{})
@@ -201,3 +210,48 @@ func TestHeartbeater_RunStopsOnContext(t *testing.T) {
 		t.Error("expected at least one re-register before cancel")
 	}
 }
+
+// TestRegistrar_NoRuntimeAdvertisesNoCapacity is the placement half of the boot
+// bug: an agent with no container runtime kept registering its full CPU and
+// memory, so the scheduler placed tasks it could only fail to start. Capacity is
+// advertised only while a runtime is there to use it.
+func TestRegistrar_NoRuntimeAdvertisesNoCapacity(t *testing.T) {
+	cp := &fakeCP{}
+	ready := false
+	r := newRegistrar(cp, testIdentity(), func() bool { return ready })
+
+	if err := r.Register(); err != nil {
+		t.Fatalf("register: %v", err)
+	}
+	if got := cp.registered().Capacity; got.CPU != 0 || got.MemoryMiB != 0 {
+		t.Fatalf("advertised capacity with no runtime: cpu=%d memory=%d", got.CPU, got.MemoryMiB)
+	}
+
+	ready = true
+	if err := r.Register(); err != nil {
+		t.Fatalf("register: %v", err)
+	}
+	want := testIdentity().Capacity
+	if got := cp.registered().Capacity; got.CPU != want.CPU || got.MemoryMiB != want.MemoryMiB {
+		t.Fatalf("capacity not restored once the runtime resolved: cpu=%d memory=%d", got.CPU, got.MemoryMiB)
+	}
+}
+
+// TestHeartbeater_NoRuntimeAdvertisesNoCapacity is the same rule on the beat,
+// which is how a runtime lost after boot reaches the control plane.
+func TestHeartbeater_NoRuntimeAdvertisesNoCapacity(t *testing.T) {
+	cp := &fakeCP{}
+	h := newHeartbeater(cp, testIdentity(), time.Second, func() bool { return false })
+
+	if err := h.beat(); err != nil {
+		t.Fatalf("beat: %v", err)
+	}
+	if got := cp.registered().Capacity; got.CPU != 0 || got.MemoryMiB != 0 {
+		t.Fatalf("beat advertised capacity with no runtime: cpu=%d memory=%d", got.CPU, got.MemoryMiB)
+	}
+	// The instance still identifies itself, so it stays visible as drained rather
+	// than vanishing from the cluster.
+	if cp.registered().InstanceID != testIdentity().InstanceID {
+		t.Error("beat dropped the instance identity")
+	}
+}

cmd/ecs-agent/runtimegate.go

@@ -0,0 +1,116 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+)
+
+// runtimeProbeTimeout bounds the liveness probe against a runtime the gate is
+// already holding, so a wedged socket cannot stall the watch loop.
+const runtimeProbeTimeout = 5 * time.Second
+
+// defaultRuntimeRetry is how often the gate dials a runtime it does not have,
+// and probes one it does. The node starts containerd and the agent back to
+// back, so the first few dials are expected to fail.
+const defaultRuntimeRetry = 10 * time.Second
+
+// runtimeGate holds the container runtime the agent has resolved, or nothing
+// when the socket has not answered. The runtime is resolved by a loop rather
+// than once at boot: the node's service manager reports the container daemon
+// started when the process spawned, not when its socket began accepting, so a
+// dial that fails in the first second of boot is a transient condition.
+//
+// Every read of the runtime goes through the gate, so a runtime that arrives
+// after boot is picked up on the next task without restarting the agent.
+type runtimeGate struct {
+	mu sync.RWMutex
+	rt ctrruntime.Runtime
+}
+
+// newRuntimeGate returns a gate holding rt, which may be nil.
+func newRuntimeGate(rt ctrruntime.Runtime) *runtimeGate {
+	return &runtimeGate{rt: rt}
+}
+
+// get returns the current runtime, or nil when none is resolved.
+func (g *runtimeGate) get() ctrruntime.Runtime {
+	g.mu.RLock()
+	defer g.mu.RUnlock()
+	return g.rt
+}
+
+// available reports whether a runtime is resolved. It is what registration
+// consults to decide whether the host has any capacity to advertise.
+func (g *runtimeGate) available() bool { return g.get() != nil }
+
+// set replaces the held runtime, closing the one it displaces.
+func (g *runtimeGate) set(rt ctrruntime.Runtime) {
+	g.mu.Lock()
+	old := g.rt
+	g.rt = rt
+	g.mu.Unlock()
+	if old != nil && old != rt {
+		_ = old.Close()
+	}
+}
+
+// clear drops and closes the held runtime, returning whether it held one.
+func (g *runtimeGate) clear() bool {
+	g.mu.Lock()
+	old := g.rt
+	g.rt = nil
+	g.mu.Unlock()
+	if old == nil {
+		return false
+	}
+	_ = old.Close()
+	return true
+}
+
+// watch keeps the gate current until ctx is cancelled: it dials while the gate
+// is empty, and probes while it is full, clearing the gate when the probe fails
+// so the next tick re-dials. A containerd stopped and restarted underneath a
+// running agent is therefore picked up the same way a slow boot is.
+func (g *runtimeGate) watch(ctx context.Context, dial func() (ctrruntime.Runtime, error), interval time.Duration) {
+	if interval <= 0 {
+		interval = defaultRuntimeRetry
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			g.step(ctx, dial)
+		}
+	}
+}
+
+// step performs one pass of the watch loop: dial when empty, probe when full.
+func (g *runtimeGate) step(ctx context.Context, dial func() (ctrruntime.Runtime, error)) {
+	if rt := g.get(); rt != nil {
+		probeCtx, cancel := context.WithTimeout(ctx, runtimeProbeTimeout)
+		// A container list is what reconcile already asks the runtime at boot and
+		// the cheapest thing it answers, so it doubles as the liveness probe.
+		_, err := rt.List(probeCtx)
+		cancel()
+		if err != nil && ctx.Err() == nil {
+			slog.Warn("ecs-agent: container runtime stopped answering, will re-dial", "err", err)
+			g.clear()
+		}
+		return
+	}
+
+	rt, err := dial()
+	if err != nil {
+		slog.Debug("ecs-agent: container runtime still unavailable", "err", err)
+		return
+	}
+	slog.Info("ecs-agent: container runtime available")
+	g.set(rt)
+}

cmd/ecs-agent/runtimegate_test.go

@@ -0,0 +1,122 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime"
+)
+
+// TestRuntimeGate_DialsUntilTheSocketAnswers is the boot case the agent used to
+// lose: the node starts containerd and the agent together, so the first dials
+// fail. Resolving once left the agent runtime-less for the life of the process.
+func TestRuntimeGate_DialsUntilTheSocketAnswers(t *testing.T) {
+	g := newRuntimeGate(nil)
+	if g.available() {
+		t.Fatal("empty gate reported a runtime")
+	}
+
+	rt := &ctrruntime.FakePuller{}
+	dials := 0
+	dial := func() (ctrruntime.Runtime, error) {
+		dials++
+		if dials < 3 {
+			return nil, errors.New("dial containerd: no such file or directory")
+		}
+		return rt, nil
+	}
+
+	for range 2 {
+		g.step(t.Context(), dial)
+		if g.available() {
+			t.Fatalf("gate resolved a runtime after %d failed dials", dials)
+		}
+	}
+
+	g.step(t.Context(), dial)
+	if g.get() != rt {
+		t.Fatalf("gate did not pick up the runtime once the socket answered (dials=%d)", dials)
+	}
+}
+
+// TestRuntimeGate_DropsARuntimeThatStopsAnswering covers a containerd restarted
+// underneath a live agent: the held runtime stops answering, so the gate clears
+// it and the next pass re-dials rather than handing out a dead client forever.
+func TestRuntimeGate_DropsARuntimeThatStopsAnswering(t *testing.T) {
+	dead := &ctrruntime.FakePuller{ListErr: errors.New("connection refused")}
+	fresh := &ctrruntime.FakePuller{}
+	g := newRuntimeGate(dead)
+
+	dial := func() (ctrruntime.Runtime, error) { return fresh, nil }
+
+	g.step(t.Context(), dial)
+	if g.available() {
+		t.Fatal("gate kept a runtime whose probe failed")
+	}
+	if !dead.Closed {
+		t.Error("displaced runtime was not closed")
+	}
+
+	g.step(t.Context(), dial)
+	if g.get() != fresh {
+		t.Fatal("gate did not re-dial after clearing")
+	}
+}
+
+// TestRuntimeGate_KeepsAHealthyRuntime guards the other direction: a probe that
+// succeeds must not churn the runtime, or every pass would drop a working client.
+func TestRuntimeGate_KeepsAHealthyRuntime(t *testing.T) {
+	rt := &ctrruntime.FakePuller{}
+	g := newRuntimeGate(rt)
+
+	dial := func() (ctrruntime.Runtime, error) {
+		t.Error("dialled while holding a healthy runtime")
+		return nil, errors.New("unexpected dial")
+	}
+	g.step(t.Context(), dial)
+
+	if g.get() != rt {
+		t.Fatal("gate dropped a runtime whose probe succeeded")
+	}
+	if rt.Closed {
+		t.Error("healthy runtime was closed")
+	}
+}
+
+// TestRuntimeGate_ClearIsIdempotent keeps Stop safe to call on an agent that
+// never resolved a runtime.
+func TestRuntimeGate_ClearIsIdempotent(t *testing.T) {
+	g := newRuntimeGate(nil)
+	if g.clear() {
+		t.Error("clear reported dropping a runtime the gate never held")
+	}
+
+	rt := &ctrruntime.FakePuller{}
+	g.set(rt)
+	if !g.clear() {
+		t.Error("clear did not report dropping the held runtime")
+	}
+	if !rt.Closed {
+		t.Error("cleared runtime was not closed")
+	}
+	if g.clear() {
+		t.Error("second clear reported dropping a runtime")
+	}
+}
+
+// TestRuntimeGate_WatchStopsWithContext keeps the watch loop from outliving the
+// agent it belongs to.
+func TestRuntimeGate_WatchStopsWithContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	g := newRuntimeGate(nil)
+
+	done := make(chan struct{})
+	go func() {
+		g.watch(ctx, func() (ctrruntime.Runtime, error) { return nil, errors.New("no socket") }, 0)
+		close(done)
+	}()
+
+	cancel()
+	<-done
+}

cmd/ecs-agent/taskrun.go

@@ -128,11 +128,12 @@ func (l *stopLedger) seen(taskID string) bool {
 // scheduler releases its capacity. Reports whether it got as far as saying
 // STOPPED: a caller may only ack the directive when it did.
 func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
-	if a.runner == nil {
+	runner := a.rt.get()
+	if runner == nil {
 		slog.Warn("ecs-agent: stop deferred, no container runtime", "task", sd.TaskID)
 		return false
 	}
-	containers, err := a.runner.List(ctx)
+	containers, err := runner.List(ctx)
 	if err != nil {
 		slog.Warn("ecs-agent: stop list failed", "task", sd.TaskID, "err", err)
 		return false
@@ -142,7 +143,7 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
 		if c.Labels[labelTaskID] != sd.TaskID {
 			continue
 		}
-		if rerr := a.runner.Stop(ctx, c.ID, containerStopTimeout(c.Labels)); rerr != nil {
+		if rerr := runner.Stop(ctx, c.ID, containerStopTimeout(c.Labels)); rerr != nil {
 			slog.Warn("ecs-agent: stop failed", "task", sd.TaskID, "container", c.ID, "err", rerr)
 		}
 		statuses = append(statuses, bus.ContainerStatus{
@@ -162,7 +163,8 @@ func (a *Agent) stopTask(ctx context.Context, sd bus.StopDirective) bool {
 // state on the bus. A missing runtime or any per-container failure reports the
 // task STOPPED with a reason; success reports RUNNING and waits for exit.
 func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
-	if a.puller == nil || a.runner == nil {
+	rt := a.rt.get()
+	if rt == nil {
 		a.reportTaskState(as, bus.TaskStatusStopped, "containerd unavailable on agent", nil)
 		return
 	}
@@ -184,7 +186,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 		// container's time-to-RUNNING from the same starting point.
 		runCtx, cancel := containerStartCtx(ctx, c.StartTimeout)
 
-		if _, err := a.puller.Pull(runCtx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil {
+		if _, err := rt.Pull(runCtx, ctrruntime.PullSpec{Ref: c.Image}, resolver); err != nil {
 			cancel()
 			slog.Error("ecs-agent: pull failed", "task", as.TaskID, "image", c.Image, "err", err)
 			a.teardownTaskNetns(as)
@@ -215,7 +217,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 			CapAdd:                 c.CapAdd,
 			CapDrop:                c.CapDrop,
 		}
-		id, err := a.runner.Run(runCtx, cid, spec)
+		id, err := rt.Run(runCtx, cid, spec)
 		cancel()
 		if err != nil {
 			slog.Error("ecs-agent: run failed", "task", as.TaskID, "container", c.Name, "err", err)
@@ -236,7 +238,12 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) {
 // waitContainer blocks until a container exits, then reports the task STOPPED
 // with the exit code. v1 stops the whole task when its first container exits.
 func (a *Agent) waitContainer(ctx context.Context, as *bus.Assign, name, containerID string) {
-	status, err := a.runner.Wait(ctx, containerID)
+	runner := a.rt.get()
+	if runner == nil {
+		slog.Warn("ecs-agent: cannot wait on container, no container runtime", "task", as.TaskID, "container", name)
+		return
+	}
+	status, err := runner.Wait(ctx, containerID)
 	if err != nil {
 		slog.Warn("ecs-agent: wait container failed", "task", as.TaskID, "container", name, "err", err)
 		return

cmd/ecs-agent/taskrun_test.go

@@ -25,13 +25,13 @@ func testAssign() *bus.Assign {
 }
 
 func newRunAgent(cp controlPlane, rt *ctrruntime.FakePuller) *Agent {
-	return newAgent(config{}, testIdentity(), cp, rt, rt, nil)
+	return newAgent(config{}, testIdentity(), cp, rt, nil)
 }
 
 // runTask with no runtime reports the task STOPPED instead of crashing.
 func TestRunTask_NoRuntimeReportsStopped(t *testing.T) {
 	cp := &fakeCP{}
-	a := newAgent(config{}, testIdentity(), cp, nil, nil, nil)
+	a := newAgent(config{}, testIdentity(), cp, nil, nil)
 	a.runTask(context.Background(), testAssign())
 
 	st := cp.taskStates()
@@ -252,7 +252,7 @@ func TestRunTask_GPUCarriedToRunSpec(t *testing.T) {
 func TestRunTask_GPUPinsAndReportsUUIDs(t *testing.T) {
 	cp := &fakeCP{}
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
-	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
 	as := testAssign()
 	as.Containers = append(as.Containers, bus.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 1,
@@ -296,7 +296,7 @@ func TestRunTask_GPUPinsAndReportsUUIDs(t *testing.T) {
 func TestRunTask_GPUIDsReachRunSpec(t *testing.T) {
 	cp := &fakeCP{}
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
-	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa", "GPU-bbb"), cp, rt, nil)
 	as := testAssign()
 	as.Containers = append(as.Containers, bus.AssignContainer{
 		Name: "trainer", Image: "registry/trainer:1", GPU: 1,
@@ -361,7 +361,7 @@ func TestStopTask_ReleasesGPUsBackToLedger(t *testing.T) {
 			"mulga.ecs.taskID": "t-001", "mulga.ecs.containerName": "trainer",
 		}},
 	}}
-	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa"), cp, rt, rt, nil)
+	a := newAgent(config{}, testIdentityWithGPUs("GPU-aaa"), cp, rt, nil)
 	if _, err := a.gpu.Pin(gpuKey("t-001", "trainer"), 1); err != nil {
 		t.Fatalf("seed pin: %v", err)
 	}
@@ -439,7 +439,7 @@ func TestPollAssignments_DispatchesOnceAndAcks(t *testing.T) {
 		nil,   // poll 3: gateway dropped it after the ack
 	}}
 	rt := &ctrruntime.FakePuller{WaitErr: errors.New("blocked")}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go a.pollAssignments(ctx, map[string]bool{})
@@ -538,7 +538,7 @@ func TestPollAssignments_StopReapsAndSuppressesAssign(t *testing.T) {
 	rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
 		{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
 	}}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	go a.pollAssignments(ctx, map[string]bool{})
@@ -593,7 +593,7 @@ func countStopAcks(acks [][]string, taskID string) int {
 func TestPollAssignments_UnperformableStopIsNotAckedAndRetried(t *testing.T) {
 	cp := &fakeCP{stopPending: []bus.StopDirective{{TaskID: "t-001", Reason: "bye"}}}
 	rt := &ctrruntime.FakePuller{ListErr: errors.New("containerd is down")}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -628,7 +628,7 @@ func TestPollAssignments_RecoveredStopIsAckedExactlyOnce(t *testing.T) {
 			{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
 		},
 	}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -667,7 +667,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
 	rt := &ctrruntime.FakePuller{Containers: []ctrruntime.Container{
 		{ID: "t-001-web", Running: true, Labels: map[string]string{"mulga.ecs.taskID": "t-001"}},
 	}}
-	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, rt, nil)
+	a := newAgent(config{PollInterval: 5 * time.Millisecond}, testIdentity(), cp, rt, nil)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -696,7 +696,7 @@ func TestPollAssignments_StoppedTaskIsNotRerunByRedeliveredAssign(t *testing.T)
 // stopTask with no runtime is a no-op: nothing to reap, no state reported.
 func TestStopTask_NilRunnerNoOp(t *testing.T) {
 	cp := &fakeCP{}
-	a := newAgent(config{}, testIdentity(), cp, nil, nil, nil)
+	a := newAgent(config{}, testIdentity(), cp, nil, nil)
 	a.stopTask(context.Background(), bus.StopDirective{TaskID: "t-001", Reason: "x"})
 	if len(cp.taskStates()) != 0 {
 		t.Fatalf("nil runner should report no state, got %+v", cp.taskStates())
@@ -807,7 +807,7 @@ func (sentinelResolver) Authorize(context.Context, string) (string, string, stri
 // exec-role resolver is built.
 func TestPullResolver(t *testing.T) {
 	fake := sentinelResolver{}
-	a := newAgent(config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, nil, fake)
+	a := newAgent(config{Region: "us-east-1"}, testIdentity(), &fakeCP{}, nil, fake)
 
 	if got := a.pullResolver(&bus.Assign{TaskID: "t1"}); got != fake {
 		t.Fatalf("no execution role: want instance resolver fallback")

spinifex/handlers/ecs/deployments.go

@@ -195,6 +195,26 @@ func updateRolloutState(svc *ServiceRecord, primary *Deployment, desired int) {
 	primary.RolloutState = RolloutStateInProgress
 }
 
+// launchBackoff returns how long to hold off the next launch for a deployment
+// with failures behind it, doubling from a base and capped. It reads the same
+// counter and the same threshold as the circuit breaker, but applies whether or
+// not the breaker is enabled: AWS leaves it off by default, so an unbraked
+// service relaunches a task that cannot start on every pass, forever.
+//
+// Below the threshold there is no hold, so an enabled breaker still sees its
+// failures at full speed and trips exactly when it did before.
+func launchBackoff(failures int) time.Duration {
+	if failures < circuitBreakerFailureThreshold {
+		return 0
+	}
+	// Bounded before the shift so a long-failing deployment cannot overflow it.
+	d := launchBackoffBase << min(failures-circuitBreakerFailureThreshold, launchBackoffMaxShift)
+	if d > launchBackoffCap {
+		return launchBackoffCap
+	}
+	return d
+}
+
 // tripCircuitBreaker fails a PRIMARY deployment whose task launches have failed
 // past the threshold when the breaker is enabled, optionally rolling back to the
 // last-good task definition. Returns true when it acted so the caller can persist

spinifex/handlers/ecs/deployments_test.go

@@ -3,6 +3,7 @@ package handlers_ecs
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ecs"
@@ -336,3 +337,73 @@ func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) {
 	assert.Equal(t, "legacy-dep", primary.ID)
 	assert.Equal(t, defaultMinimumHealthyPercent, reloaded.MinimumHealthyPercent)
 }
+
+// TestLaunchBackoff_WidensWithFailuresAndCaps is the arithmetic on its own: no
+// hold while the circuit breaker is still counting, then doubling, then a cap so
+// a long-dead deployment does not wait forever to be retried.
+func TestLaunchBackoff_WidensWithFailuresAndCaps(t *testing.T) {
+	for failures := range circuitBreakerFailureThreshold {
+		assert.Zero(t, launchBackoff(failures), "held off at %d failures, below the breaker threshold", failures)
+	}
+	assert.Equal(t, launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold))
+	assert.Equal(t, 2*launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold+1))
+	assert.Equal(t, 4*launchBackoffBase, launchBackoff(circuitBreakerFailureThreshold+2))
+	assert.Equal(t, launchBackoffCap, launchBackoff(circuitBreakerFailureThreshold+20))
+	assert.Equal(t, launchBackoffCap, launchBackoff(1_000_000))
+}
+
+// A deployment whose tasks keep dying on start stops relaunching on every pass.
+// AWS leaves the deployment circuit breaker off by default, so without this a
+// service whose node cannot run containers churns tasks forever — which is how
+// one env accumulated hundreds of STOPPED tasks at one every thirty seconds.
+func TestDeployment_RepeatedStartFailuresBackOffWithoutTheCircuitBreaker(t *testing.T) {
+	svc, _, kv := serviceTestRig(t)
+	_, err := svc.CreateService(context.Background(), &ecs.CreateServiceInput{
+		Cluster: aws.String("web"), ServiceName: aws.String("web"),
+		TaskDefinition: aws.String("app"), DesiredCount: aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+
+	// Fail every task the reconciler launches. The breaker is off, so nothing
+	// else brakes this.
+	for range circuitBreakerFailureThreshold {
+		failPending(t, svc, kv, "web", "web")
+		require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web")))
+	}
+
+	rec := reloadService(t, kv, "web", "web")
+	primary := rec.primaryDeployment()
+	require.NotNil(t, primary)
+	require.GreaterOrEqual(t, primary.FailedTasks, circuitBreakerFailureThreshold)
+	require.False(t, primary.NextLaunchAt.IsZero(), "no backoff recorded after repeated start failures")
+	assert.True(t, primary.NextLaunchAt.After(time.Now().UTC()), "backoff deadline is already in the past")
+
+	// The service is below its desired count and would otherwise launch, but the
+	// deadline holds the pass off.
+	failPending(t, svc, kv, "web", "web")
+	require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web")))
+	assert.Zero(t, liveServiceTasks(t, svc, kv, "web", "web"), "reconcile launched a task while the deployment was backed off")
+
+	// Once it passes, launching resumes: the backoff is a brake, not a stop.
+	rec = reloadService(t, kv, "web", "web")
+	rec.primaryDeployment().NextLaunchAt = time.Now().UTC().Add(-time.Second)
+	require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), rec))
+	require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web")))
+
+	assert.Positive(t, liveServiceTasks(t, svc, kv, "web", "web"), "backoff never released the deployment")
+}
+
+// liveServiceTasks counts a service's tasks that have not stopped, which is what
+// the reconciler is trying to hold at the desired count.
+func liveServiceTasks(t *testing.T, svc *Service, kv jetstream.KeyValue, cluster, name string) int {
+	t.Helper()
+	tasks, err := svc.listServiceTasks(t.Context(), kv, cluster, name)
+	require.NoError(t, err)
+	n := 0
+	for i := range tasks {
+		if tasks[i].LastStatus != TaskStatusStopped {
+			n++
+		}
+	}
+	return n
+}

spinifex/handlers/ecs/instance_recovery_test.go

@@ -136,3 +136,89 @@ func TestRegister_DoesNotUndoOperatorDrain(t *testing.T) {
 	after := instanceStatus(t, svc, "web", "i-1")
 	assert.Equal(t, InstanceStatusDraining, after.Status, "operator drain must survive agent re-register")
 }
+
+// An agent that cannot resolve a container runtime registers no capacity. That
+// is the same condition as an agent that stopped answering, so the instance is
+// drained involuntarily and the scheduler stops placing tasks it could only
+// fail to start.
+func TestRegister_NoCapacityDrainsTheInstance(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "app", 128, 256)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	// The runtime goes away: the next registration reports nothing to run with.
+	registerInstance(t, svc, "web", "i-1", 0, 0)
+
+	drained := instanceStatus(t, svc, "web", "i-1")
+	assert.Equal(t, InstanceStatusDraining, drained.Status)
+	assert.True(t, drained.Reaped, "a runtime-less instance is an involuntary drain, not an operator one")
+	assert.Zero(t, drained.TotalCPU)
+	assert.Zero(t, drained.TotalMemoryMiB)
+
+	out, err := svc.RunTask(context.Background(), &ecs.RunTaskInput{
+		Cluster:        aws.String("web"),
+		TaskDefinition: aws.String("app"),
+		Count:          aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Empty(t, out.Tasks, "a task was placed on an instance with no runtime")
+	assert.NotEmpty(t, out.Failures)
+
+	// The instance is visibly drained through the API rather than only in the
+	// node's log, which is what an operator has to go on.
+	di, err := svc.DescribeContainerInstances(context.Background(), &ecs.DescribeContainerInstancesInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+	}, testAccountID)
+	require.NoError(t, err)
+	require.Len(t, di.ContainerInstances, 1)
+	assert.Equal(t, InstanceStatusDraining, aws.StringValue(di.ContainerInstances[0].Status))
+	assert.False(t, aws.BoolValue(di.ContainerInstances[0].AgentConnected))
+}
+
+// The runtime coming back is a registration carrying real capacity again, which
+// clears the drain without the agent or the instance being restarted.
+func TestRegister_CapacityReturningClearsTheDrain(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerTaskDef(t, svc, "app", 128, 256)
+	registerInstance(t, svc, "web", "i-1", 0, 0)
+	require.Equal(t, InstanceStatusDraining, instanceStatus(t, svc, "web", "i-1").Status)
+
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	back := instanceStatus(t, svc, "web", "i-1")
+	assert.Equal(t, InstanceStatusActive, back.Status)
+	assert.False(t, back.Reaped)
+	assert.Equal(t, 1024, back.TotalCPU)
+
+	out, err := svc.RunTask(context.Background(), &ecs.RunTaskInput{
+		Cluster:        aws.String("web"),
+		TaskDefinition: aws.String("app"),
+		Count:          aws.Int64(1),
+	}, testAccountID)
+	require.NoError(t, err)
+	assert.Len(t, out.Tasks, 1)
+}
+
+// An operator drain is deliberate, so a capacity-carrying registration must not
+// undo it. Only the involuntary drain this path sets is cleared by a register.
+func TestRegister_DoesNotClearAnOperatorDrain(t *testing.T) {
+	svc, _ := newTestService(t)
+	_, err := svc.CreateCluster(context.Background(), &ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID)
+	require.NoError(t, err)
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	_, err = svc.UpdateContainerInstancesState(context.Background(), &ecs.UpdateContainerInstancesStateInput{
+		Cluster: aws.String("web"), ContainerInstances: []*string{aws.String("i-1")},
+		Status: aws.String(InstanceStatusDraining),
+	}, testAccountID)
+	require.NoError(t, err)
+	require.False(t, instanceStatus(t, svc, "web", "i-1").Reaped)
+
+	registerInstance(t, svc, "web", "i-1", 1024, 2048)
+
+	assert.Equal(t, InstanceStatusDraining, instanceStatus(t, svc, "web", "i-1").Status)
+}

spinifex/handlers/ecs/instances.go

@@ -35,6 +35,30 @@ func (s *Service) listInstanceRecords(ctx context.Context, kv jetstream.KeyValue
 	return out, nil
 }
 
+// applyRuntimeCapacity records the capacity a registration reported and derives
+// the instance's availability from it. A host reporting no CPU and no memory has
+// no container runtime to place on, which is the same condition as an agent that
+// stopped answering: it drains involuntarily so the scheduler skips it, and the
+// next registration carrying real capacity restores it. An operator drain
+// (Reaped false) is never cleared here.
+func applyRuntimeCapacity(r *InstanceRecord, cpu, memoryMiB int) {
+	r.TotalCPU = cpu
+	r.TotalMemoryMiB = memoryMiB
+	if cpu == 0 && memoryMiB == 0 {
+		if r.Status != InstanceStatusDraining {
+			slog.Warn("ECS: container instance reported no capacity, draining until a runtime is available",
+				"instance_id", r.InstanceID, "cluster", r.Cluster)
+		}
+		r.Status = InstanceStatusDraining
+		r.Reaped = true
+		return
+	}
+	if r.Status == InstanceStatusDraining && r.Reaped {
+		r.Status = InstanceStatusActive
+		r.Reaped = false
+	}
+}
+
 // RegisterContainerInstance is the AWS-API registration path. In 4e the agent
 // normally registers over the Layer-2 bus; this keeps API parity by writing the
 // same record shape from an explicit call.
@@ -49,12 +73,17 @@ func (s *Service) RegisterContainerInstance(ctx context.Context, input *ecs.Regi
 		return nil, err
 	}
 	rec, err := s.upsertInstance(ctx, kv, accountID, cluster, instanceID, func(r *InstanceRecord) {
+		// CPU and memory are read together, because it is the pair being zero that
+		// says the host has no runtime. A registration that omits either leaves the
+		// previously reported capacity alone.
+		cpu, memoryMiB := r.TotalCPU, r.TotalMemoryMiB
+		var sawCPU, sawMemory bool
 		for _, res := range input.TotalResources {
 			switch aws.StringValue(res.Name) {
 			case "CPU":
-				r.TotalCPU = int(aws.Int64Value(res.IntegerValue))
+				cpu, sawCPU = int(aws.Int64Value(res.IntegerValue)), true
 			case "MEMORY":
-				r.TotalMemoryMiB = int(aws.Int64Value(res.IntegerValue))
+				memoryMiB, sawMemory = int(aws.Int64Value(res.IntegerValue)), true
 			case "GPU":
 				// AWS reports GPU as a STRINGSET of device UUIDs; the count is the
 				// capacity, the UUIDs are the placeholder inventory (Epic C3 populates
@@ -74,10 +103,12 @@ func (s *Service) RegisterContainerInstance(ctx context.Context, input *ecs.Regi
 		if len(input.Tags) > 0 {
 			r.Tags = tagsToMap(input.Tags)
 		}
-		// The agent heartbeats by re-registering. A re-register from a reaped
-		// (involuntarily drained) instance proves the agent is back, so restore
-		// it to ACTIVE. An operator drain (Reaped=false) is left untouched.
-		if r.Status == InstanceStatusDraining && r.Reaped {
+		// The agent heartbeats by re-registering, so the capacity it reports is
+		// also the liveness signal: real capacity restores a reaped instance, and
+		// none drains it.
+		if sawCPU && sawMemory {
+			applyRuntimeCapacity(r, cpu, memoryMiB)
+		} else if r.Status == InstanceStatusDraining && r.Reaped {
 			r.Status = InstanceStatusActive
 			r.Reaped = false
 		}
@@ -300,11 +331,9 @@ func (s *Service) recordRegister(ctx context.Context, msg *bus.RegisterInstance)
 		r.AZ = msg.AZ
 		r.Hostname = msg.Hostname
 		r.AgentVersion = msg.AgentVersion
-		r.TotalCPU = msg.Capacity.CPU
-		r.TotalMemoryMiB = msg.Capacity.MemoryMiB
 		r.TotalGPU = msg.Capacity.GPU
 		r.GPUIDs = msg.Capacity.GPUIDs
-		r.Status = InstanceStatusActive
+		applyRuntimeCapacity(r, msg.Capacity.CPU, msg.Capacity.MemoryMiB)
 	})
 	return err
 }

spinifex/handlers/ecs/records.go

@@ -53,6 +53,13 @@ const (
 	// primary deployment's task launches stop before ever reaching RUNNING.
 	circuitBreakerFailureThreshold = 3
 
+	// launchBackoff bounds: past circuitBreakerFailureThreshold the wait doubles
+	// from the base on each further failed task, up to the cap. The shift bound
+	// keeps the doubling inside a duration for a long-failing deployment.
+	launchBackoffBase     = 15 * time.Second
+	launchBackoffCap      = 5 * time.Minute
+	launchBackoffMaxShift = 16
+
 	CapacityProviderStatusActive   = "ACTIVE"
 	CapacityProviderStatusInactive = "INACTIVE"
 
@@ -66,19 +73,23 @@ const (
 // one PRIMARY deployment plus zero or more ACTIVE (superseded, draining) ones
 // while a rolling update is in flight; steady state is a single PRIMARY.
 type Deployment struct {
-	ID              string    `json:"id"`
-	Status          string    `json:"status"`
-	TaskDefARN      string    `json:"taskDefArn"`
-	TaskDefFamily   string    `json:"taskDefFamily"`
-	TaskDefRevision int       `json:"taskDefRevision"`
-	DesiredCount    int       `json:"desiredCount"`
-	RunningCount    int       `json:"runningCount"`
-	PendingCount    int       `json:"pendingCount"`
-	FailedTasks     int       `json:"failedTasks"`
-	RolloutState    string    `json:"rolloutState"`
-	RolloutReason   string    `json:"rolloutStateReason,omitempty"`
-	CreatedAt       time.Time `json:"createdAt"`
-	UpdatedAt       time.Time `json:"updatedAt"`
+	ID              string `json:"id"`
+	Status          string `json:"status"`
+	TaskDefARN      string `json:"taskDefArn"`
+	TaskDefFamily   string `json:"taskDefFamily"`
+	TaskDefRevision int    `json:"taskDefRevision"`
+	DesiredCount    int    `json:"desiredCount"`
+	RunningCount    int    `json:"runningCount"`
+	PendingCount    int    `json:"pendingCount"`
+	FailedTasks     int    `json:"failedTasks"`
+	RolloutState    string `json:"rolloutState"`
+	RolloutReason   string `json:"rolloutStateReason,omitempty"`
+	// NextLaunchAt holds off the next launch attempt after tasks have failed to
+	// start, so a deployment that cannot run anywhere backs off instead of
+	// relaunching on every reconcile pass. Zero means launch now.
+	NextLaunchAt time.Time `json:"nextLaunchAt,omitzero"`
+	CreatedAt    time.Time `json:"createdAt"`
+	UpdatedAt    time.Time `json:"updatedAt"`
 }
 
 // ServiceEvent is one lifecycle event on a service, mirroring ecs.ServiceEvent.

spinifex/handlers/ecs/services.go

@@ -379,10 +379,16 @@ func (s *Service) reconcileService(ctx context.Context, kv jetstream.KeyValue, a
 		minCount := ceilPercent(desired, svc.MinimumHealthyPercent)
 
 		primaryActive := primary.RunningCount + primary.PendingCount
-		if primaryActive < desired && primary.RolloutState != RolloutStateFailed {
+		now := time.Now().UTC()
+		if primaryActive < desired && primary.RolloutState != RolloutStateFailed && !now.Before(primary.NextLaunchAt) {
 			n := min(desired-primaryActive, max(maxCount-(running+pending), 0))
 			if n > 0 {
 				s.launchDeploymentTasks(ctx, accountID, svc, primary, n)
+				// Set from the failures already behind this deployment, so the first
+				// launch is immediate and each repeat start failure widens the gap.
+				if d := launchBackoff(primary.FailedTasks); d > 0 {
+					primary.NextLaunchAt = now.Add(d)
+				}
 			} else {
 				// Wants more tasks than desired but the maximumPercent ceiling leaves
 				// no room this pass (AWS's "unable to place a task"). Unlike the