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..8b964c0 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,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. diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 73e748c..7b3f25a 100644 --- a/pkg/config/config_test.go +++ b/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) + } + } +} 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..6082c11 --- /dev/null +++ b/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 + } + }) +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 6de67b2..43ca865 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,130 @@ 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 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 + } r.cfg.Log.Info("destroying runner environment", "id", env.ID) - if r.cfg.OnTaskDestroy != nil { - r.cfg.OnTaskDestroy(env) + // 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() { + 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 true + case <-ctx.Done(): + // 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 + } } +} - // Kill the task if still running +// 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 +// 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 + // <-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,12 +1696,21 @@ 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) + } } } - // 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) } @@ -1623,8 +1751,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