fix(scheduler): bound runner teardown and retry failed env creates

cmd/ephemerd/main.go

@@ -371,6 +371,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 
 			OrphanContainerReapEnabled: cfg.Runtime.ResolvedOrphanContainerReap(),
 			OrphanContainerGrace:       cfg.Runtime.ResolvedOrphanContainerGrace(),
+			DestroyTimeout:             cfg.Runtime.ResolvedDestroyTimeout(),
 			Log:                        log,
 		})
 		if err != nil {
@@ -653,6 +654,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP
 
 		OrphanContainerReapEnabled: cfg.Runtime.ResolvedOrphanContainerReap(),
 		OrphanContainerGrace:       cfg.Runtime.ResolvedOrphanContainerGrace(),
+		DestroyTimeout:             cfg.Runtime.ResolvedDestroyTimeout(),
 		Log:                        log,
 	})
 	if err != nil {

pkg/config/config.go

@@ -387,6 +387,18 @@ type RuntimeConfig struct {
 	// duration strings ("10m", "1h"). Default 10m. See
 	// ResolvedOrphanContainerGrace.
 	OrphanContainerGrace time.Duration `toml:"orphan_container_grace"`
+
+	// DestroyTimeout bounds a single runner-environment teardown end to
+	// end. A healthy teardown takes ~15s; the bound exists for the
+	// unhealthy case — a dead containerd shim never delivers the killed
+	// task's exit event, and an unbounded Destroy then blocks forever,
+	// pinning the job's concurrency slot and wedging drains (observed in
+	// production: one dead shim held a slot for over an hour and stalled
+	// two consecutive upgrade drains). On expiry Destroy returns and the
+	// remaining teardown is abandoned to the orphan reapers (startup
+	// CleanOrphans + periodic ReapDeadContainers). Accepts Go duration
+	// strings ("5m", "90s"). Default 5m. See ResolvedDestroyTimeout.
+	DestroyTimeout time.Duration `toml:"destroy_timeout"`
 }
 
 // ResolvedOrphanContainerReap returns whether the periodic dead-container
@@ -413,6 +425,25 @@ func (r RuntimeConfig) ResolvedOrphanContainerGrace() time.Duration {
 	return r.OrphanContainerGrace
 }
 
+// ResolvedDestroyTimeout returns the runner teardown bound with the
+// default (5m) applied when unset (or non-positive). The default is far
+// above what a healthy teardown needs (~15s), so it only ever fires when
+// a teardown step is genuinely wedged — there is deliberately no way to
+// configure an unbounded destroy.
+func (r RuntimeConfig) ResolvedDestroyTimeout() time.Duration {
+	// The floor guards a TOML footgun, not an operator preference: BurntSushi
+	// decodes a bare integer `destroy_timeout = 300` as 300 NANOSECONDS with
+	// no error (a duration string like "300s" is what the operator meant).
+	// A sub-second bound would make every teardown "time out" instantly and
+	// abandon its cleanup, piling up containers, snapshots, and endpoints —
+	// the opposite of what any configured value could intend. Anything below
+	// one second is therefore treated as a misparse, not a choice.
+	if r.DestroyTimeout < time.Second {
+		return 5 * time.Minute
+	}
+	return r.DestroyTimeout
+}
+
 // ResolvedAllowNewPrivileges returns whether the runner container may
 // gain privileges through execve, applying the default when the operator
 // hasn't set the key explicitly.

pkg/config/config_test.go

@@ -2471,3 +2471,32 @@ func TestResolvedReconcileInterval(t *testing.T) {
 		t.Errorf("nil ResolvedReconcileInterval() = %v, want 30m", got)
 	}
 }
+
+// TestResolvedDestroyTimeout pins the fail-safe contract: the teardown
+// bound must be finite no matter what the operator writes — there is no
+// configuration that yields an unbounded Destroy, because an unbounded
+// Destroy is exactly what let a dead shim pin a concurrency slot for an
+// hour in production.
+func TestResolvedDestroyTimeout(t *testing.T) {
+	cases := []struct {
+		in   time.Duration
+		want time.Duration
+	}{
+		{0, 5 * time.Minute},                 // unset -> default
+		{-time.Minute, 5 * time.Minute},      // nonsense -> default, never unbounded
+		{90 * time.Second, 90 * time.Second}, // custom
+		// The TOML footgun: `destroy_timeout = 300` (bare integer, meaning
+		// seconds to the operator) decodes as 300 NANOSECONDS without error.
+		// Sub-second values are a misparse, never a choice — a near-zero
+		// bound would abandon every teardown instantly.
+		{300 * time.Nanosecond, 5 * time.Minute},
+		{999 * time.Millisecond, 5 * time.Minute},
+		{time.Second, time.Second}, // the floor itself is honored
+	}
+	for _, c := range cases {
+		r := RuntimeConfig{DestroyTimeout: c.in}
+		if got := r.ResolvedDestroyTimeout(); got != c.want {
+			t.Errorf("ResolvedDestroyTimeout(%v) = %v, want %v", c.in, got, c.want)
+		}
+	}
+}

pkg/runnerbusy/container_windows.go

@@ -43,8 +43,37 @@ var openCompute = func(id string) (hcsProcessLister, error) {
 // Same fail-safe rule as the Linux probe: anything we cannot read is
 // Unknown, and an empty listing (which HCS also returns for a compute
 // system that has gone away underneath us) is Unknown rather than Idle.
-func ContainerBusy(_ context.Context, t ContainerTask, log *slog.Logger) (State, error) {
+//
+// The hcsshim calls take no context, so the probe runs them on a
+// goroutine and selects on ctx: without this, the caller's
+// busyProbeTimeout was simply not enforced on Windows, and a dead
+// shim/GCS wedged the probe (and the sweep serialized behind it)
+// indefinitely. On ctx expiry the goroutine is abandoned — it parks
+// inside HCS until that call eventually errors — and the probe reports
+// Unknown, which is the fail-safe veto.
+func ContainerBusy(ctx context.Context, t ContainerTask, log *slog.Logger) (State, error) {
 	id := t.ID()
+
+	type result struct {
+		state State
+		err   error
+	}
+	ch := make(chan result, 1) // buffered: a late probe result must not leak the goroutine
+	go func() {
+		st, err := containerBusyHCS(id, log)
+		ch <- result{st, err}
+	}()
+
+	select {
+	case r := <-ch:
+		return r.state, r.err
+	case <-ctx.Done():
+		return Unknown, fmt.Errorf("busy probe for compute system %q: %w", id, ctx.Err())
+	}
+}
+
+// containerBusyHCS is the uncancellable HCS half of ContainerBusy.
+func containerBusyHCS(id string, log *slog.Logger) (State, error) {
 	c, err := openCompute(id)
 	if err != nil {
 		return Unknown, fmt.Errorf("opening compute system %q: %w", id, err)

pkg/runnerbusy/container_windows_test.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"log/slog"
 	"testing"
+	"time"
 
 	"github.com/Microsoft/hcsshim"
 	"github.com/containerd/containerd/v2/client"
@@ -128,6 +129,39 @@ func TestContainerBusy_Windows(t *testing.T) {
 	}
 }
 
+// TestContainerBusy_WindowsCtxBoundsAWedgedProbe pins that the caller's
+// ctx (busyProbeTimeout upstream) is actually enforced: hcsshim takes no
+// context, and before the probe ran HCS on its own goroutine a dead
+// shim/GCS wedged the whole sweep behind an uncancellable OpenContainer.
+// The wedged probe must come back Unknown — the fail-safe veto — not hang.
+func TestContainerBusy_WindowsCtxBoundsAWedgedProbe(t *testing.T) {
+	prev := openCompute
+	wedged := make(chan struct{})
+	t.Cleanup(func() {
+		openCompute = prev
+		close(wedged) // let the abandoned goroutine exit
+	})
+	openCompute = func(string) (hcsProcessLister, error) {
+		<-wedged // simulate HCS never answering
+		return nil, errors.New("wedged")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	got, err := ContainerBusy(ctx, &fakeTask{id: "job-1"}, quiet())
+	if elapsed := time.Since(start); elapsed > 5*time.Second {
+		t.Fatalf("ContainerBusy took %v; ctx did not bound the wedged probe", elapsed)
+	}
+	if got != Unknown {
+		t.Errorf("state = %v, want unknown for a wedged probe", got)
+	}
+	if err == nil {
+		t.Error("err = nil, want a ctx error for a wedged probe")
+	}
+}
+
 // TestContainerBusy_WindowsCloseErrorIsNotFatal pins that a failure to
 // release the HCS handle does not change the verdict — the answer is
 // already computed and the handle is process-local.

pkg/runtime/destroy_test.go

@@ -0,0 +1,92 @@
+package runtime
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/containerd/containerd/v2/client"
+)
+
+// TestWaitTaskExit pins the bounded kill-wait that replaced a bare
+// <-exitCh in Destroy. The production failure mode is a dead containerd
+// shim: the killed task's exit event never arrives, and the unbounded
+// receive blocked Destroy forever, pinning a scheduler concurrency slot
+// for over an hour. The wait must give up on its own — even under
+// context.Background(), which is what every Destroy caller passes.
+func TestWaitTaskExit(t *testing.T) {
+	t.Run("exit event arrives", func(t *testing.T) {
+		exitCh := make(chan client.ExitStatus, 1)
+		exitCh <- client.ExitStatus{}
+		if !waitTaskExit(context.Background(), exitCh, time.Minute) {
+			t.Fatal("waitTaskExit = false, want true when the exit event is delivered")
+		}
+	})
+
+	t.Run("dead shim never delivers: killWait bounds the wait", func(t *testing.T) {
+		exitCh := make(chan client.ExitStatus) // never written — the dead-shim case
+		start := time.Now()
+		// Background ctx on purpose: the bound must hold with no caller deadline.
+		if waitTaskExit(context.Background(), exitCh, 20*time.Millisecond) {
+			t.Fatal("waitTaskExit = true, want false when no exit event ever arrives")
+		}
+		if elapsed := time.Since(start); elapsed > 5*time.Second {
+			t.Fatalf("waitTaskExit took %v, want ~20ms — the bound did not hold", elapsed)
+		}
+	})
+
+	t.Run("ctx expiry bounds the wait before killWait", func(t *testing.T) {
+		exitCh := make(chan client.ExitStatus)
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		start := time.Now()
+		if waitTaskExit(ctx, exitCh, time.Hour) {
+			t.Fatal("waitTaskExit = true, want false on cancelled ctx")
+		}
+		if elapsed := time.Since(start); elapsed > 5*time.Second {
+			t.Fatalf("waitTaskExit took %v, want immediate return on cancelled ctx", elapsed)
+		}
+	})
+}
+
+// TestDestroyWaitCompleted pins the caller-side wait: bounded, and when a
+// teardown finishes right at the deadline the completed result must win —
+// a completed teardown mis-logged as "abandoned" sends an operator hunting
+// for a leak that does not exist.
+func TestDestroyWaitCompleted(t *testing.T) {
+	t.Run("teardown finishes first", func(t *testing.T) {
+		done := make(chan struct{}, 1)
+		done <- struct{}{}
+		if !destroyWaitCompleted(context.Background(), done) {
+			t.Fatal("destroyWaitCompleted = false, want true when teardown already finished")
+		}
+	})
+
+	t.Run("deadline expires with teardown still wedged", func(t *testing.T) {
+		done := make(chan struct{}, 1) // never written — the wedged case
+		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+		defer cancel()
+		start := time.Now()
+		if destroyWaitCompleted(ctx, done) {
+			t.Fatal("destroyWaitCompleted = true, want false when teardown never finishes")
+		}
+		if elapsed := time.Since(start); elapsed > 5*time.Second {
+			t.Fatalf("destroyWaitCompleted took %v, want ~20ms", elapsed)
+		}
+	})
+
+	t.Run("both ready: done wins over the expired ctx", func(t *testing.T) {
+		done := make(chan struct{}, 1)
+		done <- struct{}{}
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel() // ctx already expired AND done already delivered
+		// The select picks randomly between two ready cases, so run it
+		// enough times that a regression to coin-flip behavior cannot pass.
+		for i := 0; i < 100; i++ {
+			if !destroyWaitCompleted(ctx, done) {
+				t.Fatal("destroyWaitCompleted = false with done ready, want the completed teardown to win")
+			}
+			done <- struct{}{} // refill for the next iteration
+		}
+	})
+}