diff --git a/daemon/internal/jobs/manager.go b/daemon/internal/jobs/manager.go new file mode 100644 index 0000000000..c9e904168e --- /dev/null +++ b/daemon/internal/jobs/manager.go @@ -0,0 +1,722 @@ +package jobs + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "math" + "strings" + "sync" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/containerd/log" + "github.com/moby/locker" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/network" + "github.com/moby/moby/v2/daemon/internal/stringid" + "github.com/moby/moby/v2/daemon/names" + "github.com/moby/moby/v2/errdefs" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" +) + +// Backend is the container surface the manager drives runs with. +type Backend interface { + ContainerCreate(ctx context.Context, req ContainerCreateRequest) (container.CreateResponse, error) + ContainerStart(ctx context.Context, id string) error + // ContainerStop stops a container using its configured stop signal and + // timeout. + ContainerStop(ctx context.Context, id string) error + ContainerRm(ctx context.Context, id string) error + // ContainerWait resolves on the container's final exit, after its + // restart policy is exhausted. It fails only when the container does not + // exist; other failures to observe it are retried until ctx is done. + // The returned channel delivers exactly one status unless ctx is done + // first; it must never be closed without delivering it. + // If the container has already exited, the status is ready on return. + ContainerWait(ctx context.Context, id string) (<-chan StateStatus, error) +} + +// ContainerCreateRequest describes a run container to create. +type ContainerCreateRequest struct { + Name string + Config *container.Config + HostConfig *container.HostConfig + NetworkingConfig *network.NetworkingConfig +} + +// StateStatus reports a container's final exit, as delivered by +// [Backend.ContainerWait]. +type StateStatus interface { + ExitCode() int + Err() error +} + +// NameGenerator produces a default job name. Retry counts the prior +// generated names that collided, so implementations can vary their answer. +type NameGenerator func(ctx context.Context, retry int64) (string, error) + +// Reserved labels applied to every run container, correlating it with its +// job and run records across the container API. +const ( + // LabelJobID is the ID of the job that owns the container. + LabelJobID = "com.docker.job.id" + // LabelRunID is the ID of the run the container backs. + LabelRunID = "com.docker.job.run-id" +) + +// nameGenerationRetries bounds the attempts at generating an unused random +// job name, mirroring the daemon's own retry bound for container names. +const nameGenerationRetries = 6 + +// Manager owns the job lifecycle: registration with spec-hash idempotency, +// run execution through the single tryFireLocked chokepoint, and the queries +// the extension API exposes. Persistence is delegated to the Store, container +// work to the Backend. +type Manager struct { + store *Store + backend Backend + // generateName produces default job names for unnamed registrations. + generateName NameGenerator + // locks serializes state transitions per job ID: every fire, cancel, + // removal and run-completion transition takes the job's lock, so two + // triggers racing on the same job cannot both pass the concurrency + // policy. + locks *locker.Locker + // now is the manager's clock, injectable in tests. + now func() time.Time + // validateCron validates a cron expression at registration time; a + // field so tests can substitute it. + validateCron func(expr string) error + + mu sync.Mutex + // queued holds at most one deferred fire per job (the queue concurrency + // policy has depth one); fired when the current run completes. + queued map[string]*jobsv0.TriggerEvidence + // overrides forces a run's terminal state (cancelled or timed_out) + // ahead of its container exit; first writer wins. + overrides map[string]string + // timers holds the timeout timer of each in-flight run. + timers map[string]*time.Timer + // notify carries one-shot subscriptions per job ID, closed on every + // job or run transition; Wait re-checks its condition on each signal. + notify map[string][]chan struct{} + // background tracks every goroutine the manager spawns (run watchers, + // queued fires, container stops, the scheduler loop) so Shutdown can + // drain them. Goroutines spawned by a tracked goroutine register before + // their parent finishes, so the counter cannot reach zero while work is + // still being spawned. + background sync.WaitGroup + // sched fires schedule jobs; its loop runs between Start and Shutdown, + // but arming works from construction so jobs registered before Start + // are picked up when the loop begins. + sched *scheduler + started bool + stopped bool + // stop is closed by Shutdown so run watchers detach from containers + // that are not exiting — under live-restore the daemon shuts down while + // containers deliberately keep running, and waiting on them would stall + // (or, with an unbounded shutdown timeout, deadlock) the daemon's exit. + stop chan struct{} +} + +// NewManager builds a manager on top of a loaded store, generating default +// job names with generateName. The constructor does not reconcile runs +// orphaned by an unclean stop — callers run Restore before Start, as the +// extension's activation does. +func NewManager(store *Store, backend Backend, generateName NameGenerator) *Manager { + m := &Manager{ + store: store, + backend: backend, + generateName: generateName, + locks: locker.New(), + now: time.Now, + validateCron: func(expr string) error { + _, err := parseCron(expr) + return err + }, + queued: make(map[string]*jobsv0.TriggerEvidence), + overrides: make(map[string]string), + timers: make(map[string]*time.Timer), + notify: make(map[string][]chan struct{}), + stop: make(chan struct{}), + } + m.sched = newScheduler(m) + store.onRunsDropped = m.removeRunContainers + + return m +} + +// Start arms every registered schedule job and starts the scheduler loop. +// Occurrences missed while the daemon was down — the persisted next-fire +// time is in the past — follow each job's missed-fires policy: one fires a +// single catch-up run, skip drops them; either way the schedule then re-arms +// from the next occurrence, never replaying the backlog. +// A persisted occurrence that a reschedule moved past the upcoming one is +// kept. +func (m *Manager) Start(ctx context.Context) { + m.mu.Lock() + if m.started { + m.mu.Unlock() + return + } + m.started = true + m.mu.Unlock() + + now := m.now() + for _, job := range m.store.Jobs() { + if triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindSchedule || job.Paused { + continue + } + trigger := job.Spec.Trigger.Schedule + schedule, loc, err := parseScheduleTrigger(trigger) + if err != nil { + // Specs are validated at registration; only a store record from + // a buggy or newer daemon gets here. Leave the job disarmed + // rather than fail the whole startup. + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID}).Warn("leaving job with unparseable schedule disarmed") + // The record must not keep advertising an occurrence that will + // never fire. + m.persistNextFire(ctx, job.ID) + + continue + } + // A job still running has its run re-attached by reconciliation, + // which precedes Start: its catch-up goes through the concurrency + // policy like any other fire. + missed := job.NextFireAtNano + catchUp := missed > 0 && !time.Unix(0, missed).After(now) && + (trigger.MissedFires == "" || trigger.MissedFires == jobsv0.MissedFiresOne) && + !m.firedFor(job.ID, missed) + + if next, ok := schedule.next(now, loc); ok { + // A reschedule persisted the occurrence after the upcoming one; + // keep it rather than fire the occurrence it skipped. Only that + // exact occurrence is kept, so a clock stepped back cannot + // postpone the job any further. + if job.NextFireAtNano > next.UnixNano() { + if after, ok := schedule.next(next, loc); ok && after.UnixNano() == job.NextFireAtNano { + next = after + } + } + + m.sched.arm(job.ID, schedule, loc, next) + } + + if catchUp { + // The catch-up records the next occurrence once it fired, so a + // crash before then leaves it to the next startup. + m.background.Go(func() { m.fireScheduled(ctx, job.ID, time.Unix(0, missed)) }) + + continue + } + + m.persistNextFire(ctx, job.ID) + } + m.background.Go(func() { m.sched.run(ctx) }) +} + +// firedFor reports whether the job's latest run was fired for the scheduled +// occurrence: a fire is recorded before the next occurrence is, so a crash in +// between leaves a missed occurrence that already has its run. +func (m *Manager) firedFor(jobID string, scheduledAtNano int64) bool { + run, err := m.store.LatestRun(jobID) + return err == nil && run.Trigger != nil && run.Trigger.ScheduledAtNano == scheduledAtNano +} + +// parseScheduleTrigger resolves a validated schedule trigger into its parsed +// expression and timezone (empty means UTC). +func parseScheduleTrigger(trigger *jobsv0.ScheduleTrigger) (*cronSchedule, *time.Location, error) { + schedule, err := parseCron(trigger.Cron) + if err != nil { + return nil, nil, err + } + loc := time.UTC + if trigger.Timezone != "" { + if loc, err = time.LoadLocation(trigger.Timezone); err != nil { + return nil, nil, err + } + } + return schedule, loc, nil +} + +// Shutdown waits for in-flight run watchers to finish recording their +// outcomes, or for the context to expire. Callers must stop issuing API +// calls before invoking Shutdown: a call racing the drain could register +// new background work while the wait group is being awaited, which +// sync.WaitGroup forbids. When the context expires first, the internal +// waiter goroutine lingers until the background work eventually drains. +func (m *Manager) Shutdown(ctx context.Context) error { + m.mu.Lock() + if !m.stopped { + m.stopped = true + close(m.stop) + if m.started { + close(m.sched.stop) + } + // Disarm run timeouts: a deadline landing in the shutdown window + // would stop a container that live-restore deliberately keeps + // running. + for runID, timer := range m.timers { + timer.Stop() + delete(m.timers, runID) + } + } + m.mu.Unlock() + done := make(chan struct{}) + go func() { + m.background.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// Create registers a job without starting a run. It is idempotent on the +// canonical spec hash: re-submitting an identical name+spec returns the +// existing job with created=false, and a name collision with a different +// spec is a conflict carrying both hashes. +func (m *Manager) Create(ctx context.Context, name string, spec *jobsv0.JobSpec) (_ *jobsv0.Job, created bool, _ error) { + return m.create(ctx, name, spec, nil) +} + +// create is Create, calling onCreated with a newly registered job's ID +// before releasing the job's lock. +func (m *Manager) create(ctx context.Context, name string, spec *jobsv0.JobSpec, onCreated func(jobID string)) (_ *jobsv0.Job, created bool, _ error) { + if err := validateJobName(name); err != nil { + return nil, false, errdefs.InvalidParameter(err) + } + decoded, err := m.validateSpec(spec) + if err != nil { + return nil, false, err + } + hash, err := specHash(spec, decoded) + if err != nil { + return nil, false, err + } + + // Registering a schedule job arms its trigger; the next occurrence is + // computed up front so the returned job already advertises it, and an + // expression with no future occurrence is rejected rather than silently + // registering a job that would never fire. + var ( + schedule *cronSchedule + loc *time.Location + next time.Time + ) + if spec.Trigger != nil && spec.Trigger.Schedule != nil { + if schedule, loc, err = parseScheduleTrigger(spec.Trigger.Schedule); err != nil { + return nil, false, errdefs.InvalidParameter(err) // unreachable after validateSpec, kept for defense + } + var ok bool + if next, ok = schedule.next(m.now(), loc); !ok { + return nil, false, errdefs.InvalidParameter(fmt.Errorf("cron expression %q has no future occurrence", spec.Trigger.Schedule.Cron)) + } + } + + job := &jobsv0.Job{ + ID: stringid.GenerateRandomID(), + Name: name, + Spec: spec, + SpecHash: hash, + State: jobsv0.JobStateIdle, + CreatedAtNano: m.now().UnixNano(), + UpdatedAtNano: m.now().UnixNano(), + } + if schedule != nil { + job.NextFireAtNano = next.UnixNano() + } + // Hold the new job's lock until its trigger is armed: a removal by name + // landing between the store write and arming would otherwise leave the + // trigger armed for a deleted job. + m.locks.Lock(job.ID) + defer m.locks.Unlock(job.ID) + + if err := m.store.CreateJob(ctx, job); err != nil { + // A name conflict is idempotent when the stored spec matches; + // re-read under the store's authority rather than trusting a + // pre-flight lookup, so concurrent creates converge. + if cerrdefs.IsConflict(err) { + existing, lookupErr := m.store.JobByName(name) + if lookupErr != nil { + return nil, false, err + } + + existingHash := currentSpecHash(existing) + if existingHash == hash { + return m.composeJob(existing), false, nil + } + + return nil, false, cerrdefs.ErrAlreadyExists.WithMessage(fmt.Sprintf("job name %s is already in use by a different spec (stored %s, submitted %s)", name, existingHash, hash)) + } + return nil, false, err + } + if schedule != nil { + m.sched.arm(job.ID, schedule, loc, next) + } + + if onCreated != nil { + onCreated(job.ID) + } + + return m.composeJob(job), true, nil +} + +// CreateAndRun atomically resolves or registers a manual job and fires a +// run, implementing the name/spec resolution matrix of the API contract. +func (m *Manager) CreateAndRun(ctx context.Context, name string, spec *jobsv0.JobSpec) (_ *jobsv0.Job, _ *jobsv0.Run, created bool, _ error) { + if name == "" && spec == nil { + return nil, nil, false, errdefs.InvalidParameter(errors.New("at least one of name and spec is required")) + } + if spec != nil && triggerKind(spec.Trigger) != jobsv0.TriggerKindManual { + return nil, nil, false, errdefs.InvalidParameter(errors.New("create-and-run serves manual jobs only; register schedule jobs with create")) + } + + var ( + job *jobsv0.Job + run *jobsv0.Run + runErr error + ) + // A newly registered job's first run fires before the job's lock is + // released, so nothing can run or remove the job in between. + fire := func(jobID string) { run, runErr = m.fireManualLocked(ctx, jobID) } + + switch { + case name == "": + // No name: always a fresh job under a generated name. + var err error + if job, err = m.createNamed(ctx, spec, fire); err != nil { + return nil, nil, false, err + } + created = true + case spec == nil: + // The request field is a name, strictly: the create branches treat + // it as one, so resolving IDs here would be an observable asymmetry. + existing, err := m.store.JobByName(name) + if err != nil { + return nil, nil, false, err + } + if triggerKind(existing.Spec.Trigger) != jobsv0.TriggerKindManual { + return nil, nil, false, errdefs.InvalidParameter(fmt.Errorf("job %s is not a manual job", name)) + } + job = existing + default: + var err error + if job, created, err = m.create(ctx, name, spec, fire); err != nil { + return nil, nil, false, err + } + } + + if run == nil && runErr == nil { + run, runErr = m.fireManual(ctx, job.ID) + } + + if runErr != nil { + return nil, nil, false, runErr + } + + job, err := m.Inspect(ctx, job.ID) + if err != nil { + return nil, nil, false, err + } + return job, run, created, nil +} + +// createNamed registers spec under a generated name, retrying on collisions +// like the daemon does for container names. A random name landing on an +// existing job with an identical spec also counts as a collision: adopting +// that job would silently hand out foreign history. +func (m *Manager) createNamed(ctx context.Context, spec *jobsv0.JobSpec, onCreated func(jobID string)) (*jobsv0.Job, error) { + var lastErr error + for i := range nameGenerationRetries { + name, err := m.generateName(ctx, int64(i)) + if err != nil { + return nil, fmt.Errorf("generating a job name: %w", err) + } + + job, created, err := m.create(ctx, name, spec, onCreated) + switch { + case err == nil && created: + return job, nil + case err == nil: + lastErr = fmt.Errorf("generated name %s is already in use", name) + case cerrdefs.IsAlreadyExists(err) || cerrdefs.IsConflict(err): + lastErr = err + default: + return nil, err + } + } + return nil, fmt.Errorf("generating a unique job name: %w", lastErr) +} + +// resolveJob resolves a job reference, trying IDs first and names second, +// like the container API does. +func (m *Manager) resolveJob(ref string) (*jobsv0.Job, error) { + if ref == "" { + return nil, errdefs.InvalidParameter(errors.New("job reference must not be empty")) + } + job, err := m.store.Job(ref) + if err == nil { + return job, nil + } + return m.store.JobByName(ref) +} + +// composeJob attaches the derived LatestRun to a stored job. A running job +// advertises no next fire. Its record keeps the scheduler's cursor, which +// Start needs to catch up on fires missed while the daemon was down. +func (m *Manager) composeJob(job *jobsv0.Job) *jobsv0.Job { + if latest, err := m.store.LatestRun(job.ID); err == nil { + job.LatestRun = latest + } + + if job.State == jobsv0.JobStateRunning { + job.NextFireAtNano = 0 + } + + return job +} + +// maxContainerSpecSize bounds the raw container definition, the dominant +// term of the job record's size against the storage point's value limit +// (storagekvv0.MaxValueSize, 1 MiB): the spec is base64-encoded inside the +// record's JSON (4/3 growth). Rejecting it here names the offending field; +// the store caps the whole encoded record, covering the remaining +// user-controlled fields. +const maxContainerSpecSize = 512 * 1024 + +// validateSpec checks everything the contract rejects at registration time +// and returns the decoded container definition. +func (m *Manager) validateSpec(spec *jobsv0.JobSpec) (*container.CreateRequest, error) { + if spec == nil { + return nil, errdefs.InvalidParameter(errors.New("a job spec is required")) + } + if len(spec.ContainerSpec) > maxContainerSpecSize { + return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: definition exceeds the maximum size of %d bytes", maxContainerSpecSize)) + } + if err := validateTrigger(spec.Trigger, m.validateCron); err != nil { + return nil, errdefs.InvalidParameter(err) + } + if spec.TimeoutSeconds < 0 { + return nil, errdefs.InvalidParameter(errors.New("timeout must not be negative")) + } + + if spec.TimeoutSeconds > int64(math.MaxInt64/time.Second) { + return nil, errdefs.InvalidParameter(fmt.Errorf("timeout must not exceed %d seconds", int64(math.MaxInt64/time.Second))) + } + + for label := range spec.Labels { + if strings.HasPrefix(label, "com.docker.job.") { + return nil, errdefs.InvalidParameter(fmt.Errorf("label %s is reserved", label)) + } + } + + decoded, err := decodeContainerSpec(spec.ContainerSpec, true) + if err != nil { + return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: %w", err)) + } + if decoded.Image == "" { + return nil, errdefs.InvalidParameter(errors.New("container spec: an image is required")) + } + for label := range decoded.Labels { + if strings.HasPrefix(label, "com.docker.job.") { + return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: label %s is reserved", label)) + } + } + // Cancels, timeouts and reconciliation wait for the container to stop; + // a stop that never escalates to a kill would hold the job forever. + if decoded.StopTimeout != nil && *decoded.StopTimeout < 0 { + return nil, errdefs.InvalidParameter(errors.New("container spec: a negative StopTimeout is not supported for jobs; jobs must be stoppable")) + } + + if hc := decoded.HostConfig; hc != nil { + // AutoRemove races the exit-code capture the run record depends on; + // RemoveOnSuccess/RemoveOnFailure are the job-level equivalents. + if hc.AutoRemove { + return nil, errdefs.InvalidParameter(errors.New("container spec: AutoRemove is not supported for jobs; use RemoveOnSuccess or RemoveOnFailure")) + } + // A job is expected to terminate; keep-alive restart policies would + // make the run never reach an outcome. The same goes for on-failure + // without a retry cap, which Docker treats as retry-forever. + switch hc.RestartPolicy.Name { + case "", container.RestartPolicyDisabled: + case container.RestartPolicyOnFailure: + if hc.RestartPolicy.MaximumRetryCount <= 0 { + return nil, errdefs.InvalidParameter(errors.New("container spec: on-failure requires a maximum retry count for jobs; unbounded retries would keep the run from ever completing")) + } + default: + return nil, errdefs.InvalidParameter(fmt.Errorf("container spec: restart policy %s is not supported for jobs; use no or on-failure", hc.RestartPolicy.Name)) + } + } + return decoded, nil +} + +// validateTrigger enforces the trigger sum-type rules of the API contract. +func validateTrigger(trigger *jobsv0.Trigger, validateCron func(string) error) error { + if trigger == nil { + return nil // the manual shorthand + } + switch { + case trigger.Manual && trigger.Schedule != nil: + return errors.New("trigger declares more than one kind") + case trigger.Schedule != nil: + sched := trigger.Schedule + if err := validateCron(sched.Cron); err != nil { + return fmt.Errorf("schedule: %w", err) + } + if sched.Timezone != "" { + // Local resolves to the daemon host's timezone, which a spec must + // not depend on. + if sched.Timezone == "Local" { + return errors.New("schedule: timezone Local is not supported; use an IANA timezone name") + } + + if _, err := time.LoadLocation(sched.Timezone); err != nil { + return fmt.Errorf("schedule: unknown timezone %s", sched.Timezone) + } + } + switch sched.Concurrency { + case "", jobsv0.ConcurrencyForbid, jobsv0.ConcurrencyQueue: + default: + return fmt.Errorf("schedule: unsupported concurrency policy %s", sched.Concurrency) + } + switch sched.MissedFires { + case "", jobsv0.MissedFiresOne, jobsv0.MissedFiresSkip: + default: + return fmt.Errorf("schedule: unsupported missed-fires policy %s", sched.MissedFires) + } + return nil + case trigger.Manual: + return nil + default: + // Deliberate forward-compatibility failure: a trigger kind from a + // newer contract version reaches an older daemon as an empty + // message, because protobuf drops unknown fields. Registering it + // as anything would silently disarm it. + return errors.New("trigger declares no known kind; the daemon may be too old for this trigger type") + } +} + +// validateJobName applies the container-name character rules to job names. +func validateJobName(name string) error { + if name == "" { + return errors.New("a job name is required") + } + if !names.RestrictedNamePattern.MatchString(name) { + return fmt.Errorf("invalid job name %q, only %s are allowed", name, names.RestrictedNameChars) + } + return nil +} + +// decodeContainerSpec decodes the JSON container definition, in the exact +// format of the container-create API request body. +// Registration passes strict to reject unknown fields, so a typo fails +// instead of silently altering the container. +// Stored specs decode leniently: a field accepted by the daemon that +// registered the job may have been removed from the API types since. +func decodeContainerSpec(raw []byte, strict bool) (*container.CreateRequest, error) { + if len(raw) == 0 { + return nil, errors.New("a container definition is required") + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + if strict { + decoder.DisallowUnknownFields() + } + + req := &container.CreateRequest{} + if err := decoder.Decode(req); err != nil { + return nil, err + } + if decoder.More() { + return nil, errors.New("unexpected trailing data after the container definition") + } + if req.Config == nil { + req.Config = &container.Config{} + } + return req, nil +} + +// canonicalSpec is the hashing form of a job spec: daemon defaults applied, +// container definition re-encoded from its decoded form so JSON formatting +// differences (whitespace, key order — encoding/json sorts map keys) do not +// change a job's identity. Structural presence still matters: an explicit +// empty HostConfig decodes to a non-nil pointer and hashes differently from +// an absent one, a conservative trade-off over normalizing every substruct. +type canonicalSpec struct { + ContainerSpec *container.CreateRequest + TriggerKind string + Schedule *jobsv0.ScheduleTrigger `json:",omitempty"` + Labels map[string]string `json:",omitempty"` + TimeoutSeconds int64 + RemoveOnSuccess bool + RemoveOnFailure bool + RunHistoryLimit uint32 +} + +// specHash computes the canonical identity hash of a validated spec. +func specHash(spec *jobsv0.JobSpec, decoded *container.CreateRequest) (string, error) { + canonical := canonicalSpec{ + ContainerSpec: decoded, + TriggerKind: triggerKind(spec.Trigger), + Labels: spec.Labels, + TimeoutSeconds: spec.TimeoutSeconds, + RemoveOnSuccess: spec.RemoveOnSuccess, + RemoveOnFailure: spec.RemoveOnFailure, + RunHistoryLimit: spec.RunHistoryLimit, + } + if canonical.RunHistoryLimit == 0 { + canonical.RunHistoryLimit = DefaultRunHistoryLimit + } + if spec.Trigger != nil && spec.Trigger.Schedule != nil { + sched := *spec.Trigger.Schedule + if sched.Timezone == "" { + sched.Timezone = "UTC" + } + if sched.Concurrency == "" { + sched.Concurrency = jobsv0.ConcurrencyForbid + } + if sched.MissedFires == "" { + sched.MissedFires = jobsv0.MissedFiresOne + } + canonical.Schedule = &sched + } + data, err := json.Marshal(canonical) + if err != nil { + return "", fmt.Errorf("encoding canonical spec: %w", err) + } + return fmt.Sprintf("sha256:%x", sha256.Sum256(data)), nil +} + +// currentSpecHash computes a stored job's spec hash with this daemon's +// canonicalization. +// The persisted hash comes from the daemon that registered the job, and the +// canonical encoding changes whenever the container API types gain fields. +func currentSpecHash(job *jobsv0.Job) string { + decoded, err := decodeContainerSpec(job.Spec.ContainerSpec, false) + if err != nil { + return job.SpecHash + } + + hash, err := specHash(job.Spec, decoded) + if err != nil { + return job.SpecHash + } + + return hash +} + +// triggerKind names a validated trigger's kind; a nil trigger is the manual +// shorthand. +func triggerKind(trigger *jobsv0.Trigger) string { + if trigger != nil && trigger.Schedule != nil { + return jobsv0.TriggerKindSchedule + } + return jobsv0.TriggerKindManual +} diff --git a/daemon/internal/jobs/manager_test.go b/daemon/internal/jobs/manager_test.go new file mode 100644 index 0000000000..be7eb92d95 --- /dev/null +++ b/daemon/internal/jobs/manager_test.go @@ -0,0 +1,1378 @@ +package jobs + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "math" + "slices" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/container" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" + "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" + "gotest.tools/v3/poll" +) + +// fakeStatus is a container exit observed by a fake wait. +type fakeStatus struct { + code int + err error +} + +func (s fakeStatus) ExitCode() int { return s.code } +func (s fakeStatus) Err() error { return s.err } + +// fakeContainer is one container tracked by the fake backend. +type fakeContainer struct { + config ContainerCreateRequest + started bool + removed bool + exited *fakeStatus + waiters []chan StateStatus +} + +// fakeBackend implements Backend in memory. Tests drive container exits +// explicitly with exit(), and stops behave like a kill: the container exits +// with code 137. +type fakeBackend struct { + mu sync.Mutex + seq int + containers map[string]*fakeContainer + createErr error + startErr error + stopErr error + waitErr error + stopped []string +} + +func newFakeBackend() *fakeBackend { + return &fakeBackend{containers: make(map[string]*fakeContainer)} +} + +func (b *fakeBackend) ContainerCreate(ctx context.Context, config ContainerCreateRequest) (container.CreateResponse, error) { + // Like the Engine API client, fail requests whose context is done. + if err := ctx.Err(); err != nil { + return container.CreateResponse{}, err + } + + b.mu.Lock() + defer b.mu.Unlock() + if b.createErr != nil { + return container.CreateResponse{}, b.createErr + } + b.seq++ + id := fmt.Sprintf("ctr%d", b.seq) + b.containers[id] = &fakeContainer{config: config} + return container.CreateResponse{ID: id}, nil +} + +func (b *fakeBackend) ContainerStart(ctx context.Context, name string) error { + if err := ctx.Err(); err != nil { + return err + } + + b.mu.Lock() + defer b.mu.Unlock() + if b.startErr != nil { + return b.startErr + } + ctr, exists := b.containers[name] + if !exists { + return errors.New("no such container") + } + ctr.started = true + return nil +} + +func (b *fakeBackend) ContainerStop(_ context.Context, name string) error { + b.mu.Lock() + b.stopped = append(b.stopped, name) + err := b.stopErr + b.mu.Unlock() + + if err != nil { + return err + } + + b.exit(name, 137, nil) + return nil +} + +func (b *fakeBackend) ContainerRm(_ context.Context, name string) error { + b.mu.Lock() + defer b.mu.Unlock() + // Like the Engine API, accept a container name as well as an ID. + for id, ctr := range b.containers { + if id == name || ctr.config.Name == name { + ctr.removed = true + return nil + } + } + + return errors.New("no such container") +} + +func (b *fakeBackend) ContainerWait(_ context.Context, name string) (<-chan StateStatus, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.waitErr != nil { + return nil, b.waitErr + } + ctr, exists := b.containers[name] + if !exists { + return nil, errors.New("no such container") + } + waitC := make(chan StateStatus, 1) + if ctr.exited != nil { + waitC <- *ctr.exited + return waitC, nil + } + ctr.waiters = append(ctr.waiters, waitC) + return waitC, nil +} + +// exit delivers a container's final exit to its waiters; later waits observe +// it immediately. A second exit for the same container is ignored. +func (b *fakeBackend) exit(name string, code int, err error) { + b.mu.Lock() + defer b.mu.Unlock() + ctr, exists := b.containers[name] + if !exists || ctr.exited != nil { + return + } + ctr.exited = &fakeStatus{code: code, err: err} + for _, waitC := range ctr.waiters { + waitC <- *ctr.exited + } + ctr.waiters = nil +} + +func (b *fakeBackend) container(t *testing.T, id string) fakeContainer { + t.Helper() + b.mu.Lock() + defer b.mu.Unlock() + ctr, exists := b.containers[id] + assert.Assert(t, exists, "no container %s in fake backend", id) + return *ctr +} + +func (b *fakeBackend) createCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.containers) +} + +// stoppedContainers returns a copy of the stop log, for assertions that are +// not ordered after the stop by a happens-before chain. +func (b *fakeBackend) stoppedContainers() []string { + b.mu.Lock() + defer b.mu.Unlock() + return slices.Clone(b.stopped) +} + +// testNameGenerator stands in for the daemon's name generation: +// deterministic, distinct per retry. +func testNameGenerator(_ context.Context, retry int64) (string, error) { + return fmt.Sprintf("generated-job-%d", retry), nil +} + +func newTestManager(t *testing.T) (*Manager, *fakeBackend) { + t.Helper() + store, _ := newTestStore(t) + fake := newFakeBackend() + m := NewManager(store, fake, testNameGenerator) + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.WithoutCancel(t.Context()), 10*time.Second) + defer cancel() + assert.Check(t, m.Shutdown(ctx)) + }) + return m, fake +} + +// tryFire fires through the tryFireLocked chokepoint, taking the job's lock. +func (m *Manager) tryFire(ctx context.Context, jobID string, evidence *jobsv0.TriggerEvidence) (*jobsv0.Run, error) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + return m.tryFireLocked(ctx, jobID, evidence) +} + +func manualSpec() *jobsv0.JobSpec { + return &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","Cmd":["/backup.sh"]}`)} +} + +func scheduleSpec(concurrency string) *jobsv0.JobSpec { + return &jobsv0.JobSpec{ + ContainerSpec: []byte(`{"Image":"busybox"}`), + Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{ + Cron: "0 3 * * *", + Concurrency: concurrency, + }}, + } +} + +// waitTerminal blocks until the run reaches a terminal state. +func waitTerminal(t *testing.T, m *Manager, jobRef, runRef string) *jobsv0.Run { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + run, err := m.Wait(ctx, jobRef, runRef, jobsv0.WaitConditionTerminal) + assert.NilError(t, err) + return run +} + +// waitJobSettled polls Inspect until settled reports true, and returns the +// settled job. Wait's contract covers the run record only: the completion +// transition writes the run's terminal state first (a crash between the two +// must leave the run resolved), then returns the job record to idle — so +// assertions on the job record after a wait must poll that second write. +func waitJobSettled(t *testing.T, m *Manager, jobRef string, settled func(*jobsv0.Job) bool) *jobsv0.Job { + t.Helper() + var job *jobsv0.Job + poll.WaitOn(t, func(poll.LogT) poll.Result { + var err error + job, err = m.Inspect(t.Context(), jobRef) + if err != nil { + return poll.Error(err) + } + if settled(job) { + return poll.Success() + } + return poll.Continue("job not settled yet (state %s, next fire %d)", job.State, job.NextFireAtNano) + }) + return job +} + +// waitStoreTerminal observes a run in the store until it turns terminal, +// for runs whose job record is gone (tombstone history) and which the +// API-level Wait therefore correctly refuses to resolve. +func waitStoreTerminal(t *testing.T, m *Manager, jobID, runID string) *jobsv0.Run { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + for { + run, err := m.store.Run(jobID, runID) + assert.NilError(t, err) + if isTerminalRunState(run.State) { + return run + } + sub := m.subscribe(jobID) + select { + case <-ctx.Done(): + t.Fatal("run never turned terminal") + case <-sub: + } + } +} + +func TestManagerCreateIdempotency(t *testing.T) { + m, fake := newTestManager(t) + + job, created, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + assert.Check(t, created) + + // Same spec, different JSON formatting: same identity, no-op. + respaced := &jobsv0.JobSpec{ContainerSpec: []byte(`{ "Cmd": ["/backup.sh"], "Image": "busybox" }`)} + again, created, err := m.Create(t.Context(), "backup", respaced) + assert.NilError(t, err) + assert.Check(t, !created) + assert.Check(t, is.Equal(again.ID, job.ID)) + assert.Check(t, is.Equal(again.SpecHash, job.SpecHash)) + + // Different spec under the same name: already-exists carrying both + // hashes, so a gRPC adapter can map it to codes.AlreadyExists. + other := &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"alpine"}`)} + _, _, err = m.Create(t.Context(), "backup", other) + assert.Check(t, cerrdefs.IsAlreadyExists(err), "want already-exists, got %v", err) + assert.Check(t, is.ErrorContains(err, job.SpecHash)) + + // Registration never creates containers, whatever the trigger. + _, _, err = m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + assert.Check(t, is.Equal(fake.createCount(), 0)) +} + +func TestManagerSpecHashCanonicalization(t *testing.T) { + hash := func(t *testing.T, spec *jobsv0.JobSpec) string { + t.Helper() + m, _ := newTestManager(t) + decoded, err := m.validateSpec(spec) + assert.NilError(t, err) + h, err := specHash(spec, decoded) + assert.NilError(t, err) + return h + } + + base := scheduleSpec("") + assert.Check(t, is.Equal(hash(t, base), hash(t, scheduleSpec(jobsv0.ConcurrencyForbid))), + "an empty concurrency policy must hash like the explicit default") + + utc := scheduleSpec("") + utc.Trigger.Schedule.Timezone = "UTC" + assert.Check(t, is.Equal(hash(t, base), hash(t, utc)), + "an empty timezone must hash like explicit UTC") + + capped := manualSpec() + capped.RunHistoryLimit = DefaultRunHistoryLimit + assert.Check(t, is.Equal(hash(t, manualSpec()), hash(t, capped)), + "a zero history limit must hash like the explicit default") + + reordered := manualSpec() + reordered.ContainerSpec = []byte(`{"Cmd":["/backup.sh"],"Image":"busybox"}`) + assert.Check(t, is.Equal(hash(t, manualSpec()), hash(t, reordered)), + "JSON key order must not change a job's identity") + + swapped := manualSpec() + swapped.ContainerSpec = []byte(`{"Image":"busybox","Cmd":["/other.sh"]}`) + assert.Check(t, hash(t, manualSpec()) != hash(t, swapped), "Cmd is order-sensitive identity") +} + +// TestManagerJobsFromOtherDaemonVersions covers jobs registered by another +// daemon version: their persisted hash came from that daemon's +// canonicalization, and their container spec may carry a field this daemon's +// API types no longer have. +func TestManagerJobsFromOtherDaemonVersions(t *testing.T) { + store, _ := newTestStore(t) + replayed := makeJob("job1", "replayed") + assert.NilError(t, store.CreateJob(t.Context(), replayed)) + + legacy := makeJob("job2", "legacy") + legacy.Spec.ContainerSpec = []byte(`{"Image":"busybox","RemovedField":true}`) + assert.NilError(t, store.CreateJob(t.Context(), legacy)) + + fake := newFakeBackend() + m := NewManager(store, fake, testNameGenerator) + + t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) }) + + job, created, err := m.Create(t.Context(), "replayed", replayed.Spec) + assert.NilError(t, err, "replaying a stored spec must stay idempotent") + assert.Check(t, !created) + assert.Check(t, is.Equal(job.ID, "job1")) + + run, err := m.Run(t.Context(), "legacy", false) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.State, jobsv0.RunStateRunning), "run error: %q", run.Error) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, "legacy", run.ID) +} + +func TestManagerValidation(t *testing.T) { + m, _ := newTestManager(t) + for _, tc := range []struct { + doc string + name string + spec *jobsv0.JobSpec + }{ + {doc: "empty name", name: "", spec: manualSpec()}, + {doc: "invalid name", name: "no spaces allowed", spec: manualSpec()}, + {doc: "nil spec", name: "j", spec: nil}, + {doc: "empty container spec", name: "j", spec: &jobsv0.JobSpec{}}, + {doc: "no image", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{}`)}}, + {doc: "unknown container field", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","Imag":"typo"}`)}}, + {doc: "auto remove", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","HostConfig":{"AutoRemove":true}}`)}}, + {doc: "restart always", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","HostConfig":{"RestartPolicy":{"Name":"always"}}}`)}}, + {doc: "restart unless-stopped", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","HostConfig":{"RestartPolicy":{"Name":"unless-stopped"}}}`)}}, + {doc: "reserved job label", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Labels: map[string]string{"com.docker.job.id": "x"}}}, + {doc: "reserved container label", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","Labels":{"com.docker.job.run-id":"x"}}`)}}, + {doc: "negative timeout", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), TimeoutSeconds: -1}}, + {doc: "timeout overflowing a duration", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), TimeoutSeconds: math.MaxInt64/int64(time.Second) + 1}}, + {doc: "infinite stop timeout", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox","StopTimeout":-1}`)}}, + {doc: "oversized container spec", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: append([]byte(`{"Image":"busybox","Env":["`), append(bytes.Repeat([]byte{'x'}, maxContainerSpecSize), []byte(`"]}`)...)...)}}, + {doc: "empty trigger declares no kind", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{}}}, + {doc: "two trigger kinds", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Manual: true, Schedule: &jobsv0.ScheduleTrigger{Cron: "* * * * *"}}}}, + {doc: "empty cron", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{Cron: " "}}}}, + {doc: "unknown timezone", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{Cron: "* * * * *", Timezone: "Mars/Olympus"}}}}, + {doc: "host-local timezone", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{Cron: "* * * * *", Timezone: "Local"}}}}, + {doc: "unknown concurrency", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{Cron: "* * * * *", Concurrency: "replace"}}}}, + {doc: "unknown missed fires", name: "j", spec: &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"busybox"}`), Trigger: &jobsv0.Trigger{Schedule: &jobsv0.ScheduleTrigger{Cron: "* * * * *", MissedFires: "all"}}}}, + } { + t.Run(tc.doc, func(t *testing.T) { + _, _, err := m.Create(t.Context(), tc.name, tc.spec) + assert.Check(t, cerrdefs.IsInvalidArgument(err), "want invalid-argument, got %v", err) + }) + } +} + +func TestManagerRunLifecycle(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.State, jobsv0.RunStateRunning)) + assert.Check(t, is.Equal(run.Iteration, uint64(1))) + + // The run container carries the correlation labels and the derived + // name; the run-ID suffix keeps names unique across job generations. + ctr := fake.container(t, run.ContainerID) + assert.Check(t, strings.HasPrefix(ctr.config.Name, "job-backup-1-"), "container name %q", ctr.config.Name) + assert.Check(t, is.Equal(ctr.config.Config.Labels[LabelJobID], job.ID)) + assert.Check(t, is.Equal(ctr.config.Config.Labels[LabelRunID], run.ID)) + assert.Check(t, ctr.started) + + // While running, the job reports running and a second run is refused. + inspected, err := m.Inspect(t.Context(), "backup") + assert.NilError(t, err) + assert.Check(t, is.Equal(inspected.State, jobsv0.JobStateRunning)) + _, err = m.Run(t.Context(), "backup", false) + assert.Check(t, cerrdefs.IsFailedPrecondition(err), "want failed-precondition, got %v", err) + + fake.exit(run.ContainerID, 0, nil) + done := waitTerminal(t, m, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded)) + assert.Assert(t, done.ExitCode != nil) + assert.Check(t, is.Equal(done.ExitCode.Value, int64(0))) + + inspected = waitJobSettled(t, m, "backup", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + assert.Check(t, is.Equal(inspected.LatestRun.ID, run.ID)) + + // The next run continues the iteration sequence. + second, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + assert.Check(t, is.Equal(second.Iteration, uint64(2))) + fake.exit(second.ContainerID, 3, nil) + done = waitTerminal(t, m, "backup", second.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, is.Equal(done.ExitCode.Value, int64(3))) +} + +func TestManagerRunRecordsCreateFailure(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + fake.createErr = errors.New("no such image: busybox") + + // The failure is an outcome, not an error: the run record exists (it is + // written before the container is created) and carries the failure. + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.State, jobsv0.RunStateFailed)) + assert.Check(t, is.ErrorContains(errors.New(run.Error), "no such image")) + assert.Check(t, is.Nil(run.ExitCode)) + assert.Check(t, is.Equal(run.ContainerID, "")) + + job, err := m.Inspect(t.Context(), "backup") + assert.NilError(t, err) + assert.Check(t, is.Equal(job.State, jobsv0.JobStateIdle)) +} + +// A caller going away mid-fire must not fail the run: the daemon completes a +// container start regardless, and a failed run would release the job while +// its container keeps running unwatched. +func TestManagerRunOutlivesCancelledCaller(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + run, err := m.Run(ctx, "backup", false) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.State, jobsv0.RunStateRunning), "run error: %q", run.Error) + + fake.exit(run.ContainerID, 0, nil) + done := waitTerminal(t, m, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded)) +} + +type failRunStartKV struct { + KV +} + +func (k failRunStartKV) Update(ctx context.Context, key string, value []byte) error { + if strings.HasPrefix(key, runKeyPrefix) { + var record runRecord + if err := json.Unmarshal(value, &record); err != nil { + return err + } + + if record.Run.State == jobsv0.RunStateRunning { + return errors.New("injected start-record write failure") + } + } + + return k.KV.Update(ctx, key, value) +} + +func TestManagerRunStartRecordFailure(t *testing.T) { + for _, stopFails := range []bool{false, true} { + t.Run(fmt.Sprintf("stop failure=%t", stopFails), func(t *testing.T) { + t.Parallel() + + m, fake := newTestManager(t) + spec := manualSpec() + spec.TimeoutSeconds = 3600 + job, _, err := m.Create(t.Context(), "backup", spec) + assert.NilError(t, err) + + m.store.kv = failRunStartKV{KV: m.store.kv} + + if stopFails { + fake.stopErr = errors.New("injected stop failure") + } + + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + poll.WaitOn(t, func(poll.LogT) poll.Result { + if slices.Contains(fake.stoppedContainers(), run.ContainerID) { + return poll.Success() + } + + return poll.Continue("waiting for container stop") + }) + + if stopFails { + _, err := m.Run(t.Context(), job.ID, false) + assert.Check(t, cerrdefs.IsFailedPrecondition(err)) + m.mu.Lock() + _, armed := m.timers[run.ID] + m.mu.Unlock() + assert.Check(t, armed) + fake.exit(run.ContainerID, 0, nil) + } + + done := waitTerminal(t, m, job.ID, run.ID) + assert.Equal(t, done.State, jobsv0.RunStateFailed) + assert.Check(t, is.Contains(done.Error, "injected start-record write failure")) + assert.Assert(t, fake.container(t, run.ContainerID).exited != nil) + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + }) + } +} + +func TestManagerConcurrencyPolicies(t *testing.T) { + t.Run("forbid drops the fire", func(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + first, err := m.tryFire(t.Context(), job.ID, &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule}) + assert.NilError(t, err) + + _, err = m.tryFire(t.Context(), job.ID, &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule}) + assert.Check(t, cerrdefs.IsFailedPrecondition(err), "want failed-precondition, got %v", err) + fake.exit(first.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, first.ID) + }) + + t.Run("queue defers one fire", func(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(jobsv0.ConcurrencyQueue)) + assert.NilError(t, err) + first, err := m.tryFire(t.Context(), job.ID, &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule}) + assert.NilError(t, err) + + _, err = m.tryFire(t.Context(), job.ID, &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule}) + assert.Check(t, is.ErrorIs(err, errFireQueued)) + _, err = m.tryFire(t.Context(), job.ID, &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule}) + assert.Check(t, is.ErrorIs(err, errFireDropped)) + + // Completing the first run fires the queued one automatically. Wait + // for the second run to be RUNNING before making it exit: its record + // exists before its container does, so exiting on sight of the + // record alone would race container creation. + fake.exit(first.ContainerID, 0, nil) + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + for { + second, err := m.InspectRun(ctx, job.ID, "latest") + assert.NilError(t, err) + if second.Iteration == 2 && second.State == jobsv0.RunStateRunning { + fake.exit(second.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, second.ID) + return + } + sub := m.subscribe(job.ID) + select { + case <-ctx.Done(): + t.Fatal("queued fire never started") + case <-sub: + } + } + }) +} + +func TestManagerCancel(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + // Cancelling an idle job is a no-op reporting no run. + runID, err := m.Cancel(t.Context(), "backup") + assert.NilError(t, err) + assert.Check(t, is.Equal(runID, "")) + + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + runID, err = m.Cancel(t.Context(), "backup") + assert.NilError(t, err) + assert.Check(t, is.Equal(runID, run.ID)) + + // The stop makes the container exit nonzero, but the cancel override + // wins over the exit code. + done := waitTerminal(t, m, "backup", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateCancelled)) + assert.Check(t, is.Contains(fake.stopped, run.ContainerID)) +} + +func TestManagerTimeout(t *testing.T) { + m, fake := newTestManager(t) + spec := manualSpec() + spec.TimeoutSeconds = 1 + _, _, err := m.Create(t.Context(), "slow", spec) + assert.NilError(t, err) + + run, err := m.Run(t.Context(), "slow", false) + assert.NilError(t, err) + + // The container never exits by itself; the deadline stops it and the + // timed_out override wins over the kill exit code. + done := waitTerminal(t, m, "slow", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateTimedOut)) + assert.Check(t, is.Contains(fake.stopped, run.ContainerID)) +} + +func TestManagerAutoRemoval(t *testing.T) { + m, fake := newTestManager(t) + spec := manualSpec() + spec.RemoveOnSuccess = true + _, _, err := m.Create(t.Context(), "tidy", spec) + assert.NilError(t, err) + + run, err := m.Run(t.Context(), "tidy", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + done := waitTerminal(t, m, "tidy", run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded)) + // The outcome is recorded before the container is removed, so the + // removal lands an instant after the run turns terminal. + poll.WaitOn(t, func(poll.LogT) poll.Result { + gone, err := m.store.Run(run.JobID, run.ID) + if err != nil { + return poll.Error(err) + } + + if gone.ContainerGone { + return poll.Success() + } + + return poll.Continue("container removal not recorded yet") + }) + assert.Check(t, fake.container(t, run.ContainerID).removed) + + // A failed run keeps its container when only RemoveOnSuccess is set. + waitJobSettled(t, m, "tidy", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + second, err := m.Run(t.Context(), "tidy", false) + assert.NilError(t, err) + fake.exit(second.ContainerID, 1, nil) + done = waitTerminal(t, m, "tidy", second.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateFailed)) + assert.Check(t, !done.ContainerGone) + assert.Check(t, !fake.container(t, second.ContainerID).removed) +} + +func TestManagerCreateAndRunMatrix(t *testing.T) { + newManagerWithJob := func(t *testing.T) (*Manager, *fakeBackend, *jobsv0.Job) { + t.Helper() + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "existing", manualSpec()) + assert.NilError(t, err) + return m, fake, job + } + finish := func(t *testing.T, m *Manager, fake *fakeBackend, run *jobsv0.Run) { + t.Helper() + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, run.JobID, run.ID) + } + + t.Run("no name no spec", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, _, err := m.CreateAndRun(t.Context(), "", nil) + assert.Check(t, cerrdefs.IsInvalidArgument(err)) + }) + t.Run("spec only generates a name", func(t *testing.T) { + m, fake := newTestManager(t) + job, run, created, err := m.CreateAndRun(t.Context(), "", manualSpec()) + assert.NilError(t, err) + assert.Check(t, created) + assert.Check(t, job.Name != "") + finish(t, m, fake, run) + }) + t.Run("name only runs the existing job", func(t *testing.T) { + m, fake, job := newManagerWithJob(t) + got, run, created, err := m.CreateAndRun(t.Context(), "existing", nil) + assert.NilError(t, err) + assert.Check(t, !created) + assert.Check(t, is.Equal(got.ID, job.ID)) + finish(t, m, fake, run) + }) + t.Run("name only unknown job", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, _, err := m.CreateAndRun(t.Context(), "ghost", nil) + assert.Check(t, cerrdefs.IsNotFound(err)) + }) + t.Run("name and matching spec run the existing job", func(t *testing.T) { + m, fake, job := newManagerWithJob(t) + got, run, created, err := m.CreateAndRun(t.Context(), "existing", manualSpec()) + assert.NilError(t, err) + assert.Check(t, !created) + assert.Check(t, is.Equal(got.ID, job.ID)) + finish(t, m, fake, run) + }) + t.Run("name and different spec conflict", func(t *testing.T) { + m, _, _ := newManagerWithJob(t) + _, _, _, err := m.CreateAndRun(t.Context(), "existing", &jobsv0.JobSpec{ContainerSpec: []byte(`{"Image":"alpine"}`)}) + assert.Check(t, cerrdefs.IsAlreadyExists(err), "want already-exists, got %v", err) + }) + t.Run("name and spec register and run", func(t *testing.T) { + m, fake := newTestManager(t) + job, run, created, err := m.CreateAndRun(t.Context(), "fresh", manualSpec()) + assert.NilError(t, err) + assert.Check(t, created) + assert.Check(t, is.Equal(job.Name, "fresh")) + finish(t, m, fake, run) + }) + t.Run("schedule spec rejected", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, _, err := m.CreateAndRun(t.Context(), "nightly", scheduleSpec("")) + assert.Check(t, cerrdefs.IsInvalidArgument(err)) + }) + t.Run("name resolving a schedule job rejected", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + _, _, _, err = m.CreateAndRun(t.Context(), "nightly", nil) + assert.Check(t, cerrdefs.IsInvalidArgument(err)) + }) +} + +func TestManagerRemoveModes(t *testing.T) { + setup := func(t *testing.T) (*Manager, *fakeBackend, *jobsv0.Run) { + t.Helper() + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + return m, fake, waitTerminal(t, m, "backup", run.ID) + } + + t.Run("default keeps history", func(t *testing.T) { + m, fake, run := setup(t) + assert.NilError(t, m.Remove(t.Context(), "backup", "")) + _, err := m.Inspect(t.Context(), "backup") + assert.Check(t, cerrdefs.IsNotFound(err)) + kept, err := m.InspectRun(t.Context(), run.JobID, run.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(kept.State, jobsv0.RunStateSucceeded)) + assert.Check(t, !fake.container(t, run.ContainerID).removed) + }) + t.Run("remove drops history", func(t *testing.T) { + m, fake, run := setup(t) + assert.NilError(t, m.Remove(t.Context(), "backup", jobsv0.RunsRemove)) + _, err := m.store.Run(run.JobID, run.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + waitContainerRemoved(t, fake, run.ContainerID) + }) + t.Run("remove drops the in-flight run's container once it exits", func(t *testing.T) { + m, fake, _ := setup(t) + inflight, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + // The removal stops the container; the fake stop exits it, and the + // watcher then finds the run record purged. + assert.NilError(t, m.Remove(t.Context(), "backup", jobsv0.RunsRemove)) + waitContainerRemoved(t, fake, inflight.ContainerID) + }) + t.Run("remove-finished keeps the run cancelled by the removal", func(t *testing.T) { + m, fake, terminal := setup(t) + inflight, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + assert.NilError(t, m.Remove(t.Context(), "backup", jobsv0.RunsRemoveFinished)) + // The already-terminal run is dropped; the run cancelled by this + // removal was still in flight at removal time, so it is kept. + _, err = m.store.Run(terminal.JobID, terminal.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + waitContainerRemoved(t, fake, terminal.ContainerID) + + kept, err := m.store.Run(inflight.JobID, inflight.ID) + assert.NilError(t, err) + assert.Check(t, !isTerminalRunState(kept.State) || kept.State == jobsv0.RunStateCancelled) + + // The watcher records the cancellation on the tombstone history. The + // job record is gone, so the API-level Wait correctly refuses the + // reference; observe the store directly instead. + fake.exit(inflight.ContainerID, 137, nil) + done := waitStoreTerminal(t, m, inflight.JobID, inflight.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateCancelled)) + assert.Check(t, !fake.container(t, inflight.ContainerID).removed) + }) + t.Run("invalid mode", func(t *testing.T) { + m, _, _ := setup(t) + assert.Check(t, cerrdefs.IsInvalidArgument(m.Remove(t.Context(), "backup", "bogus"))) + }) +} + +// TestManagerEvictionRemovesContainers checks that history eviction removes +// the evicted runs' containers, so a frequently firing job does not +// accumulate stopped containers. +func TestManagerEvictionRemovesContainers(t *testing.T) { + m, fake := newTestManager(t) + spec := manualSpec() + spec.RunHistoryLimit = 1 + _, _, err := m.Create(t.Context(), "backup", spec) + assert.NilError(t, err) + + first, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + fake.exit(first.ContainerID, 0, nil) + waitTerminal(t, m, "backup", first.ID) + waitJobSettled(t, m, "backup", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + + second, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + _, err = m.store.Run(first.JobID, first.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + waitContainerRemoved(t, fake, first.ContainerID) + assert.Check(t, !fake.container(t, second.ContainerID).removed) +} + +// waitContainerRemoved polls until the fake backend removed the container; +// dropped runs have their containers removed in the background. +func waitContainerRemoved(t *testing.T, fake *fakeBackend, id string) { + t.Helper() + poll.WaitOn(t, func(poll.LogT) poll.Result { + if fake.container(t, id).removed { + return poll.Success() + } + + return poll.Continue("container %s not removed yet", id) + }) +} + +func TestManagerPrune(t *testing.T) { + m, fake := newTestManager(t) + idle, _, err := m.Create(t.Context(), "idle-manual", manualSpec()) + assert.NilError(t, err) + labelled := manualSpec() + labelled.Labels = map[string]string{"com.example.project": "demo"} + _, _, err = m.Create(t.Context(), "labelled-manual", labelled) + assert.NilError(t, err) + _, _, err = m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + _, _, err = m.Create(t.Context(), "busy", manualSpec()) + assert.NilError(t, err) + busyRun, err := m.Run(t.Context(), "busy", false) + assert.NilError(t, err) + + // Label-filtered prune touches only matching jobs. + removed, err := m.Prune(t.Context(), []string{"com.example.project=demo"}) + assert.NilError(t, err) + assert.Check(t, is.Equal(len(removed), 1)) + + // An unfiltered prune removes idle manual jobs, never schedule jobs + // (armed, not abandoned) nor running ones. + removed, err = m.Prune(t.Context(), nil) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(removed, []string{idle.ID})) + _, err = m.Inspect(t.Context(), "nightly") + assert.NilError(t, err) + _, err = m.Inspect(t.Context(), "busy") + assert.NilError(t, err) + + fake.exit(busyRun.ContainerID, 0, nil) + waitTerminal(t, m, "busy", busyRun.ID) +} + +func TestManagerWait(t *testing.T) { + t.Run("latest blocks until the first run", func(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + type result struct { + run *jobsv0.Run + err error + } + waited := make(chan result, 1) + go func() { + ctx, cancel := context.WithTimeout(context.WithoutCancel(t.Context()), 10*time.Second) + defer cancel() + run, err := m.Wait(ctx, "backup", "latest", "") + waited <- result{run, err} + }() + + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + got := <-waited + assert.NilError(t, got.err) + assert.Check(t, is.Equal(got.run.ID, run.ID)) + assert.Check(t, is.Equal(got.run.State, jobsv0.RunStateSucceeded)) + }) + t.Run("running condition satisfied by a terminal jump", func(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + fake.createErr = errors.New("boom") + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + got, err := m.Wait(t.Context(), "backup", run.ID, jobsv0.WaitConditionRunning) + assert.NilError(t, err) + assert.Check(t, is.Equal(got.State, jobsv0.RunStateFailed)) + }) + t.Run("explicit unknown run", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + _, err = m.Wait(t.Context(), "backup", "ghost", "") + assert.Check(t, cerrdefs.IsNotFound(err)) + }) + t.Run("invalid condition", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + _, err = m.Wait(t.Context(), "backup", "latest", "bogus") + assert.Check(t, cerrdefs.IsInvalidArgument(err)) + }) + t.Run("caller detaches on context cancellation", func(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = m.Wait(ctx, "backup", "latest", "") + assert.Check(t, is.ErrorIs(err, context.Canceled)) + }) +} + +func TestManagerPauseResume(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + _, _, err = m.Create(t.Context(), "manual", manualSpec()) + assert.NilError(t, err) + + assert.NilError(t, m.Pause(t.Context(), "nightly")) + assert.NilError(t, m.Pause(t.Context(), "nightly")) // idempotent + job, err := m.Inspect(t.Context(), "nightly") + assert.NilError(t, err) + assert.Check(t, job.Paused) + + assert.NilError(t, m.Resume(t.Context(), "nightly")) + job, err = m.Inspect(t.Context(), "nightly") + assert.NilError(t, err) + assert.Check(t, !job.Paused) + + // A no-op for manual jobs, not an error. + assert.NilError(t, m.Pause(t.Context(), "manual")) + job, err = m.Inspect(t.Context(), "manual") + assert.NilError(t, err) + assert.Check(t, !job.Paused) +} + +func TestManagerPauseDropsScheduledFires(t *testing.T) { + t.Parallel() + + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(jobsv0.ConcurrencyQueue)) + assert.NilError(t, err) + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + evidence := &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule} + _, err = m.tryFire(t.Context(), job.ID, evidence) + assert.ErrorIs(t, err, errFireQueued) + + assert.NilError(t, m.Pause(t.Context(), job.ID)) + m.mu.Lock() + _, queued := m.queued[job.ID] + m.mu.Unlock() + assert.Check(t, !queued) + + // A fire dispatched before Pause can acquire the job lock afterward. + _, err = m.tryFire(t.Context(), job.ID, evidence) + assert.ErrorIs(t, err, errFirePaused) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + _, err = m.tryFire(t.Context(), job.ID, evidence) + assert.ErrorIs(t, err, errFirePaused) + assert.Equal(t, fake.createCount(), 1) + + // Only scheduled fires are suppressed; explicit runs remain available. + run, err = m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) +} + +// failJobUpdateKV fails every job record update. +type failJobUpdateKV struct { + KV +} + +func (k failJobUpdateKV) Update(ctx context.Context, key string, value []byte) error { + if strings.HasPrefix(key, jobKeyPrefix) { + return errors.New("injected job-record write failure") + } + + return k.KV.Update(ctx, key, value) +} + +// TestManagerPauseWriteFailureKeepsSchedule checks that a pause whose record +// write fails leaves the schedule armed, matching the unpaused record. +func TestManagerPauseWriteFailureKeepsSchedule(t *testing.T) { + t.Parallel() + + m, _ := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + m.store.kv = failJobUpdateKV{KV: m.store.kv} + assert.Check(t, is.ErrorContains(m.Pause(t.Context(), job.ID), "injected job-record write failure")) + + _, armed := m.sched.nextFor(job.ID) + assert.Check(t, armed) + + stored, err := m.store.Job(job.ID) + assert.NilError(t, err) + assert.Check(t, !stored.Paused) +} + +func TestManagerList(t *testing.T) { + m, fake := newTestManager(t) + labelled := manualSpec() + labelled.Labels = map[string]string{"com.example.project": "demo"} + _, _, err := m.Create(t.Context(), "alpha", labelled) + assert.NilError(t, err) + _, _, err = m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + assert.NilError(t, m.Pause(t.Context(), "nightly")) + _, _, err = m.Create(t.Context(), "worker", manualSpec()) + assert.NilError(t, err) + run, err := m.Run(t.Context(), "worker", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 2, nil) + waitTerminal(t, m, "worker", run.ID) + + names := func(jobs []*jobsv0.Job) []string { + out := make([]string, len(jobs)) + for i, job := range jobs { + out[i] = job.Name + } + return out + } + + all, err := m.List(t.Context(), nil) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(all), []string{"alpha", "nightly", "worker"})) + + byName, err := m.List(t.Context(), &jobsv0.ListRequest{Names: []string{"alpha", "worker"}}) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(byName), []string{"alpha", "worker"})) + + byLabel, err := m.List(t.Context(), &jobsv0.ListRequest{Labels: []string{"com.example.project"}}) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(byLabel), []string{"alpha"})) + + byKind, err := m.List(t.Context(), &jobsv0.ListRequest{TriggerKinds: []string{jobsv0.TriggerKindSchedule}}) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(byKind), []string{"nightly"})) + + paused, err := m.List(t.Context(), &jobsv0.ListRequest{Paused: "true"}) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(paused), []string{"nightly"})) + + failed, err := m.List(t.Context(), &jobsv0.ListRequest{LatestRunStates: []string{jobsv0.RunStateFailed}}) + assert.NilError(t, err) + assert.Check(t, is.DeepEqual(names(failed), []string{"worker"})) + + _, err = m.List(t.Context(), &jobsv0.ListRequest{Paused: "maybe"}) + assert.Check(t, cerrdefs.IsInvalidArgument(err)) + _, err = m.List(t.Context(), &jobsv0.ListRequest{States: []string{"bogus"}}) + assert.Check(t, cerrdefs.IsInvalidArgument(err)) +} + +func TestManagerGeneratedNamesRetryOnCollision(t *testing.T) { + m, fake := newTestManager(t) + + // Exhaust a good part of the namespace cheaply: register a job, then + // verify a nameless create-and-run picks a fresh name rather than + // erroring on the first collision. Determinism comes from retrying, not + // from the generator, so assert on success and name difference only. + first, run1, _, err := m.CreateAndRun(t.Context(), "", manualSpec()) + assert.NilError(t, err) + fake.exit(run1.ContainerID, 0, nil) + waitTerminal(t, m, first.ID, run1.ID) + + second, run2, _, err := m.CreateAndRun(t.Context(), "", manualSpec()) + assert.NilError(t, err) + assert.Check(t, first.Name != second.Name) + fake.exit(run2.ContainerID, 0, nil) + waitTerminal(t, m, second.ID, run2.ID) +} + +func TestManagerRescheduleRequiresSchedule(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "adhoc", manualSpec()) + assert.NilError(t, err) + // Rebasing a cadence only means something for a schedule trigger. + _, err = m.Run(t.Context(), "adhoc", true) + assert.Check(t, cerrdefs.IsInvalidArgument(err), "want invalid-argument, got %v", err) +} + +func TestManagerWatchErrorFailsRun(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + fake.waitErr = errors.New("no such container") + + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + // A failed wait means the container is gone; the watcher records the + // failure on the run and releases the job. + done := waitTerminal(t, m, "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, m, "backup", func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) +} + +func TestManagerInspectRunRefs(t *testing.T) { + m, fake := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, "backup", run.ID) + + for _, tc := range []struct{ doc, ref string }{ + {doc: "empty means latest", ref: ""}, + {doc: "latest keyword", ref: "latest"}, + {doc: "explicit ID", ref: run.ID}, + } { + t.Run(tc.doc, func(t *testing.T) { + got, err := m.InspectRun(t.Context(), "backup", tc.ref) + assert.NilError(t, err) + assert.Check(t, is.Equal(got.ID, run.ID)) + }) + } + _, err = m.InspectRun(t.Context(), "backup", "ghost") + assert.Check(t, cerrdefs.IsNotFound(err)) +} + +func TestManagerShutdownDetachesFromRunningContainers(t *testing.T) { + m, _ := newTestManager(t) + _, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m.Run(t.Context(), "backup", false) + assert.NilError(t, err) + + // The container never exits — the live-restore shape, where the daemon + // shuts down while containers deliberately keep running. Shutdown must + // detach the watcher and return promptly instead of stalling until the + // context expires. + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + assert.NilError(t, m.Shutdown(ctx)) + + // The run record is deliberately left in flight: the crash-leftover + // shape restart reconciliation resolves on the next startup. + kept, err := m.store.Run(run.JobID, run.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(kept.State, jobsv0.RunStateRunning)) +} + +// TestManagerRunStartFailure checks that a failed start request, which may +// still have started the container, stops the container before releasing the +// job, and keeps the job occupied while it cannot be stopped. +func TestManagerRunStartFailure(t *testing.T) { + for _, stopFails := range []bool{false, true} { + t.Run(fmt.Sprintf("stop failure=%t", stopFails), func(t *testing.T) { + t.Parallel() + + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + fake.startErr = errors.New("injected start failure") + if stopFails { + fake.stopErr = errors.New("injected stop failure") + } + + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + assert.Check(t, is.Contains(fake.stoppedContainers(), run.ContainerID)) + + if stopFails { + _, err := m.Run(t.Context(), job.ID, false) + assert.Check(t, cerrdefs.IsFailedPrecondition(err), "want failed-precondition, got %v", err) + fake.exit(run.ContainerID, 0, nil) + run = waitTerminal(t, m, job.ID, run.ID) + } + + assert.Check(t, is.Equal(run.State, jobsv0.RunStateFailed)) + assert.Check(t, is.Contains(run.Error, "injected start failure")) + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + }) + } +} + +// TestManagerCancelRetriesFailedStop checks that a cancel whose stop failed +// can be retried: the first cancel decides the terminal state, and a later +// cancel stops the container again. +func TestManagerCancelRetriesFailedStop(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + fake.mu.Lock() + fake.stopErr = errors.New("injected stop failure") + fake.mu.Unlock() + + _, err = m.Cancel(t.Context(), job.ID) + assert.NilError(t, err) + poll.WaitOn(t, func(poll.LogT) poll.Result { + if len(fake.stoppedContainers()) == 1 { + return poll.Success() + } + + return poll.Continue("waiting for the failed stop") + }) + + fake.mu.Lock() + fake.stopErr = nil + fake.mu.Unlock() + + runID, err := m.Cancel(t.Context(), job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(runID, run.ID)) + + done := waitTerminal(t, m, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateCancelled)) + assert.Check(t, is.DeepEqual(fake.stoppedContainers(), []string{run.ContainerID, run.ContainerID})) +} + +// failFirstStopBackend fails the first container stop. +type failFirstStopBackend struct { + *fakeBackend + failed atomic.Bool +} + +func (b *failFirstStopBackend) ContainerStop(ctx context.Context, name string) error { + if b.failed.CompareAndSwap(false, true) { + return errors.New("injected stop failure") + } + + return b.fakeBackend.ContainerStop(ctx, name) +} + +// TestManagerTimeoutStopsAfterFailedCancel checks that the deadline still +// stops a container whose cancel failed to stop it, keeping the cancel's +// outcome. +func TestManagerTimeoutStopsAfterFailedCancel(t *testing.T) { + store, _ := newTestStore(t) + backend := &failFirstStopBackend{fakeBackend: newFakeBackend()} + m := NewManager(store, backend, testNameGenerator) + + t.Cleanup(func() { assert.Check(t, m.Shutdown(context.WithoutCancel(t.Context()))) }) + + spec := manualSpec() + spec.TimeoutSeconds = 1 + job, _, err := m.Create(t.Context(), "slow", spec) + assert.NilError(t, err) + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + _, err = m.Cancel(t.Context(), job.ID) + assert.NilError(t, err) + + done := waitTerminal(t, m, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateCancelled)) + assert.Check(t, is.DeepEqual(backend.stoppedContainers(), []string{run.ContainerID})) +} + +// TestManagerReleaseRequiresLatestRun checks that a run overtaken by a newer +// one, as when its completion was retried, does not return the job to idle. +func TestManagerReleaseRequiresLatestRun(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "backup", manualSpec()) + assert.NilError(t, err) + + stale := &jobsv0.Run{ID: "stale", JobID: job.ID, State: jobsv0.RunStateFailed, CreatedAtNano: 1} + assert.NilError(t, m.store.CreateRun(t.Context(), stale)) + run, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + m.locks.Lock(job.ID) + m.releaseJobLocked(t.Context(), job.ID, stale.ID) + m.locks.Unlock(job.ID) + + stored, err := m.store.Job(job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(stored.State, jobsv0.JobStateRunning)) + + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) +} + +// TestManagerQueuedFireKeepsNextFireTime checks that a fire deferred by the +// queue policy leaves the job's recorded next-fire time in place until the +// deferred fire runs, so a restart in between catches it up. +func TestManagerQueuedFireKeepsNextFireTime(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec(jobsv0.ConcurrencyQueue)) + assert.NilError(t, err) + first, err := m.Run(t.Context(), job.ID, false) + assert.NilError(t, err) + + fires, _ := m.sched.takeDue(time.Unix(0, job.NextFireAtNano)) + assert.Assert(t, is.Len(fires, 1)) + m.fireDue(t.Context(), fires[0].jobID, fires[0].gen, fires[0].scheduledAt) + + stored, err := m.store.Job(job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(stored.NextFireAtNano, job.NextFireAtNano)) + + following, _ := m.sched.nextFor(job.ID) + + fake.exit(first.ContainerID, 0, nil) + poll.WaitOn(t, func(poll.LogT) poll.Result { + stored, err := m.store.Job(job.ID) + if err != nil { + return poll.Error(err) + } + + if stored.NextFireAtNano == following.UnixNano() { + return poll.Success() + } + + return poll.Continue("next fire time not advanced yet") + }) + + deferred, err := m.store.LatestRun(job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(deferred.Trigger.ScheduledAtNano, job.NextFireAtNano)) + fake.exit(deferred.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, deferred.ID) +} diff --git a/daemon/internal/jobs/query.go b/daemon/internal/jobs/query.go new file mode 100644 index 0000000000..7dac9fe087 --- /dev/null +++ b/daemon/internal/jobs/query.go @@ -0,0 +1,432 @@ +package jobs + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/containerd/log" + "github.com/moby/moby/v2/errdefs" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" +) + +// Inspect returns a job with its latest run composed in. +func (m *Manager) Inspect(ctx context.Context, jobRef string) (*jobsv0.Job, error) { + job, err := m.resolveJob(jobRef) + if err != nil { + return nil, err + } + return m.composeJob(job), nil +} + +// List returns the jobs matching every filter of the request, latest runs +// composed in. +func (m *Manager) List(ctx context.Context, req *jobsv0.ListRequest) ([]*jobsv0.Job, error) { + if req == nil { + req = &jobsv0.ListRequest{} + } + switch req.Paused { + case "", "true", "false": + default: + return nil, errdefs.InvalidParameter(fmt.Errorf("invalid paused filter %q: must be true or false", req.Paused)) + } + for _, state := range req.States { + if state != jobsv0.JobStateIdle && state != jobsv0.JobStateRunning { + return nil, errdefs.InvalidParameter(fmt.Errorf("invalid state filter %q", state)) + } + } + for _, kind := range req.TriggerKinds { + if kind != jobsv0.TriggerKindManual && kind != jobsv0.TriggerKindSchedule { + return nil, errdefs.InvalidParameter(fmt.Errorf("invalid trigger filter %q", kind)) + } + } + for _, state := range req.LatestRunStates { + if !isTerminalRunState(state) && state != jobsv0.RunStatePending && state != jobsv0.RunStateRunning { + return nil, errdefs.InvalidParameter(fmt.Errorf("invalid latest-run-state filter %q", state)) + } + } + + var jobs []*jobsv0.Job + for _, job := range m.store.Jobs() { + if len(req.Names) > 0 && !slices.Contains(req.Names, job.Name) { + continue + } + if len(req.States) > 0 && !slices.Contains(req.States, job.State) { + continue + } + if len(req.TriggerKinds) > 0 && !slices.Contains(req.TriggerKinds, triggerKind(job.Spec.Trigger)) { + continue + } + if (req.Paused == "true" && !job.Paused) || (req.Paused == "false" && job.Paused) { + continue + } + if !matchesLabels(job.Spec.Labels, req.Labels) { + continue + } + job = m.composeJob(job) + if len(req.LatestRunStates) > 0 && (job.LatestRun == nil || !slices.Contains(req.LatestRunStates, job.LatestRun.State)) { + continue + } + jobs = append(jobs, job) + } + slices.SortFunc(jobs, func(a, b *jobsv0.Job) int { return strings.Compare(a.Name, b.Name) }) + return jobs, nil +} + +// Pause suppresses a job's schedule trigger. An in-flight run is not +// affected; explicit runs still work. Idempotent, a no-op for manual jobs. +func (m *Manager) Pause(ctx context.Context, jobRef string) error { + return m.setPaused(ctx, jobRef, true) +} + +// Resume re-arms a paused job's schedule trigger from the next tick. +// Idempotent, a no-op for manual jobs. +func (m *Manager) Resume(ctx context.Context, jobRef string) error { + return m.setPaused(ctx, jobRef, false) +} + +func (m *Manager) setPaused(ctx context.Context, jobRef string, paused bool) error { + job, err := m.resolveJob(jobRef) + if err != nil { + return err + } + if triggerKind(job.Spec.Trigger) == jobsv0.TriggerKindManual { + return nil + } + m.locks.Lock(job.ID) + defer m.locks.Unlock(job.ID) + job, err = m.store.Job(job.ID) + if err != nil { + return err + } + if job.Paused == paused { + return nil + } + // The scheduler and the queued fire change only once the new state is + // persisted, so a failed write leaves the job as it was. + job.Paused = paused + + var ( + schedule *cronSchedule + loc *time.Location + next time.Time + rearm bool + ) + + if paused { + job.NextFireAtNano = 0 + } else { + // Re-arm from the next occurrence; fires missed while paused are + // not backfilled. + if schedule, loc, err = parseScheduleTrigger(job.Spec.Trigger.Schedule); err != nil { + return err + } + + if next, rearm = schedule.next(m.now(), loc); rearm { + job.NextFireAtNano = next.UnixNano() + } + } + job.UpdatedAtNano = m.now().UnixNano() + if err := m.store.UpdateJob(ctx, job); err != nil { + return err + } + + if paused { + m.sched.disarm(job.ID) + // A deferred fire must not survive a pause/resume cycle. + m.mu.Lock() + delete(m.queued, job.ID) + m.mu.Unlock() + } else if rearm { + m.sched.arm(job.ID, schedule, loc, next) + } + + return nil +} + +// removeRunsPageSize is the page size Remove drains terminal runs with; a +// smaller page than the store default keeps each critical section short. +const removeRunsPageSize = 200 + +// Remove deletes a job, cancelling its in-flight run and applying the +// requested run-history retention mode. +func (m *Manager) Remove(ctx context.Context, jobRef, runsRemoval string) error { + switch runsRemoval { + case "", jobsv0.RunsKeep, jobsv0.RunsRemove, jobsv0.RunsRemoveFinished: + default: + return errdefs.InvalidParameter(fmt.Errorf("invalid runs-removal mode %q", runsRemoval)) + } + + jobID, err := m.resolveHistoryJobID(jobRef) + if err != nil { + return err + } + + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + job, err := m.store.Job(jobID) + if err != nil { + if cerrdefs.IsNotFound(err) && runsRemoval == jobsv0.RunsRemove { + return m.store.PurgeJob(ctx, jobID) + } + + return err + } + + // Cancel the in-flight run. Its record stays non-terminal until the + // watcher observes the container exit, which makes the remove-finished + // retention rule hold naturally: the run being cancelled by this very + // removal still counts as in-flight below. + if run, err := m.store.LatestRun(job.ID); err == nil && !isTerminalRunState(run.State) { + // Like Cancel, stop even when the terminal decision was claimed + // already, as an earlier stop may have failed. + m.setOverride(run.ID, jobsv0.RunStateCancelled) + containerID := run.ContainerID + stopCtx := context.WithoutCancel(ctx) + + m.background.Go(func() { + if err := m.backend.ContainerStop(stopCtx, containerID); err != nil { + log.G(stopCtx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop run container of removed job") + } + }) + } + + var removeErr error + switch runsRemoval { + case jobsv0.RunsRemove: + removeErr = m.store.PurgeJob(ctx, job.ID) + case jobsv0.RunsRemoveFinished: + for { + page, cursor, _, err := m.store.Runs(job.ID, removeRunsPageSize, "") + if err != nil { + return err + } + deleted := false + for _, run := range page { + if isTerminalRunState(run.State) { + if err := m.store.DeleteRun(ctx, job.ID, run.ID); err != nil { + return err + } + deleted = true + } + } + // At most one run is ever non-terminal (the in-flight one), so + // a page with nothing left to delete means only that run + // remains; the guard keeps the loop finite even if that + // invariant is ever broken. + if cursor == "" || !deleted { + break + } + } + removeErr = m.store.DeleteJob(ctx, job.ID) + default: + removeErr = m.store.DeleteJob(ctx, job.ID) + } + // Removing the history can fail after the job record is gone; the job + // must not stay armed or keep its waiters blocked then. + if _, err := m.store.Job(job.ID); cerrdefs.IsNotFound(err) { + m.sched.disarm(job.ID) + m.mu.Lock() + delete(m.queued, job.ID) + m.mu.Unlock() + // Wake waiters so a Wait on the removed job re-resolves to + // not-found instead of blocking until its context expires. + m.broadcast(job.ID) + } + return removeErr +} + +// Prune removes idle manual jobs matching the label filters, keeping their +// run history. Schedule jobs are never pruned: idle means armed, not +// abandoned. +func (m *Manager) Prune(ctx context.Context, labels []string) ([]string, error) { + var removed []string + for _, job := range m.store.Jobs() { + if triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindManual { + continue + } + if !matchesLabels(job.Spec.Labels, labels) { + continue + } + m.locks.Lock(job.ID) + // Re-read under the lock: the job may have started running or been + // removed since the snapshot. + current, err := m.store.Job(job.ID) + if err == nil && current.State == jobsv0.JobStateIdle { + if err := m.store.DeleteJob(ctx, job.ID); err == nil { + removed = append(removed, job.ID) + m.broadcast(job.ID) + } + } + m.locks.Unlock(job.ID) + } + slices.Sort(removed) + return removed, nil +} + +// RunsPage returns one page of a job's runs, newest first. +func (m *Manager) RunsPage(ctx context.Context, jobRef string, limit int, before string) (page []*jobsv0.Run, nextCursor string, staleCursor bool, err error) { + jobID, err := m.resolveHistoryJobID(jobRef) + if err != nil { + return nil, "", false, err + } + + return m.store.Runs(jobID, limit, before) +} + +// InspectRun returns one run; runRef accepts an ID, "latest", or empty for +// latest. +func (m *Manager) InspectRun(ctx context.Context, jobRef, runRef string) (*jobsv0.Run, error) { + jobID, err := m.resolveHistoryJobID(jobRef) + if err != nil { + return nil, err + } + + return m.resolveRun(jobID, runRef) +} + +// resolveHistoryJobID includes retained history in ID resolution, before +// trying live job names. Removed job names may belong to a new job now. +func (m *Manager) resolveHistoryJobID(ref string) (string, error) { + if _, err := m.store.LatestRun(ref); err == nil { + return ref, nil + } + + job, err := m.resolveJob(ref) + if err != nil { + return "", err + } + + return job.ID, nil +} + +func (m *Manager) resolveRun(jobID, runRef string) (*jobsv0.Run, error) { + if runRef == "" || runRef == "latest" { + return m.store.LatestRun(jobID) + } + return m.store.Run(jobID, runRef) +} + +// Wait blocks until the referenced run satisfies the condition and returns +// it. Waiting for "latest" on a job with no runs blocks until one is fired, +// and follows the first run it observes, not later ones. The job is +// resolved once, so a job re-created under the same name is not followed +// either. A run that jumps past the condition satisfies it: waiting for +// running on a run that went straight to a terminal state returns that +// terminal run. +func (m *Manager) Wait(ctx context.Context, jobRef, runRef, condition string) (*jobsv0.Run, error) { + switch condition { + case "", jobsv0.WaitConditionTerminal, jobsv0.WaitConditionRunning: + default: + return nil, errdefs.InvalidParameter(fmt.Errorf("invalid wait condition %q", condition)) + } + + job, err := m.resolveJob(jobRef) + if err != nil { + return nil, err + } + // check reports the run once it meets the condition, pinning runRef to + // the first run observed. + check := func() (_ *jobsv0.Run, met bool, _ error) { + run, err := m.resolveRun(job.ID, runRef) + if err != nil { + if runRef == "" || runRef == "latest" { + // No run fired yet. + return nil, false, nil + } + // An explicit run ID either exists or never will. + return nil, false, err + } + + runRef = run.ID + + return run, waitConditionMet(run, condition), nil + } + for { + if run, met, err := check(); err != nil || met { + return run, err + } + // Subscribe, then re-check both the job and the run: a transition — + // or the job's removal, which broadcasts exactly once — between the + // checks above and the subscription would otherwise be missed + // forever. Returns that were not woken withdraw their subscription + // so entries do not accumulate on jobs that stopped transitioning. + ch := m.subscribe(job.ID) + if _, err := m.store.Job(job.ID); err != nil { + m.unsubscribe(job.ID, ch) + return nil, err + } + + if run, met, err := check(); err != nil || met { + m.unsubscribe(job.ID, ch) + return run, err + } + select { + case <-ctx.Done(): + m.unsubscribe(job.ID, ch) + return nil, ctx.Err() + case <-ch: + } + } +} + +// waitConditionMet reports whether a run satisfies a wait condition. The +// zero condition means terminal. +func waitConditionMet(run *jobsv0.Run, condition string) bool { + if condition == jobsv0.WaitConditionRunning { + return run.State != jobsv0.RunStatePending + } + return isTerminalRunState(run.State) +} + +// matchesLabels reports whether labels satisfy every filter entry, each +// either "key" (presence) or "key=value" (equality), matching the container +// API's label-filter semantics. +func matchesLabels(labels map[string]string, filterEntries []string) bool { + for _, entry := range filterEntries { + key, value, hasValue := strings.Cut(entry, "=") + stored, present := labels[key] + if !present || (hasValue && stored != value) { + return false + } + } + return true +} + +// Run executes an existing job, creating its next run. The returned run may +// already be terminal when the container could not be created or started; +// the failure is recorded on the run record, per the API contract. +// +// With reschedule, the manual fire stands in for the job's upcoming +// scheduled occurrence: the schedule re-arms on the occurrence after it, +// instead of keeping the original cadence. An occurrence a tick already +// consumed but has not fired yet counts as the upcoming one. +func (m *Manager) Run(ctx context.Context, jobRef string, reschedule bool) (*jobsv0.Run, error) { + job, err := m.resolveJob(jobRef) + if err != nil { + return nil, err + } + if reschedule && triggerKind(job.Spec.Trigger) != jobsv0.TriggerKindSchedule { + return nil, errdefs.InvalidParameter(errors.New("reschedule requires a schedule trigger")) + } + + // Holding the job's lock across the fire and the replacement keeps a + // tick's fire for the replaced occurrence waiting until it is dropped. + m.locks.Lock(job.ID) + defer m.locks.Unlock(job.ID) + + run, err := m.fireManualLocked(ctx, job.ID) + if err != nil { + return nil, err + } + + if reschedule && m.sched.replaceNext(job.ID) { + m.persistNextFireLocked(ctx, job.ID) + } + return run, nil +} diff --git a/daemon/internal/jobs/run.go b/daemon/internal/jobs/run.go new file mode 100644 index 0000000000..f8a734db9c --- /dev/null +++ b/daemon/internal/jobs/run.go @@ -0,0 +1,604 @@ +package jobs + +import ( + "context" + "errors" + "fmt" + "slices" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/containerd/log" + "github.com/moby/moby/v2/daemon/internal/stringid" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" +) + +// errFireQueued reports that a fire was deferred behind the current run by +// the queue concurrency policy; the deferred fire happens when that run +// completes. Callers (the scheduler) treat it as a skip, not a failure. +var errFireQueued = errors.New("fire queued behind the current run") + +// errFireDropped reports that a fire was dropped because a fire is already +// queued (the queue has depth one). +var errFireDropped = errors.New("fire dropped: a fire is already queued") + +var errFirePaused = errors.New("scheduled fire dropped: the job is paused") + +// terminalOutcome is what a run ends with. +type terminalOutcome struct { + state string + exitCode *int64 + errMsg string +} + +// fireManual fires a run for an explicit Run or CreateAndRun call. +func (m *Manager) fireManual(ctx context.Context, jobID string) (*jobsv0.Run, error) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + return m.fireManualLocked(ctx, jobID) +} + +// fireManualLocked is fireManual for callers already holding the job's lock. +func (m *Manager) fireManualLocked(ctx context.Context, jobID string) (*jobsv0.Run, error) { + return m.tryFireLocked(ctx, jobID, &jobsv0.TriggerEvidence{ + Kind: jobsv0.TriggerKindManual, + FiredAtNano: m.now().UnixNano(), + }) +} + +// tryFireLocked is the single chokepoint every fire goes through, whatever +// the trigger: called under the job's lock, it applies the concurrency policy +// against the job's state re-read under that lock, and only then creates the +// run. Two triggers racing on the same job therefore cannot both pass. +func (m *Manager) tryFireLocked(ctx context.Context, jobID string, evidence *jobsv0.TriggerEvidence) (*jobsv0.Run, error) { + // A fire outlives its trigger. The daemon completes a container start + // even when the API caller goes away, so a cancelled request must not + // record a failure and release the job while its container runs. + ctx = context.WithoutCancel(ctx) + + job, err := m.store.Job(jobID) + if err != nil { + return nil, err + } + + if job.Paused && evidence.Kind == jobsv0.TriggerKindSchedule { + return nil, errFirePaused + } + + if job.State == jobsv0.JobStateRunning { + if evidence.Kind == jobsv0.TriggerKindSchedule && scheduleConcurrency(job) == jobsv0.ConcurrencyQueue { + m.mu.Lock() + _, full := m.queued[jobID] + if !full { + m.queued[jobID] = evidence + } + m.mu.Unlock() + if full { + return nil, errFireDropped + } + return nil, errFireQueued + } + return nil, cerrdefs.ErrFailedPrecondition.WithMessage(fmt.Sprintf("job %s is already running", job.Name)) + } + + now := m.now().UnixNano() + run := &jobsv0.Run{ + ID: stringid.GenerateRandomID(), + JobID: job.ID, + State: jobsv0.RunStatePending, + CreatedAtNano: now, + Trigger: evidence, + } + // The run record is persisted BEFORE the container is created. Every + // crash-safety property depends on this ordering: a daemon crash or a + // container-create failure leaves a discoverable run record instead of + // an orphaned container. Do not reorder. + if err := m.store.CreateRun(ctx, run); err != nil { + return nil, err + } + job.State = jobsv0.JobStateRunning + job.UpdatedAtNano = now + + if err := m.store.UpdateJob(ctx, job); err != nil { + m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "recording job state: " + err.Error()}) + return m.store.Run(job.ID, run.ID) + } + m.broadcast(job.ID) + + if err := m.startContainer(ctx, job, run); err != nil { + // The failure is recorded on the run, not returned: the run exists + // and its record is the outcome, per the API contract. + m.failUnstartedLocked(ctx, job, run, err.Error()) + } + return m.store.Run(job.ID, run.ID) +} + +// startContainer creates and starts the run's container, arms the timeout, +// and hands off to the exit watcher. Called under the job's lock. +func (m *Manager) startContainer(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) error { + req, err := decodeContainerSpec(job.Spec.ContainerSpec, false) + if err != nil { + // Validated at registration; only a corrupted store record gets here. + return fmt.Errorf("decoding container spec: %w", err) + } + if req.Config.Labels == nil { + req.Config.Labels = make(map[string]string, 2) + } + req.Config.Labels[LabelJobID] = job.ID + req.Config.Labels[LabelRunID] = run.ID + + created, err := m.backend.ContainerCreate(ctx, ContainerCreateRequest{ + Name: runContainerName(job, run), + Config: req.Config, + HostConfig: req.HostConfig, + NetworkingConfig: req.NetworkingConfig, + }) + if err != nil { + return fmt.Errorf("creating run container: %w", err) + } + run.ContainerID = created.ID + if err := m.store.UpdateRun(ctx, run); err != nil { + return fmt.Errorf("recording run container: %w", err) + } + + if err := m.backend.ContainerStart(ctx, created.ID); err != nil { + return fmt.Errorf("starting run container: %w", err) + } + run.State = jobsv0.RunStateRunning + run.StartedAtNano = m.now().UnixNano() + + if err := m.store.UpdateRun(ctx, run); err != nil { + // The container is already running. Keep observing it until exit, + // even if stopping fails, rather than free the job for another run. + run.Error = "recording run start: " + err.Error() + + m.background.Go(func() { + if err := m.backend.ContainerStop(ctx, run.ContainerID); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop run container after recording its start failed") + } + }) + } else { + m.broadcast(job.ID) + } + + m.watch(ctx, job, run) + + return nil +} + +// runContainerName names a run's container. The run-ID suffix keeps the name +// unique across job generations: a removed job leaves its kept containers +// behind, and a re-created job under the same name restarts its iterations +// at one. +func runContainerName(job *jobsv0.Job, run *jobsv0.Run) string { + return fmt.Sprintf("job-%s-%d-%s", job.Name, run.Iteration, stringid.TruncateID(run.ID)) +} + +// failUnstartedLocked fails a run whose start is unknown: a failed start +// request may still have started the container. A 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. Called under the job's lock. +func (m *Manager) failUnstartedLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, errMsg string) { + run.Error = errMsg + if run.ContainerID != "" { + if err := m.backend.ContainerStop(ctx, run.ContainerID); err != nil && !cerrdefs.IsNotFound(err) { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop container of unstarted run; watching it until it exits") + m.watch(ctx, job, run) + + return + } + } + + m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: errMsg}) +} + +// watch records an existing exit or arms the timeout and observes a future +// exit in the background. Called under the job's lock. +func (m *Manager) watch(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) { + // Observation ends with the manager: under live-restore the container + // keeps running past shutdown. + waitCtx, cancelWait := context.WithCancel(ctx) + + waitC, err := m.backend.ContainerWait(waitCtx, run.ContainerID) + if err != nil { + cancelWait() + // Per the Backend contract, the container no longer exists. + m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "waiting on run container: " + err.Error()}) + return + } + // Reconcile an exit before enforcing a deadline: the container may have + // finished within its timeout while the daemon was down. + select { + case status := <-waitC: + cancelWait() + m.completeRunStatusLocked(ctx, job, run, status) + return + default: + } + + // A run whose start was never recorded has no deadline to enforce. + if job.Spec.TimeoutSeconds > 0 && run.StartedAtNano != 0 { + deadline := time.Unix(0, run.StartedAtNano).Add(time.Duration(job.Spec.TimeoutSeconds) * time.Second) + m.armTimeout(run.ID, run.ContainerID, deadline.Sub(m.now())) + } + + m.background.Go(func() { + defer cancelWait() + + select { + case status := <-waitC: + m.locks.Lock(job.ID) + defer m.locks.Unlock(job.ID) + + m.completeRunStatusLocked(ctx, job, run, status) + case <-m.stop: + // Live-restore keeps the container running. Leave the run in + // flight on disk for reconciliation on the next startup. + } + }) +} + +func (m *Manager) completeRunStatusLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, status StateStatus) { + if status == nil { + // The Backend contract forbids closing without a status. + m.completeRunLocked(ctx, job, run, terminalOutcome{state: jobsv0.RunStateFailed, errMsg: "waiting on run container: wait channel closed without a status"}) + return + } + + code := int64(status.ExitCode()) + + outcome := terminalOutcome{exitCode: &code, errMsg: run.Error} + if waitErr := status.Err(); waitErr != nil { + outcome.errMsg = waitErr.Error() + } + switch override := m.peekOverride(run.ID); { + case override != "": + outcome.state = override + case outcome.errMsg != "": + outcome.state = jobsv0.RunStateFailed + case code == 0: + outcome.state = jobsv0.RunStateSucceeded + default: + outcome.state = jobsv0.RunStateFailed + } + m.completeRunLocked(ctx, job, run, outcome) +} + +// completeRunLocked writes the run's terminal record, removes its container +// when the spec asks for it, returns the job to idle, and fires any queued +// trigger. Called under the job's lock; job is the snapshot taken at fire +// time (its spec is immutable, its runtime state is re-read). +func (m *Manager) completeRunLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run, outcome terminalOutcome) { + // A terminal outcome must be recorded even when the triggering context + // was cancelled: dropping the write would leave the job and run stuck + // in running until restart reconciliation. + ctx = context.WithoutCancel(ctx) + m.disarmTimeout(run.ID) + m.clearOverride(run.ID) + + run.State = outcome.state + run.ExitCode = nil + if outcome.exitCode != nil { + run.ExitCode = &jobsv0.ExitCode{Value: *outcome.exitCode} + } + run.Error = outcome.errMsg + run.FinishedAtNano = m.now().UnixNano() + m.finishRunLocked(ctx, job, run) +} + +// recordRetryInterval is how long a failed write of a run's terminal record, +// or of its job's return to idle, waits before it is retried. +const recordRetryInterval = 5 * time.Second + +// finishRunLocked persists a terminal run record and releases the job. +// Called under the job's lock. +func (m *Manager) finishRunLocked(ctx context.Context, job *jobsv0.Job, run *jobsv0.Run) { + switch err := m.store.UpdateRun(ctx, run); { + case err == nil: + // The outcome is recorded before the container is removed, so a + // crash in between cannot lose it. + // Timed-out and cancelled runs always keep their container for + // postmortem inspection. + remove := (run.State == jobsv0.RunStateSucceeded && job.Spec.RemoveOnSuccess) || + (run.State == jobsv0.RunStateFailed && job.Spec.RemoveOnFailure) + if !remove || run.ContainerID == "" { + break + } + + if err := m.backend.ContainerRm(ctx, run.ContainerID); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("keeping run container that could not be auto-removed") + break + } + + if err := m.store.MarkContainerGone(ctx, job.ID, run.ID); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not record run container removal") + } + case cerrdefs.IsNotFound(err): + // The job was purged while the run was in flight, which left its + // still-running container behind; it goes with the record. + m.removeRunContainers(ctx, job.ID, []jobsv0.Run{*run}) + default: + // Releasing the job would leave the run in flight with nothing left + // to finish it, and a later run would hide it from restart + // reconciliation. Keep the job occupied and retry; a shutdown + // leaves the run in flight for that reconciliation. + log.G(ctx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not record run outcome; retrying") + m.retryLater(job.ID, func() { m.finishRunLocked(ctx, job, run) }) + + return + } + + m.releaseJobLocked(ctx, job.ID, run.ID) +} + +// releaseJobLocked returns a job whose run is terminal to idle, and fires any +// queued trigger. A failed write is retried on its own, as the terminal run +// record cannot be written again; until it lands the job stays running, so +// no other run starts. Called under the job's lock. +func (m *Manager) releaseJobLocked(ctx context.Context, jobID, runID string) { + // Only the job's latest run releases it. A run whose completion was + // retried can be overtaken when marking its job running failed: the job + // stayed idle on disk, and a newer run took it meanwhile. + if latest, err := m.store.LatestRun(jobID); err == nil && latest.ID != runID { + m.broadcast(jobID) + return + } + + // The job may have been removed while the run was in flight; its runs + // are then a tombstone and there is no state to restore. + if current, err := m.store.Job(jobID); err == nil { + current.State = jobsv0.JobStateIdle + current.UpdatedAtNano = m.now().UnixNano() + + if err := m.store.UpdateJob(ctx, current); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("could not return job to idle; retrying") + m.retryLater(jobID, func() { m.releaseJobLocked(ctx, jobID, runID) }) + + return + } + } + + m.broadcast(jobID) + + m.mu.Lock() + _, queued := m.queued[jobID] + m.mu.Unlock() + + if queued { + // Fire outside the held job lock. + m.background.Go(func() { m.fireQueued(ctx, jobID) }) + } +} + +// retryLater runs retry under the job's lock after recordRetryInterval, +// unless the manager shuts down first. +func (m *Manager) retryLater(jobID string, retry func()) { + m.background.Go(func() { + select { + case <-m.stop: + return + case <-time.After(recordRetryInterval): + } + + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + retry() + }) +} + +// fireQueued fires a job's deferred fire. The fire is claimed under the job's +// lock, so a pause or removal that takes the lock first discards it for good. +func (m *Manager) fireQueued(ctx context.Context, jobID string) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + m.mu.Lock() + evidence := m.queued[jobID] + delete(m.queued, jobID) + m.mu.Unlock() + + if evidence == nil { + return + } + // A failure to fire is the queued trigger's outcome, reported like a + // scheduler skip. + if _, err := m.tryFireLocked(ctx, jobID, evidence); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("queued fire was not started") + } + // The deferred occurrence stayed recorded until now. + m.persistNextFireLocked(ctx, jobID) +} + +// removeRunContainers removes the containers of dropped run records in the +// background: a run container lives as long as its record, so neither history +// eviction nor removing a job's runs leaves containers behind. In-flight runs +// are skipped, as their containers cannot be removed while running; their +// completion removes them instead. +func (m *Manager) removeRunContainers(ctx context.Context, jobID string, runs []jobsv0.Run) { + var ids []string + + for _, run := range runs { + if isTerminalRunState(run.State) && run.ContainerID != "" && !run.ContainerGone { + ids = append(ids, run.ContainerID) + } + } + + if len(ids) == 0 { + return + } + + ctx = context.WithoutCancel(ctx) + + m.background.Go(func() { + for _, id := range ids { + if err := m.backend.ContainerRm(ctx, id); err != nil && !cerrdefs.IsNotFound(err) { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID, "container": id}).Warn("could not remove container of dropped run") + } + } + }) +} + +// Cancel stops the job's in-flight run. It returns the in-flight run's ID, +// empty when no run was in flight; triggers keep firing. When a timeout +// claims the run's terminal decision first, the returned run is recorded +// timed_out: the run's recorded terminal state is authoritative, not this +// reply. +func (m *Manager) Cancel(ctx context.Context, jobRef string) (string, error) { + job, err := m.resolveJob(jobRef) + if err != nil { + return "", err + } + m.locks.Lock(job.ID) + defer m.locks.Unlock(job.ID) + + run, err := m.store.LatestRun(job.ID) + if err != nil { + if cerrdefs.IsNotFound(err) { + return "", nil + } + return "", err + } + if isTerminalRunState(run.State) { + return "", nil + } + // A timeout or an earlier cancel may have claimed the terminal decision + // already; the container is stopped regardless, as an earlier stop may + // have failed. + m.setOverride(run.ID, jobsv0.RunStateCancelled) + // A pending run without a container cannot be observed here: container + // creation happens synchronously under the job lock this method holds. + containerID := run.ContainerID + stopCtx := context.WithoutCancel(ctx) + m.background.Go(func() { + // Stopping can take the stop grace period; do not hold the job lock + // for it. The watcher records the terminal state. + if err := m.backend.ContainerStop(stopCtx, containerID); err != nil { + log.G(stopCtx).WithError(err).WithFields(log.Fields{"job": job.ID, "run": run.ID}).Warn("could not stop cancelled run container") + } + }) + return run.ID, nil +} + +// armTimeout schedules the run's deadline: past it, the container is +// stopped and the run ends timed_out (unless already cancelled). +func (m *Manager) armTimeout(runID, containerID string, timeout time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + // Shutdown disarmed the timers already; a timer armed after it would + // stop a container that live-restore keeps running. + if m.stopped { + return + } + + m.timers[runID] = time.AfterFunc(timeout, func() { + // Claim the timer entry and the override in one critical section: + // a completion racing this firing has already disarmed the entry, + // and claiming the override past that point would leak it forever + // (nothing clears overrides of completed runs) and stop a fresh + // container. + m.mu.Lock() + if _, armed := m.timers[runID]; !armed { + m.mu.Unlock() + return + } + delete(m.timers, runID) + // A cancel may have claimed the terminal decision already; the + // container is stopped regardless, as the cancel's stop may have + // failed. + if _, claimed := m.overrides[runID]; !claimed { + m.overrides[runID] = jobsv0.RunStateTimedOut + } + // Register the stop while still holding the mutex: Shutdown + // disarms timers under it before draining background work, so the + // drain cannot miss the stop or close the backend under it. + m.background.Go(func() { + if err := m.backend.ContainerStop(context.Background(), containerID); err != nil { + log.G(context.Background()).WithError(err).WithFields(log.Fields{"run": runID}).Warn("could not stop timed-out run container") + } + }) + m.mu.Unlock() + }) +} + +// disarmTimeout stops and forgets the run's timeout timer. +func (m *Manager) disarmTimeout(runID string) { + m.mu.Lock() + defer m.mu.Unlock() + if timer, armed := m.timers[runID]; armed { + timer.Stop() + delete(m.timers, runID) + } +} + +// setOverride claims the run's terminal decision; the first claimant wins. +func (m *Manager) setOverride(runID, state string) { + m.mu.Lock() + defer m.mu.Unlock() + + if _, claimed := m.overrides[runID]; !claimed { + m.overrides[runID] = state + } +} + +// peekOverride reads the run's forced terminal state without consuming it; +// clearOverride forgets it once the terminal record is written. +func (m *Manager) peekOverride(runID string) string { + m.mu.Lock() + defer m.mu.Unlock() + return m.overrides[runID] +} + +func (m *Manager) clearOverride(runID string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.overrides, runID) +} + +// subscribe returns a channel closed on the job's next state transition. +// Callers that return without being woken must unsubscribe, or the entry +// lingers until the job's next transition, which may never come. +func (m *Manager) subscribe(jobID string) chan struct{} { + ch := make(chan struct{}) + m.mu.Lock() + defer m.mu.Unlock() + m.notify[jobID] = append(m.notify[jobID], ch) + return ch +} + +// unsubscribe withdraws a subscription that was not woken. A channel already +// consumed by broadcast is simply no longer found. +func (m *Manager) unsubscribe(jobID string, ch chan struct{}) { + m.mu.Lock() + defer m.mu.Unlock() + subs := m.notify[jobID] + for i, sub := range subs { + if sub == ch { + m.notify[jobID] = slices.Delete(subs, i, i+1) + break + } + } + if len(m.notify[jobID]) == 0 { + delete(m.notify, jobID) + } +} + +// broadcast wakes every waiter subscribed to the job. +func (m *Manager) broadcast(jobID string) { + m.mu.Lock() + defer m.mu.Unlock() + for _, ch := range m.notify[jobID] { + close(ch) + } + delete(m.notify, jobID) +} + +// scheduleConcurrency returns the job's effective concurrency policy. +func scheduleConcurrency(job *jobsv0.Job) string { + if job.Spec != nil && job.Spec.Trigger != nil && job.Spec.Trigger.Schedule != nil && job.Spec.Trigger.Schedule.Concurrency != "" { + return job.Spec.Trigger.Schedule.Concurrency + } + return jobsv0.ConcurrencyForbid +} diff --git a/daemon/internal/jobs/scheduler.go b/daemon/internal/jobs/scheduler.go new file mode 100644 index 0000000000..c54c10b0e2 --- /dev/null +++ b/daemon/internal/jobs/scheduler.go @@ -0,0 +1,377 @@ +package jobs + +import ( + "container/heap" + "context" + "errors" + "sync" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/containerd/log" + jobsv0 "github.com/moby/moby/v2/extpoints/jobs/api/v0" +) + +// schedulerIdleWait bounds the loop's sleep when no job is armed, so a +// missed wake signal can only ever delay a fire, not lose it. +const schedulerIdleWait = time.Minute + +// schedEntry is one armed schedule job. +type schedEntry struct { + jobID string + schedule *cronSchedule + loc *time.Location + next time.Time + index int // position in the heap, -1 once removed +} + +// schedHeap is a min-heap of armed entries ordered by next fire time. +type schedHeap []*schedEntry + +func (h schedHeap) Len() int { return len(h) } +func (h schedHeap) Less(i, j int) bool { return h[i].next.Before(h[j].next) } +func (h schedHeap) Swap(i, j int) { + h[i], h[j] = h[j], h[i] + h[i].index = i + h[j].index = j +} + +func (h *schedHeap) Push(x any) { + entry := x.(*schedEntry) + entry.index = len(*h) + *h = append(*h, entry) +} + +func (h *schedHeap) Pop() any { + old := *h + entry := old[len(old)-1] + old[len(old)-1] = nil + entry.index = -1 + *h = old[:len(old)-1] + return entry +} + +// scheduler fires schedule jobs on the daemon's clock: a single goroutine +// sleeping until the earliest armed entry is due, dispatching fires through +// the manager's tryFireLocked chokepoint like any other trigger. +// +// Lock order: the scheduler's mutex is taken while holding a job lock (Arm +// and Disarm run under the manager's transitions) but never the other way +// around — tick releases the mutex before dispatching, so fires and +// persistence take job locks lock-free of the scheduler. +type scheduler struct { + m *Manager + + mu sync.Mutex + heap schedHeap + byJob map[string]*schedEntry + // gens holds the generation each job was last armed with. A tick's fire + // is dropped when its job was disarmed, re-armed or rescheduled since, + // so a delayed fire cannot run an occurrence of a superseded schedule or + // one a manual run replaced. An entry that tick exhausts keeps its + // generation: its last fire still stands. + gens map[string]uint64 + gen uint64 + // pending counts each job's occurrences consumed by a tick whose fires + // have not been claimed yet. + pending map[string]int + + wake chan struct{} // buffered kick after arm/disarm, so the loop re-plans + stop chan struct{} +} + +func newScheduler(m *Manager) *scheduler { + return &scheduler{ + m: m, + byJob: make(map[string]*schedEntry), + gens: make(map[string]uint64), + pending: make(map[string]int), + wake: make(chan struct{}, 1), + stop: make(chan struct{}), + } +} + +// arm registers (or re-registers) a job's next occurrence. +func (s *scheduler) arm(jobID string, schedule *cronSchedule, loc *time.Location, next time.Time) { + s.mu.Lock() + + s.gen++ + s.gens[jobID] = s.gen + delete(s.pending, jobID) + + if entry, armed := s.byJob[jobID]; armed { + entry.schedule, entry.loc, entry.next = schedule, loc, next + heap.Fix(&s.heap, entry.index) + } else { + entry := &schedEntry{jobID: jobID, schedule: schedule, loc: loc, next: next} + heap.Push(&s.heap, entry) + s.byJob[jobID] = entry + } + s.mu.Unlock() + s.kick() +} + +// disarm forgets a job. It is a no-op for unarmed jobs. +func (s *scheduler) disarm(jobID string) { + s.mu.Lock() + delete(s.gens, jobID) + delete(s.pending, jobID) + + if entry, armed := s.byJob[jobID]; armed { + heap.Remove(&s.heap, entry.index) + delete(s.byJob, jobID) + } + s.mu.Unlock() + s.kick() +} + +// replaceNext implements the reschedule semantics of the run API: a manual +// fire stands in for the job's upcoming occurrence. When a tick already +// consumed that occurrence and its fire is still waiting for the job's lock, +// which the caller holds, the fire is dropped; otherwise the upcoming +// occurrence is skipped. Either way exactly one occurrence is replaced. It +// reports whether the job was armed. +func (s *scheduler) replaceNext(jobID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + + if _, armed := s.gens[jobID]; !armed { + return false + } + + if s.pending[jobID] > 0 { + s.gen++ + s.gens[jobID] = s.gen + delete(s.pending, jobID) + + return true + } + + entry, armed := s.byJob[jobID] + if !armed { + return true + } + + if next, ok := entry.schedule.next(entry.next, entry.loc); ok { + entry.next = next + heap.Fix(&s.heap, entry.index) + } else { + heap.Remove(&s.heap, entry.index) + delete(s.byJob, jobID) + } + + return true +} + +// nextFor reports a job's armed occurrence. +func (s *scheduler) nextFor(jobID string) (time.Time, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if entry, armed := s.byJob[jobID]; armed { + return entry.next, true + } + return time.Time{}, false +} + +// claimDue claims a fire a tick dispatched, reporting false when its job was +// disarmed, re-armed or rescheduled since. +func (s *scheduler) claimDue(jobID string, gen uint64) bool { + s.mu.Lock() + defer s.mu.Unlock() + + if s.gens[jobID] != gen { + return false + } + + s.pending[jobID]-- + if s.pending[jobID] <= 0 { + delete(s.pending, jobID) + } + + return true +} + +// kick nudges the loop to re-plan its sleep after the heap changed. +func (s *scheduler) kick() { + select { + case s.wake <- struct{}{}: + default: + } +} + +// dueFire is an occurrence a tick consumed, with the generation its job was +// armed with. +type dueFire struct { + jobID string + gen uint64 + scheduledAt time.Time +} + +// tick fires every due entry and returns how long the loop may sleep before +// the next one. Dispatch happens after the scheduler lock is released: fires +// take job locks, which are held around arm/disarm calls into the scheduler. +func (s *scheduler) tick(ctx context.Context) time.Duration { + fires, wait := s.takeDue(s.m.now()) + for _, fire := range fires { + // Firing takes the job's lock, which a run may hold across container + // operations; doing it here would stall every other schedule. + s.m.background.Go(func() { s.m.fireDue(ctx, fire.jobID, fire.gen, fire.scheduledAt) }) + } + + return wait +} + +// takeDue consumes the occurrences due at now and returns how long the loop +// may sleep before the next one. +func (s *scheduler) takeDue(now time.Time) ([]dueFire, time.Duration) { + var fires []dueFire + + s.mu.Lock() + defer s.mu.Unlock() + + // An entry overdue by several occurrences (a long sleep or system + // suspend) fires once and re-arms from now: the intermediate backlog is + // deliberately dropped, mirroring how the missed-fires policy caps the + // catch-up at one run across a daemon restart. + for len(s.heap) > 0 && !s.heap[0].next.After(now) { + entry := s.heap[0] + fires = append(fires, dueFire{jobID: entry.jobID, gen: s.gens[entry.jobID], scheduledAt: entry.next}) + s.pending[entry.jobID]++ + + if next, ok := entry.schedule.next(now, entry.loc); ok { + entry.next = next + heap.Fix(&s.heap, entry.index) + } else { + // The expression has no future occurrence (e.g. a specific + // date now past): the job stays registered but disarmed. + heap.Remove(&s.heap, entry.index) + delete(s.byJob, entry.jobID) + } + } + wait := schedulerIdleWait + if len(s.heap) > 0 { + wait = min(s.heap[0].next.Sub(now), schedulerIdleWait) + } + + return fires, wait +} + +// run is the scheduler loop; it exits when stop is closed. +func (s *scheduler) run(ctx context.Context) { + timer := time.NewTimer(0) + defer timer.Stop() + for { + select { + case <-s.stop: + return + case <-s.wake: + case <-timer.C: + } + if !timer.Stop() { + // Drain a concurrently fired timer so Reset arms cleanly. + select { + case <-timer.C: + default: + } + } + timer.Reset(s.tick(ctx)) + } +} + +// fireDue fires the occurrence a tick consumed and records the job's next +// one, unless the job was disarmed, re-armed or rescheduled since the tick: a +// pause, resume, removal or reschedule taking the job's lock first +// supersedes the fire. +func (m *Manager) fireDue(ctx context.Context, jobID string, gen uint64, scheduledAt time.Time) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + if !m.sched.claimDue(jobID, gen) { + return + } + + m.fireScheduledLocked(ctx, jobID, scheduledAt) +} + +// fireScheduled routes a cron fire through the tryFireLocked chokepoint and +// turns its refusals into skip logs: for a scheduler, a fire refused by the +// concurrency policy is normal operation, not an error. The job's next +// occurrence is recorded only after the fire, so a crash in between leaves +// the occurrence to the restart's catch-up instead of losing it. A fire +// deferred by the queue policy leaves the occurrence recorded until the +// deferred fire runs, for the same reason. +func (m *Manager) fireScheduled(ctx context.Context, jobID string, scheduledAt time.Time) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + m.fireScheduledLocked(ctx, jobID, scheduledAt) +} + +// fireScheduledLocked is fireScheduled for callers already holding the job's +// lock. +func (m *Manager) fireScheduledLocked(ctx context.Context, jobID string, scheduledAt time.Time) { + evidence := &jobsv0.TriggerEvidence{ + Kind: jobsv0.TriggerKindSchedule, + ScheduledAtNano: scheduledAt.UnixNano(), + FiredAtNano: m.now().UnixNano(), + } + + _, err := m.tryFireLocked(ctx, jobID, evidence) + switch { + case err == nil: + case errors.Is(err, errFirePaused): + log.G(ctx).WithFields(log.Fields{"job": jobID}).Debug("scheduled fire skipped: the job is paused") + case errors.Is(err, errFireQueued): + log.G(ctx).WithFields(log.Fields{"job": jobID}).Debug("scheduled fire queued behind the current run") + + return + case errors.Is(err, errFireDropped): + log.G(ctx).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire dropped: a fire is already queued") + + return + case cerrdefs.IsFailedPrecondition(err): + log.G(ctx).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire skipped: the previous run is still in flight") + case cerrdefs.IsNotFound(err): + // The job was removed while the fire was in flight; the removal + // already disarmed it. + default: + log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("scheduled fire failed") + } + + m.persistNextFireLocked(ctx, jobID) +} + +// persistNextFire records a job's armed occurrence (zero for none) on its +// stored record, where inspect and list read it from, and where Start finds +// the fires missed while the daemon was down. +func (m *Manager) persistNextFire(ctx context.Context, jobID string) { + m.locks.Lock(jobID) + defer m.locks.Unlock(jobID) + + m.persistNextFireLocked(ctx, jobID) +} + +// persistNextFireLocked is persistNextFire for callers already holding the +// job's lock. It reads the occurrence from the scheduler under that lock +// instead of taking it from the caller, so a delayed write cannot replace a +// newer one. +func (m *Manager) persistNextFireLocked(ctx context.Context, jobID string) { + var atNano int64 + if next, armed := m.sched.nextFor(jobID); armed { + atNano = next.UnixNano() + } + + job, err := m.store.Job(jobID) + if err != nil { + return // removed meanwhile; nothing to record on + } + if job.NextFireAtNano == atNano { + return + } + + job.NextFireAtNano = atNano + job.UpdatedAtNano = m.now().UnixNano() + if err := m.store.UpdateJob(ctx, job); err != nil { + log.G(ctx).WithError(err).WithFields(log.Fields{"job": jobID}).Warn("could not record next fire time") + } +} diff --git a/daemon/internal/jobs/scheduler_test.go b/daemon/internal/jobs/scheduler_test.go new file mode 100644 index 0000000000..570c99ed35 --- /dev/null +++ b/daemon/internal/jobs/scheduler_test.go @@ -0,0 +1,373 @@ +package jobs + +import ( + "context" + "sync" + "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" +) + +// fakeClock is an injectable manager clock; the scheduler is exercised by +// advancing it and calling tick directly, so no test ever sleeps on cron +// granularity. +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +// newTestManagerAt builds a manager whose clock starts at the given instant. +func newTestManagerAt(t *testing.T, start time.Time) (*Manager, *fakeBackend, *fakeClock) { + t.Helper() + m, fake := newTestManager(t) + clk := &fakeClock{t: start} + m.now = clk.Now + return m, fake, clk +} + +// jan15 is an arbitrary fixed reference: 2026-01-15 00:00 UTC, a Thursday. +var jan15 = time.Date(2026, time.January, 15, 0, 0, 0, 0, time.UTC) + +func TestSchedulerFiresAndRearms(t *testing.T) { + m, fake, clk := newTestManagerAt(t, jan15) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + // Registration armed the trigger and advertised the next occurrence. + assert.Check(t, is.Equal(job.NextFireAtNano, jan15.Add(3*time.Hour).UnixNano())) + + // Before the occurrence, a tick fires nothing. + m.sched.tick(t.Context()) + _, err = m.store.LatestRun(job.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + + // Past the occurrence, a tick fires exactly one run with schedule + // evidence carrying the planned time. + clk.Advance(3*time.Hour + 10*time.Second) + m.sched.tick(t.Context()) + run, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.Trigger.Kind, jobsv0.TriggerKindSchedule)) + assert.Check(t, is.Equal(run.Trigger.ScheduledAtNano, jan15.Add(3*time.Hour).UnixNano())) + + // While the run is in flight the record advertises no next fire. + inFlight, err := m.Inspect(t.Context(), job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(inFlight.NextFireAtNano, int64(0))) + + fake.exit(run.ContainerID, 0, nil) + done := waitTerminal(t, m, job.ID, run.ID) + assert.Check(t, is.Equal(done.State, jobsv0.RunStateSucceeded)) + + // After completion the job advertises the re-armed next occurrence. + rearmed := jan15.Add(27 * time.Hour).UnixNano() + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed }) +} + +func TestSchedulerForbidSkipsWhileRunning(t *testing.T) { + m, fake, clk := newTestManagerAt(t, jan15) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + clk.Advance(3*time.Hour + time.Second) + m.sched.tick(t.Context()) + first, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning) + assert.NilError(t, err) + + // The next occurrence comes due while the first run is still in + // flight: forbid (the default) drops the fire. The fire is driven + // synchronously here — tick's asynchronous dispatch of due entries is + // covered above — so the count below cannot race the dispatch. + clk.Advance(24 * time.Hour) + m.fireScheduled(t.Context(), job.ID, jan15.Add(27*time.Hour)) + page, _, _, err := m.store.Runs(job.ID, 10, "") + assert.NilError(t, err) + assert.Check(t, is.Equal(len(page), 1)) + + fake.exit(first.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, first.ID) +} + +// shutdownCtx bounds the drain waits used mid-test. +func shutdownCtx(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.WithoutCancel(t.Context()), 10*time.Second) + t.Cleanup(cancel) + return ctx +} + +func TestSchedulerPauseResume(t *testing.T) { + m, _, clk := newTestManagerAt(t, jan15) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + assert.NilError(t, m.Pause(t.Context(), job.ID)) + paused, err := m.Inspect(t.Context(), job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(paused.NextFireAtNano, int64(0))) + + // The occurrence passes while paused: nothing fires, and resuming + // re-arms from the next tick without backfilling. + clk.Advance(4 * time.Hour) + m.sched.tick(t.Context()) + _, err = m.store.LatestRun(job.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + + assert.NilError(t, m.Resume(t.Context(), job.ID)) + resumed, err := m.Inspect(t.Context(), job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(resumed.NextFireAtNano, jan15.Add(27*time.Hour).UnixNano())) +} + +func TestSchedulerMissedFiresOnStart(t *testing.T) { + for _, tc := range []struct { + doc string + policy string + expectCatchUp bool + }{ + {doc: "one fires a single catch-up", policy: "", expectCatchUp: true}, + {doc: "skip drops missed fires", policy: jobsv0.MissedFiresSkip, expectCatchUp: false}, + } { + t.Run(tc.doc, func(t *testing.T) { + m, fake, clk := newTestManagerAt(t, jan15) + spec := scheduleSpec("") + spec.Trigger.Schedule.MissedFires = tc.policy + job, _, err := m.Create(t.Context(), "nightly", spec) + assert.NilError(t, err) + + // The daemon was down across the 03:00 occurrence: the clock + // advances but no tick observed it. Start applies the policy + // from the persisted next-fire time. + clk.Advance(5 * time.Hour) + m.Start(t.Context()) + + if tc.expectCatchUp { + run, err := m.Wait(t.Context(), job.ID, "latest", jobsv0.WaitConditionRunning) + assert.NilError(t, err) + assert.Check(t, is.Equal(run.Trigger.ScheduledAtNano, jan15.Add(3*time.Hour).UnixNano())) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) + } else { + assert.NilError(t, m.Shutdown(shutdownCtx(t))) + _, err := m.store.LatestRun(job.ID) + assert.Check(t, cerrdefs.IsNotFound(err)) + } + + // Either way the schedule re-armed from the next occurrence, + // not from the backlog. The catch-up fire runs concurrently + // with Start's own arm+persist, so the re-arm write can land + // an instant after the run turns terminal. + rearmed := jan15.Add(27 * time.Hour).UnixNano() + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed }) + }) + } +} + +func TestSchedulerReschedule(t *testing.T) { + m, fake, clk := newTestManagerAt(t, jan15) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + // A manual fire with reschedule stands in for the upcoming occurrence: + // the schedule skips 03:00 today and re-arms on tomorrow's. + run, err := m.Run(t.Context(), job.ID, true) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) + + rearmed := jan15.Add(27 * time.Hour).UnixNano() + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.NextFireAtNano == rearmed }) + + // Today's occurrence passes without a fire. + clk.Advance(4 * time.Hour) + m.sched.tick(t.Context()) + assert.NilError(t, m.Shutdown(shutdownCtx(t))) + page, _, _, err := m.store.Runs(job.ID, 10, "") + assert.NilError(t, err) + assert.Check(t, is.Equal(len(page), 1)) +} + +func TestSchedulerRemoveDisarms(t *testing.T) { + m, _, clk := newTestManagerAt(t, jan15) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + assert.NilError(t, m.Remove(t.Context(), job.ID, "")) + clk.Advance(4 * time.Hour) + m.sched.tick(t.Context()) + assert.NilError(t, m.Shutdown(shutdownCtx(t))) + _, err = m.store.LatestRun(job.ID) + assert.Check(t, cerrdefs.IsNotFound(err), "a removed job must not fire") +} + +func TestSchedulerStartWithUnparseableStoredCron(t *testing.T) { + // A schedule the validation would never accept can still sit in the + // store (written by a newer or buggy daemon); Start must leave the job + // disarmed instead of failing the whole startup. + store, _ := newTestStore(t) + job := makeJob("job1", "broken") + job.Spec.Trigger.Schedule.Cron = "not a cron" + assert.NilError(t, store.CreateJob(t.Context(), job)) + + m := NewManager(store, newFakeBackend(), testNameGenerator) + clk := &fakeClock{t: jan15} + m.now = clk.Now + m.Start(t.Context()) + t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) }) + + clk.Advance(24 * time.Hour) + m.sched.tick(t.Context()) + _, err := m.store.LatestRun("job1") + assert.Check(t, cerrdefs.IsNotFound(err), "a job with an unparseable schedule must never fire") +} + +func TestSchedulerHonorsTimezone(t *testing.T) { + m, _, _ := newTestManagerAt(t, jan15) + spec := scheduleSpec("") + spec.Trigger.Schedule.Timezone = "Europe/Paris" + job, _, err := m.Create(t.Context(), "paris-nightly", spec) + assert.NilError(t, err) + + // 03:00 Paris is 02:00 UTC in January. + paris, err := time.LoadLocation("Europe/Paris") + assert.NilError(t, err) + assert.Check(t, is.Equal(job.NextFireAtNano, time.Date(2026, time.January, 15, 3, 0, 0, 0, paris).UnixNano())) + assert.Check(t, is.Equal(job.NextFireAtNano, jan15.Add(2*time.Hour).UnixNano())) +} + +func TestSchedulerCreateRejectsNeverFiringCron(t *testing.T) { + m, _ := newTestManager(t) + spec := scheduleSpec("") + spec.Trigger.Schedule.Cron = "0 0 30 2 *" + _, _, err := m.Create(t.Context(), "never", spec) + assert.Check(t, cerrdefs.IsInvalidArgument(err), "want invalid-argument, got %v", err) +} + +// TestSchedulerStartKeepsRescheduledOccurrence checks that a restart keeps the +// occurrence a reschedule persisted instead of firing the one it skipped, and +// keeps no later one. +func TestSchedulerStartKeepsRescheduledOccurrence(t *testing.T) { + schedule, err := parseCron("0 3 * * *") + assert.NilError(t, err) + + upcoming, _ := schedule.next(time.Now(), time.UTC) + rescheduled, _ := schedule.next(upcoming, time.UTC) + + for _, tc := range []struct { + doc string + stored time.Time + expected time.Time + }{ + {doc: "the occurrence after the upcoming one is kept", stored: rescheduled, expected: rescheduled}, + {doc: "a later occurrence is recomputed", stored: rescheduled.Add(24 * time.Hour), expected: upcoming}, + } { + t.Run(tc.doc, func(t *testing.T) { + store, _ := newTestStore(t) + job := makeJob("job1", "nightly") + job.NextFireAtNano = tc.stored.UnixNano() + assert.NilError(t, store.CreateJob(t.Context(), job)) + + m := NewManager(store, newFakeBackend(), testNameGenerator) + m.Start(t.Context()) + t.Cleanup(func() { assert.Check(t, m.Shutdown(shutdownCtx(t))) }) + + next, armed := m.sched.nextFor(job.ID) + assert.Assert(t, armed) + assert.Check(t, is.Equal(next.UnixNano(), tc.expected.UnixNano())) + }) + } +} + +// TestSchedulerDropsSupersededFire checks that a tick's fire delayed past a +// pause neither overwrites the cleared next-fire time nor, after a resume, +// runs the occurrence of the superseded schedule. +func TestSchedulerDropsSupersededFire(t *testing.T) { + m, _ := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + m.sched.mu.Lock() + gen := m.sched.gens[job.ID] + m.sched.mu.Unlock() + + scheduledAt := time.Unix(0, job.NextFireAtNano) + + assert.NilError(t, m.Pause(t.Context(), job.ID)) + m.fireDue(t.Context(), job.ID, gen, scheduledAt) + paused, err := m.store.Job(job.ID) + assert.NilError(t, err) + assert.Check(t, is.Equal(paused.NextFireAtNano, int64(0))) + + assert.NilError(t, m.Resume(t.Context(), job.ID)) + m.fireDue(t.Context(), job.ID, gen, scheduledAt) + _, err = m.store.LatestRun(job.ID) + assert.Check(t, cerrdefs.IsNotFound(err), "a superseded fire must not run") +} + +// TestSchedulerRescheduleReplacesConsumedOccurrence checks that a reschedule +// landing after a tick consumed the upcoming occurrence, but before its fire +// ran, replaces that occurrence instead of letting it fire and skipping the +// following one too. +func TestSchedulerRescheduleReplacesConsumedOccurrence(t *testing.T) { + m, fake := newTestManager(t) + job, _, err := m.Create(t.Context(), "nightly", scheduleSpec("")) + assert.NilError(t, err) + + fires, _ := m.sched.takeDue(time.Unix(0, job.NextFireAtNano)) + assert.Assert(t, is.Len(fires, 1)) + + following, _ := m.sched.nextFor(job.ID) + + run, err := m.Run(t.Context(), job.ID, true) + assert.NilError(t, err) + fake.exit(run.ContainerID, 0, nil) + waitTerminal(t, m, job.ID, run.ID) + waitJobSettled(t, m, job.ID, func(j *jobsv0.Job) bool { return j.State == jobsv0.JobStateIdle }) + + m.fireDue(t.Context(), fires[0].jobID, fires[0].gen, fires[0].scheduledAt) + page, _, _, err := m.store.Runs(job.ID, 10, "") + assert.NilError(t, err) + assert.Check(t, is.Len(page, 1), "the replaced occurrence must not fire") + + next, _ := m.sched.nextFor(job.ID) + assert.Check(t, is.Equal(next.UnixNano(), following.UnixNano())) +} + +// TestSchedulerStartSkipsCatchUpAlreadyFired checks that a restart does not +// catch up an occurrence whose run was recorded before the daemon died, ahead +// of the next fire time. +func TestSchedulerStartSkipsCatchUpAlreadyFired(t *testing.T) { + store, _ := newTestStore(t) + missed := time.Now().Add(-time.Hour).UnixNano() + job := makeJob("job1", "nightly") + job.NextFireAtNano = missed + assert.NilError(t, store.CreateJob(t.Context(), job)) + + run := makeRun(job.ID, "run1", jobsv0.RunStateSucceeded) + run.Trigger = &jobsv0.TriggerEvidence{Kind: jobsv0.TriggerKindSchedule, ScheduledAtNano: missed, FiredAtNano: missed} + assert.NilError(t, store.CreateRun(t.Context(), run)) + + m := NewManager(store, newFakeBackend(), testNameGenerator) + m.Start(t.Context()) + assert.NilError(t, m.Shutdown(shutdownCtx(t))) + + page, _, _, err := m.store.Runs(job.ID, 10, "") + assert.NilError(t, err) + assert.Check(t, is.Len(page, 1)) +}