From e64b50cd1822d166fa7f6e2b6f6e0d96c32e7792 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Thu, 5 Nov 2026 19:50:28 +0100 Subject: [PATCH] daemon/internal/jobs: Reconcile runs on daemon restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolve the runs an unclean stop leaves behind, after a crash or a live-restore shutdown, before the scheduler re-arms. A run whose start was never recorded is failed, as its outcome cannot be trusted; a container it left behind is stopped first, so it cannot overlap the job's next run; one whose ID was never recorded is found by its name and removed. A started run is re-attached to the exit watcher, which records an exit from the downtime before enforcing a timeout re-armed from the original start. A job stuck running with a terminal run is returned to idle. Co-authored-by: Paweł Gronowski Signed-off-by: Guillaume Lours Signed-off-by: Paweł Gronowski --- daemon/internal/jobs/restore.go | 82 ++++++ daemon/internal/jobs/restore_test.go | 408 +++++++++++++++++++++++++++ 2 files changed, 490 insertions(+) create mode 100644 daemon/internal/jobs/restore.go create mode 100644 daemon/internal/jobs/restore_test.go diff --git a/daemon/internal/jobs/restore.go b/daemon/internal/jobs/restore.go new file mode 100644 index 0000000000..837a454812 --- /dev/null +++ b/daemon/internal/jobs/restore.go @@ -0,0 +1,82 @@ +package jobs + +import ( + "context" + + cerrdefs "github.com/containerd/errdefs" + "github.com/containerd/log" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" +) + +// Restore reconciles state left behind by an unclean daemon stop — a crash, +// or a live-restore shutdown that deliberately detached from still-running +// containers. It runs after the daemon's own container restore and before +// Start arms schedule triggers. +// +// For each job's in-flight run (at most one, by the concurrency invariant): +// +// - a run whose start was never recorded is failed outright: whether its +// container was never created, created but never started, or started +// just as the daemon died, the start evidence is lost, and guessing an +// outcome from a container that never ran would fabricate a success. +// Any container left behind is stopped before the job is released, so +// it cannot overlap the job's next run; if stopping fails, the job +// stays occupied until the container is observed exiting. A container +// whose ID was never recorded never started; it is found by its name +// and removed; +// - a started run is re-attached to the standard exit watcher, which +// resolves it from the container's actual state — immediately if the +// container exited during the downtime, or whenever it exits if it is +// still running. A timeout deadline is re-armed from the original start +// time, and enforced immediately when it expired while the daemon was +// down; +// - a job stuck in the running state with a terminal (or no) run — a +// crash between the run's terminal write and the job's idle write — is +// returned to idle. +// +// Runs kept as tombstone history of removed jobs are not reconciled: they +// may stay in flight forever, the documented cost of retaining history past +// a job's removal. +// +// Restore must be called at most once, before Start: a second call would +// attach duplicate watchers to the re-attached runs. +func (m *Manager) Restore(ctx context.Context) { + // Reconciliation outlives the startup call the same way watchers do. + ctx = context.WithoutCancel(ctx) + for _, job := range m.store.Jobs() { + m.locks.Lock(job.ID) + m.restoreJobLocked(ctx, job) + m.locks.Unlock(job.ID) + } +} + +func (m *Manager) restoreJobLocked(ctx context.Context, job *jobsv0.Job) { + run, err := m.store.LatestRun(job.ID) + if err != nil || isTerminalRunState(run.State) { + if job.State == jobsv0.JobStateRunning { + var runID string + if err == nil { + runID = run.ID + } + + m.releaseJobLocked(ctx, job.ID, runID) + } + return + } + + if run.StartedAtNano == 0 { + if run.ContainerID == "" { + // A crash between creating the container and recording its ID + // leaves a container known only by its name. It never started, + // so there is nothing to keep it for. + if err := m.backend.ContainerRm(ctx, runContainerName(job, run)); err != nil && !cerrdefs.IsNotFound(err) { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not remove unrecorded container of unstarted restored run") + } + } + + m.failUnstartedLocked(ctx, job, run, "daemon restarted before the run start was recorded") + return + } + + m.watch(ctx, job, run) +} diff --git a/daemon/internal/jobs/restore_test.go b/daemon/internal/jobs/restore_test.go new file mode 100644 index 0000000000..18cda64fc7 --- /dev/null +++ b/daemon/internal/jobs/restore_test.go @@ -0,0 +1,408 @@ +package jobs + +import ( + "errors" + "testing" + "time" + + cerrdefs "github.com/containerd/errdefs" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" + "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" + "gotest.tools/v3/poll" +) + +// newRestartableManager builds a manager whose storage is kept so a test +// can simulate a daemon restart against the same persisted records. +func newRestartableManager(t *testing.T) (*Manager, *fakeBackend, *testKV) { + t.Helper() + kv := newTestKV(t) + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + fake := newFakeBackend() + m := NewManager(store, fake, testNameGenerator) + t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) }) + return m, fake, kv +} + +// restartManager shuts the previous manager down, reloads the store from +// storage — the path a real daemon restart takes — and reconciles. +func restartManager(t *testing.T, prev *Manager, kv *testKV, fake *fakeBackend) *Manager { + t.Helper() + assert.NilError(t, prev.Shutdown(shutdownCtx(t))) + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + m := NewManager(store, fake, testNameGenerator) + t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) }) + m.Restore(t.Context()) + return m +} + +func TestRestoreReattachesRunningContainer(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + _, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + // The container is still running across the restart (live-restore). + m2 := restartManager(t, m1, kv, fake) + kept, err := m2.InspectRun(t.Context(), "backup", run.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(kept.State, jobsv0.RunStateRunning)) + + // The re-attached watcher records the exit like any other run. + fake.exit(run.ContainerID, 0, nil) + done := waitTerminal(t, m2, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded)) + waitJobSettled(t, m2, "backup", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) +} + +func TestRestoreResolvesContainerExitedDuringDowntime(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + _, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + assert.NilError(t, m1.Shutdown(shutdownCtx(t))) + // The container exits while the daemon is down; the record still says + // running when the store reloads. + fake.exit(run.ContainerID, 3, nil) + + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + m2 := NewManager(store, fake, testNameGenerator) + t.Cleanup(func() { assert.Check(t, m2.Shutdown(shutdownCtx(t))) }) + m2.Restore(t.Context()) + + done := waitTerminal(t, m2, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Assert(t, done.ExitCode != nil) + assert.Check(t, is.Equal(done.ExitCode.Value, int64(3))) +} + +func TestRestoreFailsLostContainer(t *testing.T) { + m1, _, kv := newRestartableManager(t) + _, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + // The container vanished entirely while the daemon was down. + m2 := restartManager(t, m1, kv, newFakeBackend()) + done := waitTerminal(t, m2, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, is.ErrorContains(errors.New(done.Error), "waiting on run container")) + waitJobSettled(t, m2, "backup", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) +} + +func TestRestoreFailsUnstartedRun(t *testing.T) { + for _, tc := range []struct { + doc string + withContainer bool + }{ + {doc: "no container was created"}, + {doc: "a container was left behind", withContainer: true}, + } { + t.Run(tc.doc, func(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + job, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + // Craft the crash shape directly in the store: a pending run + // whose start was never recorded, on a job stuck running. + run := &jobsv0.Run{ID: "orphan", JobID: job.ID, State: jobsv0.RunStatePending, CreatedAtNano: 1} + if tc.withContainer { + created, err := fake.ContainerCreate(t.Context(), ContainerCreateRequest{Name: "job-backup-orphan"}) + assert.NilError(t, err) + run.ContainerID = created.ID + } + assert.NilError(t, m1.store.CreateRun(t.Context(), run)) + stored, err := m1.store.Job(job.ID) + assert.NilError(t, err) + stored.State = jobsv0.JobStateRunning + assert.NilError(t, m1.store.UpdateJob(t.Context(), stored)) + + m2 := restartManager(t, m1, kv, fake) + done := waitStoreTerminal(t, m2, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, is.ErrorContains(errors.New(done.Error), "before the run start was recorded")) + waitJobSettled(t, m2, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + if tc.withContainer { + // The container is stopped before the job is released. + assert.Check(t, is.Contains(fake.stoppedContainers(), run.ContainerID)) + } + }) + } +} + +// TestRestoreUnstoppableUnstartedRunKeepsJobOccupied checks that a container +// left behind by an unstarted run, which may have started just as the daemon +// died, keeps its job occupied while it cannot be stopped. +func TestRestoreUnstoppableUnstartedRunKeepsJobOccupied(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + job, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + created, err := fake.ContainerCreate(t.Context(), ContainerCreateRequest{Name: "job-backup-orphan"}) + assert.NilError(t, err) + assert.NilError(t, fake.ContainerStart(t.Context(), created.ID)) + + run := &jobsv0.Run{ID: "orphan", JobID: job.ID, State: jobsv0.RunStatePending, CreatedAtNano: 1, ContainerID: created.ID} + assert.NilError(t, m1.store.CreateRun(t.Context(), run)) + stored, err := m1.store.Job(job.ID) + assert.NilError(t, err) + + stored.State = jobsv0.JobStateRunning + assert.NilError(t, m1.store.UpdateJob(t.Context(), stored)) + + fake.stopErr = errors.New("injected stop failure") + m2 := restartManager(t, m1, kv, fake) + _, err = m2.Run(t.Context(), job.ID, false) + assert.Check(t, cerrdefs.IsFailedPrecondition(err), "want failed-precondition, got %v", err) + + fake.exit(run.ContainerID, 0, nil) + done := waitStoreTerminal(t, m2, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, is.ErrorContains(errors.New(done.Error), "before the run start was recorded")) + waitJobSettled(t, m2, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) +} + +// TestRestoreCatchesUpFireMissedDuringRun checks that a schedule occurrence +// missed while the daemon was down still gets its catch-up run when the +// daemon stopped during a run. +func TestRestoreCatchesUpFireMissedDuringRun(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + clk := &fakeClock{t: jan15} + m1.now = clk.Now + job, _, err := m1.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + clk.Advance(3*time.Hour + time.Second) + m1.sched.tick(t.Context()) + run, err := m1.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning) + assert.NilError(t, err) + + // The run finishes and the next occurrence passes while the daemon is + // down. + assert.NilError(t, m1.Shutdown(shutdownCtx(t))) + fake.exit(run.ContainerID, 0, nil) + clk.Advance(48 * time.Hour) + + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + + m2 := NewManager(store, fake, testNameGenerator) + m2.now = clk.Now + + t.Cleanup(func() { assert.Check(t, m2.Shutdown(shutdownCtx(t))) }) + m2.Restore(t.Context()) + m2.Start(t.Context()) + + var catchUp *jobsv0.Run + + poll.WaitOn(t, func(poll.LogT) poll.Result { + latest, err := m2.store.LatestRun(job.ID) + if err != nil { + return poll.Error(err) + } + + if latest.ID == run.ID || latest.State != jobsv0.RunStateRunning { + return poll.Continue("no catch-up run yet") + } + + catchUp = latest + + return poll.Success() + }) + assert.Check(t, is.Equal(catchUp.Trigger.ScheduledAtNano, jan15.Add(27*time.Hour).UnixNano())) + fake.exit(catchUp.ContainerID, 0, nil) + waitTerminal(t, m2, job.ID, catchUp.ID) +} + +func TestRestoreIdlesStuckJob(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + _, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "backup", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m1, "backup", run.ID) + + // Crash shape: the run's terminal write landed, the job's idle write + // did not. + stored, err := m1.store.Job(run.JobID) + assert.NilError(t, err) + stored.State = jobsv0.JobStateRunning + assert.NilError(t, m1.store.UpdateJob(t.Context(), stored)) + + m2 := restartManager(t, m1, kv, fake) + job, err := m2.Inspect(t.Context(), "backup") + assert.NilError(t, err) + assert.Check(t, is.Equal(job.State, jobsv0.JobStateIdle)) +} + +func TestRestoreTimeouts(t *testing.T) { + t.Run("completed run keeps its outcome past the deadline", func(t *testing.T) { + t.Parallel() + + m1, fake, kv := newRestartableManager(t) + clk := &fakeClock{t: jan15} + m1.now = clk.Now + spec := manualSpec() + spec.TimeoutSeconds = 60 + job, _, err := m1.Create(t.Context(), "quick", spec) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + assert.NilError(t, m1.Shutdown(shutdownCtx(t))) + + // The container exits within its timeout, but startup happens later. + fake.exit(run.ContainerID, 0, nil) + clk.Advance(time.Hour) + + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + + m2 := NewManager(store, fake, testNameGenerator) + m2.now = clk.Now + + t.Cleanup(func() { assert.Check(t, m2.Shutdown(shutdownCtx(t))) }) + m2.Restore(t.Context()) + + done := waitTerminal(t, m2, job.ID, run.ID) + assert.Equal(t, done.State, jobsv0.RunStateSucceeded) + assert.Assert(t, done.ExitCode != nil) + assert.Equal(t, done.ExitCode.Value, int64(0)) + assert.Equal(t, len(fake.stoppedContainers()), 0) + }) + + t.Run("expired deadline is enforced on restore", func(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + clk := &fakeClock{t: jan15} + m1.now = clk.Now + spec := manualSpec() + spec.TimeoutSeconds = 60 + _, _, err := m1.Create(t.Context(), "slow", spec) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "slow", false) + assert.NilError(t, err) + + assert.NilError(t, m1.Shutdown(shutdownCtx(t))) + store, err := NewStore(t.Context(), kv) + assert.NilError(t, err) + m2 := NewManager(store, fake, testNameGenerator) + t.Cleanup(func() { assert.Check(t, m2.Shutdown(shutdownCtx(t))) }) + // The daemon comes back long past the run's deadline. + clk.Advance(time.Hour) + m2.now = clk.Now + m2.Restore(t.Context()) + + done := waitTerminal(t, m2, "slow", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateTimedOut)) + assert.Check(t, is.Contains(fake.stoppedContainers(), run.ContainerID)) + }) + + t.Run("remaining deadline is re-armed", func(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + spec := manualSpec() + spec.TimeoutSeconds = 3600 + _, _, err := m1.Create(t.Context(), "slow", spec) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), "slow", false) + assert.NilError(t, err) + + m2 := restartManager(t, m1, kv, fake) + m2.mu.Lock() + _, armed := m2.timers[run.ID] + m2.mu.Unlock() + assert.Check(t, armed, "the surviving deadline must be re-armed") + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m2, "slow", run.ID) + }) +} + +// TestRestoreQueuesCatchUpBehindReattachedRun checks that an occurrence +// missed while the daemon was down and a live-restored run kept going is +// queued behind that run under the queue concurrency policy. +func TestRestoreQueuesCatchUpBehindReattachedRun(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + job, _, err := m1.Create(t.Context(), "nightly", scheduleSpec(jobsv0.ConcurrencyQueue)) + assert.NilError(t, err) + run, err := m1.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + missed := time.Now().Add(-time.Hour).UnixNano() + stored, err := m1.store.Job(job.ID) + assert.NilError(t, err) + + stored.NextFireAtNano = missed + assert.NilError(t, m1.store.UpdateJob(t.Context(), stored)) + + m2 := restartManager(t, m1, kv, fake) + m2.Start(t.Context()) + poll.WaitOn(t, func(poll.LogT) poll.Result { + m2.mu.Lock() + defer m2.mu.Unlock() + + if m2.queued[job.ID] != nil { + return poll.Success() + } + + return poll.Continue("catch-up not queued yet") + }) + + // Until the deferred catch-up runs, the record keeps the missed + // occurrence, so another restart would catch it up again. + queued, err := m2.store.Job(job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(queued.NextFireAtNano, missed)) + + fake.exit(run.ContainerID, 0, nil) + + var catchUp *jobsv0.Run + + poll.WaitOn(t, func(poll.LogT) poll.Result { + latest, err := m2.store.LatestRun(job.ID) + if err != nil { + return poll.Error(err) + } + + if latest.ID == run.ID || latest.State != jobsv0.RunStateRunning { + return poll.Continue("no catch-up run yet") + } + + catchUp = latest + + return poll.Success() + }) + assert.Check(t, is.Equal(catchUp.Trigger.ScheduledAtNano, missed)) + fake.exit(catchUp.ContainerID, 0, nil) + waitTerminal(t, m2, job.ID, catchUp.ID) +} + +// TestRestoreRemovesUnrecordedContainer checks that a container created just +// before the daemon died, whose ID its run never recorded, is found by its +// name and removed. +func TestRestoreRemovesUnrecordedContainer(t *testing.T) { + m1, fake, kv := newRestartableManager(t) + job, _, err := m1.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + run := &jobsv0.Run{ID: "orphan", JobID: job.ID, State: jobsv0.RunStatePending, CreatedAtNano: 1} + assert.NilError(t, m1.store.CreateRun(t.Context(), run)) + created, err := fake.ContainerCreate(t.Context(), ContainerCreateRequest{Name: runContainerName(job, run)}) + assert.NilError(t, err) + + stored, err := m1.store.Job(job.ID) + assert.NilError(t, err) + + stored.State = jobsv0.JobStateRunning + assert.NilError(t, m1.store.UpdateJob(t.Context(), stored)) + + m2 := restartManager(t, m1, kv, fake) + done := waitStoreTerminal(t, m2, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, fake.container(t, created.ID).removed) +}