From e9505f069234c588b1f6a0051ae4464170e4d135 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 1 Sep 2026 07:45:14 -0700 Subject: [PATCH 1/2] fix(scheduler): bound runner teardown and retry failed env creates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two production incidents on a Windows node (2026-09-01), one root cause: a dead containerd shim. Incident 1: after job completion, Destroy blocked forever on a bare <-exitCh — the killed task's exit event never arrives from a dead shim, and every caller passes context.Background(). The hung destroy sat behind the wait-goroutine's deferred slot release, pinning a MaxConcurrent slot for 1h+ and wedging two consecutive upgrade drains. Incident 2: five seconds into that hang, the next env create failed with "failed to create shim task: ttrpc: closed" and the job waited 1h43m, because the create-failure branch — unlike the claim-failure branch 15 lines up — never enqueued a retry, and GitHub does not re-deliver workflow_job webhooks. Teardown is now bounded end to end: - Runtime.Destroy derives a WithTimeout ctx (new runtime.destroy_timeout knob; default 5m, non-positive falls back to the default — the zero-value Config fails safe, there is no unbounded configuration) and runs the teardown steps on their own goroutine. On expiry Destroy returns an error and ABANDONS the remaining teardown: steps that take no ctx (Dind.Stop, HCS/HNS internals) may stay parked, and what they leave behind is reclaimed by the periodic dead-container reaper (ReapDeadContainers) and the startup CleanOrphans sweep. - The kill-wait itself is separately bounded (30s) so a missing exit event cannot starve the teardown steps that still can succeed. - The wait-goroutines (local sem and linuxSem twin) release their concurrency slot as soon as the job is untracked, BEFORE destroy, so even a bounded teardown can never delay the next job's admission. Safe because every env is keyed by its unique claim.RunnerName — container ID, runner dir, netns/HCN endpoint all derive from it — so a freshly admitted job cannot collide with an in-flight teardown. Failed env creates now retry: - Runtime.Create gets a small in-place ladder (5s, 15s) for transient errors while the slot and JIT claim are still held; skipped for non-retryable errors, during drain, and on ctx cancellation. - When the ladder runs out, the create-failure branches (local and Linux dispatch) enqueue on the existing retry queue, mirroring the claim-failure branch — "ttrpc: closed" classifies retryable. Hardening: the Windows busy probe (runnerbusy.ContainerBusy) now runs its uncancellable hcsshim calls on a goroutine and honors ctx, so busyProbeTimeout is actually enforced on Windows and a dead shim/GCS cannot wedge the sweep behind it. Composes with #151: only post-completion/exit destroys change behavior; the busy-veto and sweep nomination logic are untouched, and the reap path's destroy picks up the same bound via Runtime.Destroy. --- cmd/ephemerd/main.go | 2 + pkg/config/config.go | 24 ++++ pkg/config/config_test.go | 22 ++++ pkg/runnerbusy/container_windows.go | 31 ++++- pkg/runnerbusy/container_windows_test.go | 34 ++++++ pkg/runtime/destroy_test.go | 50 ++++++++ pkg/runtime/runtime.go | 104 +++++++++++++++- pkg/scheduler/retry_create_test.go | 127 +++++++++++++++++++ pkg/scheduler/scheduler.go | 149 ++++++++++++++++++++--- 9 files changed, 522 insertions(+), 21 deletions(-) create mode 100644 pkg/runtime/destroy_test.go create mode 100644 pkg/scheduler/retry_create_test.go diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 3511f1e..bb1a712 100644 --- a/cmd/ephemerd/main.go +++ b/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 { diff --git a/pkg/config/config.go b/pkg/config/config.go index bcb1d2e..959bd7f 100644 --- a/pkg/config/config.go +++ b/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,18 @@ 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 { + if r.DestroyTimeout <= 0 { + 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. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 73e748c..3c2cc70 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -2471,3 +2471,25 @@ 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 + } + 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) + } + } +} diff --git a/pkg/runnerbusy/container_windows.go b/pkg/runnerbusy/container_windows.go index 5da6eb1..c34328a 100644 --- a/pkg/runnerbusy/container_windows.go +++ b/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) diff --git a/pkg/runnerbusy/container_windows_test.go b/pkg/runnerbusy/container_windows_test.go index 3c7678b..340a8a0 100644 --- a/pkg/runnerbusy/container_windows_test.go +++ b/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. diff --git a/pkg/runtime/destroy_test.go b/pkg/runtime/destroy_test.go new file mode 100644 index 0000000..4eebe9a --- /dev/null +++ b/pkg/runtime/destroy_test.go @@ -0,0 +1,50 @@ +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) + } + }) +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 6de67b2..af431f3 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -173,7 +173,13 @@ type Config struct { // with a zero grace. Callers pass // config.RuntimeConfig.ResolvedOrphanContainerGrace(). OrphanContainerGrace time.Duration - Log *slog.Logger + // DestroyTimeout bounds a single Destroy end to end. Non-positive + // falls back to defaultDestroyTimeout, so the zero-value Config fails + // SAFE: a construction site that forgets this field gets a finite + // bound, never an unbounded teardown. Callers pass + // config.RuntimeConfig.ResolvedDestroyTimeout(). + DestroyTimeout time.Duration + Log *slog.Logger } // resolveRuntimeName picks the containerd runtime handler for a job @@ -1559,17 +1565,85 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err return env, nil } +// defaultDestroyTimeout is the fallback bound for Destroy when +// Config.DestroyTimeout is non-positive. A healthy teardown takes ~15s; +// 5m is generous headroom for a slow disk or an HNS hiccup while still +// guaranteeing the caller gets its thread (and concurrency slot) back. +const defaultDestroyTimeout = 5 * time.Minute + +// destroyKillWait is how long Destroy waits for a killed task's exit +// event before giving up on it and pressing on with the rest of the +// teardown. Deliberately much shorter than the overall destroy bound: a +// dead shim never delivers the exit event at all, and burning the whole +// destroy budget on it would starve the steps that still CAN succeed +// (network teardown, dind stop, directory removal). +const destroyKillWait = 30 * time.Second + // Destroy tears down a runner environment completely. +// +// Destroy is BOUNDED: it returns within Config.DestroyTimeout (default +// defaultDestroyTimeout) even when a teardown step wedges. This is a +// production requirement, not a nicety — a dead containerd shim makes the +// killed task's exit event never arrive AND makes task/container RPCs +// hang, and an unbounded Destroy then blocks its caller forever (observed +// on metal: one dead shim pinned a scheduler slot for 1h+ and stalled two +// upgrade drains in a row). +// +// Contract on timeout: Destroy returns an error and ABANDONS whatever +// teardown remains — the worker goroutine may still be blocked inside an +// uncancellable call (env.Dind.Stop and the HCS/HNS operations take no +// ctx; rewriting those subsystems is out of scope). Everything an +// abandoned teardown leaves behind is reclaimed asynchronously: the +// periodic dead-container reaper (ReapDeadContainers) deletes the +// container + snapshot once its task is provably dead past the grace +// window, and the startup CleanOrphans / dind.CleanupStaleDindNamespaces +// sweep the rest on the next boot. Leaking one parked goroutine per +// wedged shim is the accepted cost of never wedging the scheduler. func (r *Runtime) Destroy(ctx context.Context, env *RunnerEnv) error { ctx = namespaces.WithNamespace(ctx, namespace) + timeout := r.cfg.DestroyTimeout + if timeout <= 0 { + timeout = defaultDestroyTimeout + } + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + r.cfg.Log.Info("destroying runner environment", "id", env.ID) if r.cfg.OnTaskDestroy != nil { r.cfg.OnTaskDestroy(env) } - // Kill the task if still running + // Run the teardown steps on their own goroutine so Destroy can return + // on ctx expiry even while a step that ignores ctx is still wedged. + // The channel is buffered: a late-finishing teardown sends its result + // and exits instead of leaking on a channel nobody reads. + done := make(chan struct{}, 1) + go func() { + r.destroySteps(ctx, env) + done <- struct{}{} + }() + + select { + case <-done: + return nil + case <-ctx.Done(): + r.cfg.Log.Error("destroy timed out; abandoning teardown of runner environment to the orphan reaper", + "id", env.ID, "timeout", timeout) + return fmt.Errorf("destroying %s: %w", env.ID, ctx.Err()) + } +} + +// destroySteps is the teardown sequence proper. Every step is +// best-effort (log and continue) so one wedged or failing step cannot +// keep the later, independent steps from running. Steps that take a ctx +// share Destroy's bounded ctx; steps that don't (Dind.Stop, HCS/HNS +// internals) are bounded only by Destroy abandoning this goroutine. +func (r *Runtime) destroySteps(ctx context.Context, env *RunnerEnv) { + // Kill the task if still running. The exit-event wait is bounded: a + // dead shim never delivers it (this exact receive used to be a bare + // <-exitCh and blocked forever in production). status, err := env.Task.Status(ctx) if err == nil && status.Status == client.Running { if err := env.Task.Kill(ctx, 9); err != nil { @@ -1577,7 +1651,10 @@ func (r *Runtime) Destroy(ctx context.Context, env *RunnerEnv) error { } exitCh, err := env.Task.Wait(ctx) if err == nil { - <-exitCh + if !waitTaskExit(ctx, exitCh, destroyKillWait) { + r.cfg.Log.Warn("killed task never reported exit; proceeding with teardown", + "id", env.ID, "waited", destroyKillWait) + } } } @@ -1623,8 +1700,27 @@ func (r *Runtime) Destroy(ctx context.Context, env *RunnerEnv) error { // Windows sharing violation from the just-exited container's file handles. removeJobWorkdir(r.cfg.DataDir, env.ID, r.cfg.Log) + // Logged here rather than in Destroy so a teardown that finishes late + // — after Destroy already timed out and returned — still leaves an + // operator-visible record that it eventually completed. r.cfg.Log.Info("runner environment destroyed", "id", env.ID) - return nil +} + +// waitTaskExit waits for a killed task's exit notification, but never +// longer than killWait (or the caller's ctx). Factored out of +// destroySteps so the bound is unit-testable without faking a full +// containerd client.Task. Returns true when the exit event arrived. +func waitTaskExit(ctx context.Context, exitCh <-chan client.ExitStatus, killWait time.Duration) bool { + timer := time.NewTimer(killWait) + defer timer.Stop() + select { + case <-exitCh: + return true + case <-ctx.Done(): + return false + case <-timer.C: + return false + } } // Wait blocks until the runner environment's task exits. diff --git a/pkg/scheduler/retry_create_test.go b/pkg/scheduler/retry_create_test.go new file mode 100644 index 0000000..358cf7b --- /dev/null +++ b/pkg/scheduler/retry_create_test.go @@ -0,0 +1,127 @@ +package scheduler + +import ( + "context" + "errors" + "testing" + "time" +) + +// tinyBackoffs keeps the in-place retry tests fast while still exercising +// the sleep-between-attempts path. +var tinyBackoffs = []time.Duration{time.Millisecond, time.Millisecond} + +// TestRetryEnvCreate pins the in-place create retry that rides out +// transient runtime failures (the production case: a dead containerd shim +// failing Create with "ttrpc: closed") without giving up the held slot +// and JIT claim. The short-circuits matter as much as the retries: a +// permanent error or a draining node must not burn backoff time. +func TestRetryEnvCreate(t *testing.T) { + transient := errors.New("failed to create shim task: ttrpc: closed") + permanent := errors.New("image does not exist") + + alwaysRetryable := func(error) bool { return true } + notDraining := func() bool { return false } + + t.Run("first attempt succeeds: no retries", func(t *testing.T) { + calls := 0 + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + alwaysRetryable, notDraining, + func() error { calls++; return nil }) + if err != nil || calls != 1 { + t.Fatalf("err = %v, calls = %d; want nil, 1", err, calls) + } + }) + + t.Run("transient failure heals on a later attempt", func(t *testing.T) { + calls := 0 + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + alwaysRetryable, notDraining, + func() error { + calls++ + if calls < 2 { + return transient + } + return nil + }) + if err != nil || calls != 2 { + t.Fatalf("err = %v, calls = %d; want nil, 2", err, calls) + } + }) + + t.Run("ladder exhausted: last error returned", func(t *testing.T) { + calls := 0 + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + alwaysRetryable, notDraining, + func() error { calls++; return transient }) + if !errors.Is(err, transient) { + t.Fatalf("err = %v, want the create error", err) + } + if want := 1 + len(tinyBackoffs); calls != want { + t.Fatalf("calls = %d, want %d (initial + ladder)", calls, want) + } + }) + + t.Run("permanent error short-circuits", func(t *testing.T) { + calls := 0 + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + func(error) bool { return false }, notDraining, + func() error { calls++; return permanent }) + if !errors.Is(err, permanent) || calls != 1 { + t.Fatalf("err = %v, calls = %d; want permanent error after 1 call", err, calls) + } + }) + + t.Run("draining node does not retry", func(t *testing.T) { + calls := 0 + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + alwaysRetryable, func() bool { return true }, + func() error { calls++; return transient }) + if !errors.Is(err, transient) || calls != 1 { + t.Fatalf("err = %v, calls = %d; want error after 1 call during drain", err, calls) + } + }) + + t.Run("drain starting mid-backoff stops the next attempt", func(t *testing.T) { + calls := 0 + draining := false + err := retryEnvCreate(context.Background(), testLogger(), tinyBackoffs, + alwaysRetryable, func() bool { return draining }, + func() error { + calls++ + draining = true // drain begins after the first failure + return transient + }) + if !errors.Is(err, transient) || calls != 1 { + t.Fatalf("err = %v, calls = %d; want error after 1 call", err, calls) + } + }) + + t.Run("ctx cancellation aborts the backoff", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + calls := 0 + start := time.Now() + err := retryEnvCreate(ctx, testLogger(), []time.Duration{time.Hour}, + alwaysRetryable, notDraining, + func() error { calls++; return transient }) + if !errors.Is(err, transient) || calls != 1 { + t.Fatalf("err = %v, calls = %d; want error after 1 call", err, calls) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("took %v; cancelled ctx must not sit out the backoff", elapsed) + } + }) +} + +// TestClassifyErr_DeadShimIsRetryable pins the classification the create +// retry paths (both in-place and the retry-queue enqueue) depend on: the +// exact error a dead shim produced in production must NOT classify as +// non-retryable, or the job is dropped forever (observed: 1h43m of a job +// sitting queued because nothing retried a single failed create). +func TestClassifyErr_DeadShimIsRetryable(t *testing.T) { + err := errors.New(`creating task for abc123: failed to create shim task: ttrpc: closed`) + if class := classifyErr(err); class == errNonRetryable { + t.Fatalf("classifyErr(dead shim create error) = %v, want retryable", class) + } +} diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go index 566b83b..c3f72d6 100644 --- a/pkg/scheduler/scheduler.go +++ b/pkg/scheduler/scheduler.go @@ -1205,13 +1205,18 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent jobCtx, cancel := s.jobContext() if err := s.cfg.LinuxDispatcher.Create(jobCtx, claim.RunnerName, image, claim.RunnerConfig, event.Provider.Name(), event.Repo); err != nil { - log.Error("dispatch create failed", "error", err) + log.Error("dispatch create failed", "error", err, "error_class", classifyErr(err)) if rmErr := event.Provider.ReleaseJob(ctx, claim); rmErr != nil { log.Warn("failed to remove ghost runner", "runner_id", claim.RunnerID, "error", rmErr) } unsee() cancel() <-s.linuxSem + // Mirror the claim-failure branch above (and the local create + // path): a failed dispatch create used to drop the job forever — + // webhooks are never re-delivered. Slot released first; no-op for + // non-retryable errors or while draining. + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -1230,7 +1235,17 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent // Wait for the job to finish in the background go func() { - defer func() { <-s.linuxSem }() + // Same slot-release ordering as the local wait-goroutine: release + // once the job is untracked, never behind the destroy. The + // deferred call is only the panic/early-return backstop. + released := false + release := func() { + if !released { + released = true + <-s.linuxSem + } + } + defer release() exitCode, err := s.cfg.LinuxDispatcher.Wait(jobCtx, claim.RunnerName) if err != nil { @@ -1257,6 +1272,11 @@ func (s *Scheduler) handleLinuxJob(ctx context.Context, event providers.JobEvent metrics.JobsActive.Dec() } + // Release BEFORE the destroy RPC, mirroring the local path: the + // dispatched env is keyed by its unique claim.RunnerName, so a + // newly admitted job cannot collide with this teardown. + release() + if err := s.cfg.LinuxDispatcher.Destroy(context.Background(), claim.RunnerName); err != nil { log.Warn("dispatch destroy failed", "error", err) } @@ -1613,16 +1633,30 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent // Create the runner environment with job timeout. // Derived from jobsCtx, not ctx: the job keeps running across SIGTERM // until it finishes or drain() gives up (see bindContexts). + // + // Create is retried IN PLACE for transient failures (a dead shim's + // "ttrpc: closed", a containerd restart, ...): the slot is held and + // the JIT claim already minted, so a couple of short-backoff attempts + // here are far cheaper than release-and-requeue. The retry queue + // below is the real safety net when these run out. jobCtx, cancel := s.jobContext() - env, err := s.cfg.Runtime.Create(jobCtx, runtime.CreateConfig{ - ID: claim.RunnerName, - Image: image, - JITConfig: claim.RunnerConfig, - Env: claim.Env, - Entrypoint: claim.Entrypoint, - }) + var env *runtime.RunnerEnv + err = retryEnvCreate(jobCtx, log, createEnvBackoffs, + func(err error) bool { return classifyErr(err) != errNonRetryable }, + s.isDraining, + func() error { + var cerr error + env, cerr = s.cfg.Runtime.Create(jobCtx, runtime.CreateConfig{ + ID: claim.RunnerName, + Image: image, + JITConfig: claim.RunnerConfig, + Env: claim.Env, + Entrypoint: claim.Entrypoint, + }) + return cerr + }) if err != nil { - log.Error("failed to create runner environment", "error", err) + log.Error("failed to create runner environment", "error", err, "error_class", classifyErr(err)) // Remove the ghost runner since the container won't start if rmErr := event.Provider.ReleaseJob(ctx, claim); rmErr != nil { log.Warn("failed to remove ghost runner", "runner_id", claim.RunnerID, "error", rmErr) @@ -1633,6 +1667,13 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent unsee() cancel() <-s.sem + // Mirror the claim-failure branch above: without this a job whose + // env create failed was dropped forever (GitHub never re-delivers + // the webhook) — on metal that was 1h43m of a job sitting queued + // after a dead shim failed a single create. Slot released FIRST so + // the retry never fires while we still hold it; no-op when the + // error is non-retryable or the node is draining. + s.enqueueRetryIfEligible(ctx, event, err) return } @@ -1652,9 +1693,20 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent // Wait for the job to finish in the background go func() { - defer func() { - <-s.sem // release concurrency slot - }() + // Slot release is decoupled from teardown. It used to hang off a + // bare defer, which meant the slot came back only after Destroy + // returned — so a wedged teardown (dead shim) pinned a + // MaxConcurrent slot for its whole hang. The slot is released as + // soon as the job is untracked below; the deferred call is only + // the panic/early-return backstop. + released := false + release := func() { + if !released { + released = true + <-s.sem + } + } + defer release() exitCode, err := s.cfg.Runtime.Wait(jobCtx, env) if err != nil { @@ -1676,7 +1728,21 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent rj, exists := s.running[key] if exists { s.untrackRunningLocked(key, rj) - s.mu.Unlock() + } + s.mu.Unlock() + + // Release the concurrency slot BEFORE destroying: the job is + // untracked, so its demand is gone, and the teardown must not be + // able to delay the next job's admission (Destroy is bounded, but + // even a bounded 5m hang is 5m of a dead slot). This is safe + // against a new job racing this env's teardown: every env is + // keyed by its unique claim.RunnerName — container ID, per-job + // runner dir, netns/HCN endpoint, dind socket all derive from it — + // so a freshly admitted job cannot touch anything this teardown + // still owns. + release() + + if exists { metrics.JobsActive.Dec() if err := s.cfg.Runtime.Destroy(context.Background(), env); err != nil { log.Warn("failed to destroy runner environment", "error", err) @@ -1689,8 +1755,6 @@ func (s *Scheduler) handleLocalJob(ctx context.Context, event providers.JobEvent log.Debug("deregister runner after local cleanup", "error", err) } } - } else { - s.mu.Unlock() } // Self-heal: re-provision if this runner's dispatched job never ran. @@ -2303,6 +2367,59 @@ func (s *Scheduler) enqueueRetryIfEligible(ctx context.Context, event providers. s.retry.Add(event, s.retryHandler, err) } +// isDraining reports whether the scheduler has stopped taking new work +// (shutdown drain or operator cordon). +func (s *Scheduler) isDraining() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.draining +} + +// createEnvBackoffs is the short in-place backoff ladder for a failed +// runner-environment create. Deliberately tiny: the caller holds a +// concurrency slot and a minted JIT claim for the whole ladder, so this +// only rides out blips (a shim crash mid-create, a containerd restart) — +// anything longer-lived goes through release-and-requeue on the retry +// queue instead. +var createEnvBackoffs = []time.Duration{5 * time.Second, 15 * time.Second} + +// retryEnvCreate runs create, retrying in place up to len(backoffs) +// extra times with the given backoff between attempts. It gives up early +// — returning the last error — when: +// - retryable(err) is false (a permanent error won't heal in 15s), or +// - drained() is true (a draining node must not sit on a slot waiting +// to start NEW work; the caller releases and returns, as it always +// did), or +// - ctx is cancelled mid-backoff (shutdown, job timeout). +// +// Pure aside from the injected callbacks so the attempt/short-circuit +// behavior is unit-testable without a Scheduler. +func retryEnvCreate(ctx context.Context, log *slog.Logger, backoffs []time.Duration, + retryable func(error) bool, drained func() bool, create func() error) error { + err := create() + for i := 0; err != nil && i < len(backoffs); i++ { + if !retryable(err) || drained() { + return err + } + log.Warn("failed to create runner environment; retrying in place", + "error", err, "attempt", i+1, "backoff", backoffs[i]) + timer := time.NewTimer(backoffs[i]) + select { + case <-timer.C: + case <-ctx.Done(): + timer.Stop() + return err + } + // Drain may have started during the backoff; re-check before + // spending another create against a node that stopped taking work. + if drained() { + return err + } + err = create() + } + return err +} + // retryHandler is the callback the retry queue invokes on each fire. // It re-enters the top-level dispatch (handleQueued) with the ORIGINAL // event. Because handleQueued would otherwise dedup our own retry via From ee70361e5c4a08ef29ecf32aebde3c67fcfe45c4 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Tue, 1 Sep 2026 08:41:55 -0700 Subject: [PATCH 2/2] fix(runtime): close the review holes in the destroy bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the parent commit found the bound leaked at both ends; all fixed here: - The OnTaskDestroy hook ran synchronously BEFORE the bounded goroutine. On Windows the hook closes an HCS handle — exactly the class of call that wedges on a dead shim — so a wedged hook blocked Destroy before any timeout machinery existed. The hook now runs inside the teardown goroutine. - destroySteps shared Destroy's caller-bounded ctx, whose deferred cancel fired the moment Destroy returned on timeout — so the 'continues in the background' contract was false: every remaining ctx-aware step failed instantly with context.Canceled once a wedge cleared. The teardown goroutine now gets its own detached ctx (Background + namespace, budget 2x DestroyTimeout) so an abandoned teardown can actually finish late. - Task.Delete in destroySteps lacked WithProcessKill: after an abandoned kill-wait, a task whose process died but whose exit event was lost failed the delete with FailedPrecondition, the container delete failed too, and the periodic reaper could never converge (it fails safe on a Running-reading task) — the snapshot survived until restart. - ResolvedDestroyTimeout now floors at 1s: TOML decodes a bare integer destroy_timeout = 300 as 300 NANOSECONDS without error, which would have made every teardown 'time out' instantly and abandon its cleanup. - destroyWaitCompleted prefers the done channel when both it and the deadline are ready, so a teardown finishing at the buzzer is not mis-logged as abandoned; factored out and unit-tested (100-iteration anti-coin-flip test). - The abandonment contract comment now states honestly that leaked HNS endpoints + pool IPs are reclaimed by NOTHING (restart re-adopts and re-reserves them; pool headroom absorbs the leak) — the known residual, accepted over holding the scheduler hostage. --- pkg/config/config.go | 9 ++- pkg/config/config_test.go | 7 +++ pkg/runtime/destroy_test.go | 42 ++++++++++++++ pkg/runtime/runtime.go | 109 ++++++++++++++++++++++++++---------- 4 files changed, 137 insertions(+), 30 deletions(-) diff --git a/pkg/config/config.go b/pkg/config/config.go index 959bd7f..8b964c0 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -431,7 +431,14 @@ func (r RuntimeConfig) ResolvedOrphanContainerGrace() time.Duration { // a teardown step is genuinely wedged — there is deliberately no way to // configure an unbounded destroy. func (r RuntimeConfig) ResolvedDestroyTimeout() time.Duration { - if r.DestroyTimeout <= 0 { + // 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 diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 3c2cc70..7b3f25a 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -2485,6 +2485,13 @@ func TestResolvedDestroyTimeout(t *testing.T) { {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} diff --git a/pkg/runtime/destroy_test.go b/pkg/runtime/destroy_test.go index 4eebe9a..6082c11 100644 --- a/pkg/runtime/destroy_test.go +++ b/pkg/runtime/destroy_test.go @@ -48,3 +48,45 @@ func TestWaitTaskExit(t *testing.T) { } }) } + +// 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 + } + }) +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index af431f3..43ca865 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -1589,57 +1589,102 @@ const destroyKillWait = 30 * time.Second // on metal: one dead shim pinned a scheduler slot for 1h+ and stalled two // upgrade drains in a row). // -// Contract on timeout: Destroy returns an error and ABANDONS whatever -// teardown remains — the worker goroutine may still be blocked inside an -// uncancellable call (env.Dind.Stop and the HCS/HNS operations take no -// ctx; rewriting those subsystems is out of scope). Everything an -// abandoned teardown leaves behind is reclaimed asynchronously: the -// periodic dead-container reaper (ReapDeadContainers) deletes the -// container + snapshot once its task is provably dead past the grace -// window, and the startup CleanOrphans / dind.CleanupStaleDindNamespaces -// sweep the rest on the next boot. Leaking one parked goroutine per -// wedged shim is the accepted cost of never wedging the scheduler. +// Contract on timeout: Destroy returns an error and ABANDONS the wait — +// the worker goroutine keeps running on its own detached context (budget +// 2×DestroyTimeout) and will usually finish late, logging "runner +// environment destroyed" when it does. It may instead stay blocked inside +// an uncancellable call (env.Dind.Stop and the HCS/HNS operations take no +// ctx; rewriting those subsystems is out of scope). What a teardown that +// never finishes leaves behind, and who reclaims it: +// - container + snapshot: the periodic dead-container reaper +// (ReapDeadContainers), once the task is provably dead past the grace +// window; startup CleanOrphans otherwise. +// - per-job dind namespace: startup dind.CleanupStaleDindNamespaces. +// - the HNS endpoint + its pool IP (Windows): NOT reaped by anything — +// a restart merely re-adopts the endpoint and re-reserves its address +// (adoptExistingEndpoints), which is why the L2Bridge pool is sized +// with headroom. A repeatedly wedging node leaks one address per +// abandoned teardown; the per-step Warn logs are the observable. +// Known residual, accepted over holding the scheduler hostage. +// - one parked goroutine per wedged uncancellable call. func (r *Runtime) Destroy(ctx context.Context, env *RunnerEnv) error { - ctx = namespaces.WithNamespace(ctx, namespace) - timeout := r.cfg.DestroyTimeout if timeout <= 0 { timeout = defaultDestroyTimeout } - ctx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() r.cfg.Log.Info("destroying runner environment", "id", env.ID) - if r.cfg.OnTaskDestroy != nil { - r.cfg.OnTaskDestroy(env) - } - - // Run the teardown steps on their own goroutine so Destroy can return - // on ctx expiry even while a step that ignores ctx is still wedged. + // The teardown goroutine gets its OWN context, detached from the + // caller's: derived from Background so the caller canceling (or this + // function returning and running its deferred cancel) cannot kill the + // in-flight background teardown — that was a review finding: sharing + // the caller-bounded ctx meant "continues in the background" was + // false, because the moment Destroy returned on timeout, cancel() + // fired and every remaining ctx-aware step (task delete, network + // teardown, container delete) failed instantly with context.Canceled. + // The steps budget is twice the caller's wait so that when a single + // wedged call finally clears, the rest of the teardown still has time + // to actually run. + stepsCtx, stepsCancel := context.WithTimeout( + namespaces.WithNamespace(context.Background(), namespace), 2*timeout) + + // The OnTaskDestroy hook runs INSIDE the bounded goroutine, not + // before it: on Windows the hook closes an HCS handle + // (closers.closeAndRemove), and HCS calls are exactly the class that + // wedges on a dead shim — invoked synchronously here it would block + // Destroy before any of the timeout machinery existed. + // // The channel is buffered: a late-finishing teardown sends its result // and exits instead of leaking on a channel nobody reads. done := make(chan struct{}, 1) go func() { - r.destroySteps(ctx, env) + defer stepsCancel() + if r.cfg.OnTaskDestroy != nil { + r.cfg.OnTaskDestroy(env) + } + r.destroySteps(stepsCtx, env) done <- struct{}{} }() + waitCtx, waitCancel := context.WithTimeout(ctx, timeout) + defer waitCancel() + if destroyWaitCompleted(waitCtx, done) { + return nil + } + r.cfg.Log.Error("destroy timed out; abandoning teardown of runner environment to the orphan reaper", + "id", env.ID, "timeout", timeout) + return fmt.Errorf("destroying %s: %w", env.ID, waitCtx.Err()) +} + +// destroyWaitCompleted blocks until the teardown goroutine reports done or +// ctx expires. When both are ready simultaneously (a teardown finishing +// right at the deadline), done wins — otherwise a completed teardown gets +// mis-logged as abandoned, which sends an operator hunting for a leak that +// doesn't exist. Factored out so that preference is unit-testable. +func destroyWaitCompleted(ctx context.Context, done <-chan struct{}) bool { select { case <-done: - return nil + return true case <-ctx.Done(): - r.cfg.Log.Error("destroy timed out; abandoning teardown of runner environment to the orphan reaper", - "id", env.ID, "timeout", timeout) - return fmt.Errorf("destroying %s: %w", env.ID, ctx.Err()) + // The select above picks randomly when both cases are ready; + // re-check done so a just-finished teardown is reported as such. + select { + case <-done: + return true + default: + return false + } } } // destroySteps is the teardown sequence proper. Every step is // best-effort (log and continue) so one wedged or failing step cannot // keep the later, independent steps from running. Steps that take a ctx -// share Destroy's bounded ctx; steps that don't (Dind.Stop, HCS/HNS -// internals) are bounded only by Destroy abandoning this goroutine. +// receive the goroutine's own detached ctx (budget 2×DestroyTimeout, +// survives Destroy returning); steps that don't (Dind.Stop, HCS/HNS +// internals) are unbounded within this goroutine — Destroy's caller-side +// wait is what stays bounded. func (r *Runtime) destroySteps(ctx context.Context, env *RunnerEnv) { // Kill the task if still running. The exit-event wait is bounded: a // dead shim never delivers it (this exact receive used to be a bare @@ -1658,8 +1703,14 @@ func (r *Runtime) destroySteps(ctx context.Context, env *RunnerEnv) { } } - // Delete task - if _, err := env.Task.Delete(ctx); err != nil { + // Delete task. WithProcessKill matters after an abandoned kill-wait: + // without it a task whose process died but whose exit event was lost + // (dead shim) fails the delete with FailedPrecondition, the container + // delete below then fails too, and the periodic reaper can never + // converge — containerTaskLiveness fails safe on a task that still + // reads as Running, so the pinned snapshot would survive until the + // next daemon restart. Matches teardownContainer. + if _, err := env.Task.Delete(ctx, client.WithProcessKill); err != nil { r.cfg.Log.Warn("failed to delete task", "id", env.ID, "error", err) }