executor

type AgentLostCandidate ¶

type AgentLostCandidate struct {
	TaskInstanceID string
	TenantID       string
	DagRunID       string
	DagID          string
	TaskID         string
	// TryNumber is the attempt the candidate row is on, so the reaper can
	// tear down EXACTLY that attempt's pod after failing it (#474). A retry
	// bumps try_number in place and dispatches a new pod with a new
	// try-number label, so pinning it here means a newer live attempt's pod
	// can never be deleted by mistake.
	TryNumber int
	// AttemptEpoch is the epoch the row's current execution was dispatched
	// with (ADR 0051 amendment). The mark and the pod teardown are pinned to it
	// as well as to TryNumber, so a row re-placed or re-dispatched between the
	// list and the write is a different attempt and is left alone.
	AttemptEpoch int
	// StartedAt is when the attempt entered running (zero when unknown). The
	// reaper measures LastHeartbeat from it against
	// auth.max_attempt_credential_lifetime: an attempt that went silent past the
	// ceiling stopped getting its credential renewed, so its silence is the
	// credential lapsing, not a lost agent (#1461).
	StartedAt     time.Time
	LastHeartbeat time.Time
}

AgentLostCandidate is one task instance in `running` whose agent may have gone silent, with the timestamp of its most recent heartbeat. The reaper compares the gap from this stamp to "now" against a stall threshold; a non-zero gap larger than the threshold means the agent is presumed gone and the TI is failed with reason "agent_lost".

type Attempt ¶ added in v0.5.1

type Attempt struct {
	RunID        string
	TaskID       string
	TryNumber    int
	AttemptEpoch int
}

Attempt names exactly one execution of a task instance: its run, its task, the try it is on and the attempt epoch the dispatcher claimed for it (ADR 0051 amendment). The try alone is not enough: an infra re-place, a reschedule poke or a repeated dispatch starts a new execution on the same try, and only the epoch tells two pods of one try apart. A pod stamped before the epoch existed carries no epoch label and is epoch 0.

type AttemptStopper ¶ added in v0.5.3

type AttemptStopper interface {
	// StopAttempt stops the attempt's task (SIGTERM, a grace, then SIGKILL to
	// its whole process group) and reports true once nothing of it is left.
	// false with a nil error means nothing could be verified and nothing was
	// signaled.
	StopAttempt(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)
}

AttemptStopper is implemented by a ProcessLiveness that can stop the task of an attempt whose agent is still alive (Lite: the subprocess executor signals the task's recorded process group). The agent-lost reaper uses it for an attempt still running past the credential ceiling (auth.max_attempt_credential_lifetime), the Lite counterpart of a task pod's activeDeadlineSeconds (#1511): it fails the attempt as credential_ceiling first, then stops its task. A ProcessLiveness without it (or none, the pod path) leaves that attempt to the pod deadline and the silent-agent path.

type BusyWarmWorkerSource ¶

type BusyWarmWorkerSource interface {
	ListBusyWarmWorkerPods(ctx context.Context) (map[string]bool, error)
}

BusyWarmWorkerSource yields the set of warm-worker pod names currently serving a `running` attempt (ADR 0058 N1d-b): a warm worker is BUSY iff some `running` task_instance is durably bound to it (warm_worker_id = the pod's own name — the binding landed in N1d-a1/a2). Returned as a set keyed by pod name so the reconciler classifies each live pod in O(1).

Implemented on the storage side and defined HERE so the reconciler depends on a narrow capability rather than importing storage. With warm pools off no TI is ever bound, so the set is always empty and every worker classifies as idle — byte-for-byte today's dedicated pod-per-task behavior.

type DecisionRecorder ¶

type DecisionRecorder interface {
	RecordSchedulerDecision(decisionType string)
}

DecisionRecorder records the reaper's per-tick decision metrics. It is the narrow slice of the scheduler's metrics sink the reapers need — only the one counter — so the executor depends on a capability, not on the observability package. A nil DecisionRecorder is tolerated (tests need not stub it).

type DispatchLostReapStore ¶

type DispatchLostReapStore interface {
	// ListStaleQueuedCandidates returns one page of the `queued` TIs that
	// entered the queue at least threshold ago, with that timestamp, ordered
	// by (queued_at, id) and strictly after the cursor (the zero cursor is the
	// start), at most ReapPageSize rows. The threshold is applied before the
	// page limit, so attempts inside it never take the slots of those past it;
	// the reaper still re-checks it in Go.
	ListStaleQueuedCandidates(ctx context.Context, threshold time.Duration, after ReapCursor) ([]StaleQueuedCandidate, error)
	// MarkTaskDispatchLost transitions one TI to `failed` with
	// error_message='dispatch_lost'. The WHERE state='queued' guard makes
	// this idempotent: a second call on a now-non-queued TI is a no-op. It is
	// pinned to the listed (tryNumber, attemptEpoch) and returns whether a row
	// was actually updated: false means the row moved on between the list and
	// this write (its agent reported RUNNING, or a dispatch claimed a new
	// epoch), so the caller must NOT treat it as reaped (no false log, no pod
	// delete).
	MarkTaskDispatchLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
	// MarkTaskStartExpired fails one queued TI as an infra failure because its
	// pod never started before the attempt credential would run out (#1126).
	// Same guards and return as MarkTaskDispatchLost.
	MarkTaskStartExpired(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
}

DispatchLostReapStore is the slice of scheduler.Store the dispatch-lost reaper needs. The full scheduler.Store embeds this interface so production wires through one type; unit tests fake just this surface.

type Disposition ¶

type Disposition int

Disposition is the typed outcome of a dispatch attempt, returned by Executor.Execute so the scheduler can act on WHY a dispatch failed without reaching into Kubernetes error types itself. Classification lives on the execution layer — the only layer that knows how a given runtime signals backpressure — and travels up the seam as this enum (ADR 0051 Phase 4).

const (
	// Dispatched means the task was handed to the runtime (a pod created, an agent
	// subprocess started). It is not a terminal outcome: the task's real result
	// arrives asynchronously over gRPC. It is the zero value, so a Request that
	// never touches classification reads as a clean hand-off.
	Dispatched Disposition = iota
	// Backpressure is transient cluster backpressure from the Kubernetes apiserver
	// — a ResourceQuota 403 or an API Priority & Fairness 429 — that clears once
	// the cluster has headroom again. The scheduler backs the task off and
	// re-offers it indefinitely, never counting it against the dispatch-attempt
	// budget and never driving the task to dispatch_failed: the cluster asking
	// Dexaflow to slow down is not the user's task failing.
	Backpressure
	// Rejected is a permanent dispatch failure that will not clear on its own: an
	// invalid image, an RBAC denial, an admission-webhook rejection, a bad spec,
	// or any error the classifier does not recognize (including every error a Lite
	// subprocess executor can return). The scheduler keeps the historical
	// bounded-backoff → dispatch_failed behavior (ADR 0031 Amendment A).
	Rejected
	// Deferred means the dispatch was never attempted: the control plane's own
	// buffered dispatch queue was full (or shutting down), or the Kubernetes
	// client's own rate limiter refused the request before sending it. Nothing reached the
	// runtime, so the scheduler leaves the task scheduled for a later tick and
	// records nothing: no backoff, no dispatch-attempt increment, no
	// dispatch_failed. It is local backpressure, not a failed dispatch.
	Deferred
	// Refused is a permanent "this task may not run here" decided before the
	// runtime is called, such as a task larger than its size under an operator
	// resource unit (ADR 0066 §3). A retry cannot change the verdict, so the
	// scheduler fails the task on the first attempt, with the message as its
	// reason, and spends no dispatch retries on it. Unlike Rejected, which also
	// covers errors that clear on retry, Refused is never retried.
	Refused
)

func (d Disposition) String() string

type Executor ¶

type Executor interface {
	Execute(ctx context.Context, req Request) (Disposition, error)
}

Executor runs or dispatches a task. For asynchronous executors (Kubernetes/Docker/subprocess) the return reflects dispatch only, and the agent reports the final state over gRPC. The Disposition tells the scheduler WHY a dispatch failed — transient cluster Backpressure vs a permanent Rejected — so the orchestration layer never has to inspect runtime-specific error types itself (ADR 0051 Phase 4). A successful dispatch returns (Dispatched, nil); a failure returns the classified disposition alongside the non-nil cause (its text feeds the scheduler's note/log).

type HeartbeatReapStore ¶

type HeartbeatReapStore interface {
	// ListAgentLostCandidates returns every `running` TI whose last heartbeat
	// is non-null (it has heartbeated at least once). The reaper applies the
	// threshold per candidate so the SQL stays simple and the decision is
	// purely in Go.
	ListAgentLostCandidates(ctx context.Context) ([]AgentLostCandidate, error)
	// MarkTaskAgentLost transitions one TI to `failed` with
	// error_message='agent_lost'. The WHERE state='running' guard makes this
	// idempotent. It returns whether a row was actually updated: false means a
	// late terminal report transitioned the TI between the list and this write,
	// so the caller must NOT treat it as reaped (no false log, no pod delete).
	MarkTaskAgentLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
	// MarkTaskCredentialCeiling transitions one TI to `failed` with the
	// credential_ceiling reason as a TASK failure (no infra kind), for an
	// attempt that outlived auth.max_attempt_credential_lifetime (#1461). Same
	// guards and return contract as MarkTaskAgentLost.
	MarkTaskCredentialCeiling(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
}

HeartbeatReapStore is the slice of scheduler.Store the TI heartbeat reaper needs. The full scheduler.Store embeds this interface so production wires through one type; unit tests fake just this surface.

type InfraConfirmer ¶ added in v0.5.1

type InfraConfirmer interface {
	ListProvisionalInfraFailures(ctx context.Context) ([]ProvisionalInfraFailure, error)
	ConfirmInfraFailure(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
}

InfraConfirmer is the store surface of the confirmation pass: list the provisional marks, and confirm one, guarded on its exact attempt. The scheduler store implements it.

type InfraOverride ¶ added in v0.5.2

type InfraOverride struct {
	Mark     string
	TenantID string
	DagID    string
	DagRunID string
	TaskID   string
}

InfraOverride describes a durable SUCCESS settled over a reaper's infra mark: the mark it overrode (agent_lost, pod_lost or dispatch_lost) and the attempt's log location, for the system line that records it.

type InfraOverrideRecorder ¶ added in v0.5.2

type InfraOverrideRecorder interface {
	RecordInfraOverride(mark string)
}

InfraOverrideRecorder counts overrides by the mark they overrode.

type InfraOverrideReporter ¶ added in v0.5.2

type InfraOverrideReporter interface {
	SucceedTaskIfActive(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
	SucceedTaskOverInfraMark(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (InfraOverride, bool, error)
}

InfraOverrideReporter is implemented by an OutcomeReporter that can tell a SUCCESS settle that changed no row apart, and settle a durable SUCCESS over a provisional infra mark of the same attempt (ADR 0052 amendment, part 1). The reconciler uses it when the reporter implements it.

type KubernetesExecutor ¶

type KubernetesExecutor struct {
	// contains filtered or unexported fields
}

KubernetesExecutor runs each task as an ephemeral pod (ADR 0002).

func NewKubernetesExecutor(clientset kubernetes.Interface, namespace string) *KubernetesExecutor

func (e *KubernetesExecutor) DeleteRunPods(ctx context.Context, runID string) error

func (e *KubernetesExecutor) DeleteTaskPod(ctx context.Context, a Attempt) error

func (e *KubernetesExecutor) Execute(ctx context.Context, req Request) (Disposition, error)

func (e *KubernetesExecutor) GCStagingClaims(ctx context.Context, ttl time.Duration) error

func (e *KubernetesExecutor) SetPodMetrics(r PodCreationRecorder)

func (e *KubernetesExecutor) SetStagingStore(s StagingStore)

func (e *KubernetesExecutor) SetTeardownRecorder(r DecisionRecorder)

func (e *KubernetesExecutor) TaskPodPending(ctx context.Context, a Attempt) (bool, error)

func (e *KubernetesExecutor) TaskPodPresence(ctx context.Context, a Attempt) (PodPresence, error)

type KubernetesWarmPods ¶

type KubernetesWarmPods struct {
	// contains filtered or unexported fields
}

KubernetesWarmPods is the production WarmPodClient: it lists warm pods by the warm-worker label, builds new ones via the injected spec func + BuildWarmPod, and deletes them, all in one namespace. It owns the label selector so the executor's warm-worker label contract stays private to this package.

func NewKubernetesWarmPods(cs kubernetes.Interface, namespace string, newSpec WarmPodSpecFunc) *KubernetesWarmPods

func (k *KubernetesWarmPods) CreateWarmPod(ctx context.Context, t WarmTarget, anchorName, anchorUID string) (string, error)

func (k *KubernetesWarmPods) DeleteWarmAnchor(ctx context.Context, dagVersionID string) error

func (k *KubernetesWarmPods) DeleteWarmPod(ctx context.Context, name string) error

func (k *KubernetesWarmPods) EnsureWarmAnchor(ctx context.Context, dagVersionID string) (string, error)

func (k *KubernetesWarmPods) LastTerminationReason(podName string) string

func (k *KubernetesWarmPods) ListWarmAnchors(ctx context.Context) ([]string, error)

func (k *KubernetesWarmPods) ListWarmPods(ctx context.Context) ([]WarmPodInfo, error)

type LadderWarning ¶

type LadderWarning struct {
	// Msg is the human-readable sentence: what is off and what it costs.
	Msg string
	// Key is the config key an operator would set to restore the backstop —
	// the attribute an alert rule matches on.
	Key string
	// Value is the offending setting as configured. slog's JSON handler renders
	// a duration as an integer nanosecond count, so an alert rule matches it
	// numerically (value <= 0 for a disabled ceiling), not as "0s" — that
	// spelling is the text handler's.
	Value time.Duration
}

LadderWarning is one boot WARN about a ladder setting that is valid but removes a resilience backstop: the operator-readable sentence plus the two things a monitoring rule needs as FIELDS. Logging only a pre-formatted string leaves the JSON record with nothing but msg, so the single alert an operator would actually want — "some instance booted with the credential ceiling disabled" — can only be written as a substring match on prose that a later reword silently breaks (#924). Msg stays self-contained (it names the key and the value too) so a plain-text log needs no field expansion to be read. Every rung of the ladder is a duration, so Value is one.

func ResilienceLadderWarnings(l ResilienceLadder) []LadderWarning

type OrphanStopper ¶ added in v0.5.1

type OrphanStopper interface {
	// StopOrphanedTask stops the attempt's task when its agent is confirmed
	// dead and the task is verifiably the one that agent started, and reports
	// true once nothing of it is left. false with a nil error means nothing was
	// stopped (the agent is alive, or the task could not be verified) and the
	// attempt must still be treated as alive.
	StopOrphanedTask(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)
}

OrphanStopper is implemented by a ProcessLiveness that can stop a task left running by an agent that died (Lite: the task's process group outlives an agent killed outright, #916). The reapers call it for an attempt that reads alive, before deferring it, so the infra re-place that follows a reap never starts a second copy beside an orphan.

type OutcomeReporter ¶

type OutcomeReporter interface {
	FailTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, reason string) error
	SucceedTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) error
	RescheduleTask(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, at time.Time) error
	// FailTaskInfra records the attempt as failed for an infrastructure reason
	// (last_failure_kind 'infra', confirmed at once: the terminal pod is the
	// confirmation), so the planner re-places it on the infra budget instead of
	// spending a task retry (ADR 0051, #1595).
	FailTaskInfra(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, reason string) error
}

OutcomeReporter records a terminal task-instance outcome the reconciler recovered from a pod (its durable outcome record, or its phase). Every method is guarded by the attempt (try_number and attempt_epoch, from the pod's labels) so a stale reconciler acting on a previous attempt's pod never clobbers a live retry or re-place (#1130), and is idempotent: a settle on an already-terminal or superseded instance is a no-op, not an error.

type PodCreationRecorder ¶ added in v0.5.4

type PodCreationRecorder interface {
	// ObservePodCreation records one pod create for dagID: result is "success"
	// or "error" (ADR 0010), d the API call's latency.
	ObservePodCreation(dagID, result string, d time.Duration)
}

PodCreationRecorder meters task pod creates. observability.Metrics implements it.

type PodIdentity ¶

type PodIdentity struct {
	TaskInstanceID string `json:"ti"`
	TenantID       string `json:"tenant"`
	DagID          string `json:"dag"`
	RunID          string `json:"run"`
	TaskID         string `json:"task"`
	TryNumber      int    `json:"try"`
	// AttemptEpoch is the execution of TryNumber this pod runs (ADR 0051
	// amendment). nil on a pod created before the epoch existed, which the
	// exchange then mints a legacy token for.
	AttemptEpoch *int `json:"epoch,omitempty"`
}

PodIdentity is the JSON payload of AgentIdentityAnnotation: the full task-instance identity the control plane mints the exchanged JWT for.

func ParseAgentIdentity(raw string) (PodIdentity, error)

type PodInformer ¶

type PodInformer struct {
	// contains filtered or unexported fields
}

PodInformer is a shared-informer read-path over task pods, scoped by namespace and the leoflow.io/run-id label (ADR 0002 pods). It replaces the reapers' per-running-TI-per-second apiserver LIST storm and the reconciler's 30s LIST with one long-lived watch feeding a local cache.

Its readings are trusted only in the safe direction (#461): CachedPodActive is consulted ONLY to DEFER a reap when a pod is present and Pending/Running — a cache "absent" reading is never authoritative and callers MUST fall through to the live TaskPodPresence (quorum) read before any destructive action. Cache lag can therefore only ever delay a reap by a tick, never cause a false-positive one. SnapshotTaskPods is safe for the reconciler because presence of a terminal pod is monotonic and every settle is attempt/state-guarded (ADR 0052).

It is constructed only in the scheduler/all role (ADR 0049), never api-only, and is nil in Lite/subprocess (no pods), where consumers keep their live paths.

func NewPodInformer(clientset kubernetes.Interface, namespace string) *PodInformer

func (p *PodInformer) CachedPodActive(a Attempt) bool

func (p *PodInformer) HasSynced() bool

func (p *PodInformer) Shutdown()

func (p *PodInformer) SnapshotTaskPods() ([]*corev1.Pod, error)

func (p *PodInformer) Start(ctx context.Context)

func (p *PodInformer) WaitForCacheSync(ctx context.Context) bool

type PodLostCandidate ¶

type PodLostCandidate struct {
	TaskInstanceID string
	DagRunID       string
	DagID          string
	TaskID         string
	// TryNumber pins the best-effort pod delete to exactly this attempt, so a
	// retry's newer pod is never touched (same invariant as the #474 teardown).
	TryNumber int
	// AttemptEpoch is the epoch the row's current execution was dispatched
	// with (ADR 0051 amendment). The mark and the pod teardown are pinned to it
	// as well as to TryNumber, so a row re-placed or re-dispatched between the
	// list and the write is a different attempt and is left alone.
	AttemptEpoch int
	RunningSince time.Time
	// Heartbeated reports whether the attempt has heartbeated at least once.
	// Pod-lost ignores it; Lite uses it to judge an agent that died before its
	// first heartbeat, which agent-lost never lists (#916).
	Heartbeated bool
	// TenantID locates the attempt's log stream for the killed marker: pod_lost
	// on Kubernetes, and the agent_lost one Lite writes when it fails a
	// never-heartbeated attempt.
	TenantID string
}

PodLostCandidate is one task instance in `running` whose backing pod may have vanished — deleted, evicted, OOM-killed, or lost with its node — before any other reaper could catch it. RunningSince is when the TI entered `running`; the reaper only checks pod liveness once the TI has been running past a grace period, so a just-dispatched TI whose pod is still materializing is never reaped on a transient "no pod yet".

type PodLostReapStore ¶

type PodLostReapStore interface {
	// ListRunningTasks returns one page of the `running` TIs that have been
	// running for at least grace, with the timestamp each entered running,
	// ordered by (started_at, id) and strictly after the cursor (the zero
	// cursor is the start), at most ReapPageSize rows. Warm-pool attempts are
	// excluded: they have no per-task pod, so a presence check would always
	// read them as lost, and the warm-worker-lost reaper owns them. The grace
	// filter is applied before the page limit, so attempts still inside it
	// never take the slots of those the reaper can act on.
	ListRunningTasks(ctx context.Context, grace time.Duration, after ReapCursor) ([]PodLostCandidate, error)
	// MarkTaskPodLost transitions one TI to `failed` with
	// error_message='pod_lost'. The WHERE state='running' guard makes this
	// idempotent. It returns whether a row was actually updated: false means a
	// late terminal report transitioned the TI between the list and this write,
	// so the caller must NOT treat it as reaped (no false log, no pod delete).
	MarkTaskPodLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
}

PodLostReapStore is the slice of scheduler.Store the pod-lost reaper needs. The full scheduler.Store embeds this interface so production wires through one type; unit tests fake just this surface.

type PodManager ¶

type PodManager interface {
	// DeleteTaskPod deletes the pod for exactly one reaped attempt: the
	// (run, task, try, epoch) tuple. Pinning try-number and the attempt epoch
	// guarantees a newer live attempt (a retry on a new try, or an infra
	// re-place on the same try with a new epoch) is never deleted (#901). Each
	// pod is deleted by name with a UID precondition, so a pod the call did not
	// list is never acted on. Tolerates a missing pod.
	//
	// A pod already in a terminal phase is SKIPPED, not deleted (#928): the
	// teardown exists to stop a running container, a terminal pod has none, and
	// its termination message is the outcome record the reconciler settles the
	// attempt from (ADR 0052). That is why a reaper may call this unconditionally
	// after its mark without reading presence first — the guard is at the delete
	// site, so it holds for every reaper including the ones (agent-lost,
	// orphan-run) that read no presence at all.
	DeleteTaskPod(ctx context.Context, a Attempt) error
	// DeleteRunPods deletes every task pod of one reaped run. Used by the
	// orphan-run reaper, which abandons the whole run. The run-id is unique
	// per run, so no other run's pod can match. Tolerates missing pods. The
	// terminal-phase skip above applies per pod within the run (#928).
	DeleteRunPods(ctx context.Context, runID string) error
	// TaskPodPresence reports what the apiserver holds for exactly the
	// (run, task, try, epoch) attempt: a live pod, a present-but-finished pod,
	// or nothing. A reaper defers on a live pod: the dispatch landed, the node
	// is merely slow (#461). Try-number and epoch are pinned so a retried or
	// re-placed TI's liveness gate asks about the attempt it is about to fail,
	// not any older attempt whose pod may still linger (#723).
	TaskPodPresence(ctx context.Context, a Attempt) (PodPresence, error)
	// TaskPodPending reports whether exactly the (run, task, try, epoch)
	// attempt has a Pending pod and no Running one: its agent has not started
	// (image pull, node provisioning, init containers). The dispatch-lost
	// reaper asks it only for a live pod past the start deadline, to tell an
	// agent that can no longer start with a live credential from one that
	// already did (#1126).
	TaskPodPending(ctx context.Context, a Attempt) (bool, error)
}

PodManager is the slice of the Kubernetes executor the reapers use to (1) tear down a reaped task's pod and (2) check whether a queued TI's pod is actually live before declaring its dispatch lost (#474, #461).

Before #474 the reapers only wrote DB state, so a reaped TI's pod kept running to completion — breaking at-most-once execution. The reapers now call these AFTER the durable DB transition so the pod is actually stopped.

It is nil in Lite/subprocess, where tasks are host processes with no pods. Every call site guards the nil: a nil PodManager means "no pods to manage", and the dispatch-lost reaper falls back to its pure time-threshold behavior.

The Kubernetes executor implements this; the interface lives here so the scheduler depends on a capability, not on the executor package.

type PodPresence ¶

type PodPresence int

PodPresence is the three-way answer to "what does the apiserver hold for this attempt's pod?".

The distinction is load-bearing, not cosmetic. A bare bool collapsed the last two states, and the pod-lost reaper read that collapse as authorization to reap: for a task pod that had already finished it marked the task instance pod_lost and then DELETED the pod — destroying the termination-log evidence the reconciler recovers the task's durable outcome from. A finished task then reads as failed, and its run with it. Presence must be three-valued so a reaper can defer on "present but finished" while still reaping on a genuine absence, which is the state pod-lost exists for.

const (
	// PodPresenceLive means a pod for the attempt exists and is Pending or
	// Running: work may still be happening, so every reaper must defer. It is
	// deliberately the zero value — an unset or error-path presence then
	// authorizes nothing destructive.
	PodPresenceLive PodPresence = iota
	// PodPresenceTerminal means a pod for the attempt exists but none of them is
	// Pending or Running. Whatever happened to that attempt is recorded on the
	// pod object, so settling it belongs to the reconciler and no reaper may
	// delete it. Phase Unknown counts here as well: the pod object is still
	// there and it is not an absence. Note the backstop for a pod stuck in
	// Unknown is NOT the reconciler — classifyPod groups Unknown with
	// Pending/Running, so the reconciler neither settles nor collects it — but
	// the agent-lost reaper (heartbeats stop when a node goes unreachable) and
	// the orphan-run reaper. The phase has not been set by kubelet since 2015
	// and is deprecated upstream, so the practical exposure is nil; classifying
	// it as presence is the conservative direction. Note this is deliberately
	// WIDER than terminalForTeardown, which the pod teardown uses: presence asks
	// "is this an absence?" (Unknown is not), teardown asks "is there a container
	// to stop, and will anything else collect this?" (for Unknown, possibly yes
	// and no) — so Unknown defers a reap here and is still deleted there (#928).
	PodPresenceTerminal
	// PodPresenceAbsent means the apiserver holds no pod for the attempt at all
	// — the attempt's pod is genuinely gone (deleted, evicted, lost with its
	// node) and there is nothing left to settle from. This is the only presence
	// that authorizes a pod-lost reap.
	PodPresenceAbsent
)

func (p PodPresence) String() string

type PodPresenceCache ¶

type PodPresenceCache interface {
	// CachedPodActive reports whether the cache holds a Pending/Running pod for
	// exactly the (run, task, try, epoch) attempt. Only a true return is
	// trusted (to defer); false is a "no speedup" signal that must not drive a
	// reap. Try-number and epoch are pinned so the cache gate cannot defer on an
	// older attempt's lingering pod (#723).
	CachedPodActive(a Attempt) bool
}

PodPresenceCache is an optional read-through cache of pod presence — backed by a shared informer (PR-10) — that the pod-lost and dispatch-lost reapers consult ONLY to DEFER a reap, never to authorize one. Its trust is asymmetric (#461):

• CachedPodActive == true => a pod is present and Pending/Running; the reaper may skip the live LIST and defer, because presence is monotonic-safe against cache lag (a pod the cache still shows as live cannot have been gone longer than the lag, and deferring one tick is harmless).
• CachedPodActive == false => NOT authoritative. The reaper MUST fall through to the live TaskPodPresence (quorum) read before any destructive action, so a lagged/cold cache can only ever delay a reap by a tick, never cause a false-positive one.

It exists to remove the O(running-TIs)/sec apiserver LIST storm from the read path while keeping the kill decision on the live read. Nil in Lite/subprocess and before the informer warms: every candidate then uses the live path.

type PodSecurity ¶

type PodSecurity struct {
	// RunAsNonRoot refuses to start a task container whose image resolves to
	// UID 0. It completes the `restricted` set and is on by default: the images
	// Dexaflow ships now satisfy it. runtime/Dockerfile runs as the numeric
	// non-root UID 65532 (`USER 65532:65532` — a name the kubelet cannot resolve
	// is what previously blocked this), and every examples/*/image inherits it.
	// When set, BuildPod also stamps a pod-level fsGroup (nonRootFSGroup) so the
	// per-run staging PVC (ADR 0022) stays writable by that non-root user.
	//
	// It stays a knob rather than a constant so an operator can turn it off for a
	// fleet whose task images legitimately run as root.
	RunAsNonRoot bool

	// ReadOnlyRootFilesystem mounts the container root read-only. Off by
	// default on purpose: `restricted` does not require it, and it breaks
	// ordinary Python tasks that write to /tmp, the pip cache or a matplotlib
	// config dir. Opt in for a task that is known not to write.
	ReadOnlyRootFilesystem bool
}

PodSecurity holds the task-pod hardening knobs whose defaults are behavioral rather than free. Both zero values are the safe choice, so a Request that never touches this struct gets a pod that Pod Security Admission's `restricted` profile admits.

type PodSnapshotter ¶

type PodSnapshotter interface {
	// SnapshotTaskPods returns the managed task pods currently known, or an error
	// (e.g. the cache has not synced) so the reconciler retries next tick rather
	// than treating an unsynced cache as an empty cluster.
	SnapshotTaskPods() ([]*corev1.Pod, error)
}

PodSnapshotter supplies the reconciler's task-pod set from a local cache instead of a live LIST every tick (PR-10). It is safe here without a live confirm: the signal the reconciler acts on is presence of a terminal pod, which is monotonic (a pod that reached Failed/Succeeded stays terminal), and every settle is attempt- and state-guarded (ADR 0052), so at worst cache lag delays a settle by a tick. A nil snapshotter (Lite/subprocess, or a cold start) keeps the live LIST.

type ProcessLiveness ¶ added in v0.5.1

type ProcessLiveness interface {
	// AttemptProcessAlive reports whether the agent process spawned for the
	// (run, task, try) attempt is alive. false with a nil error means no live
	// agent is known for the attempt (it exited, or was never spawned); a
	// non-nil error means liveness is unknown and the caller must defer.
	AttemptProcessAlive(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)
}

ProcessLiveness answers whether the local agent process of one attempt is still alive. It is the Lite (subprocess executor) counterpart of pod liveness: with no pods there is nothing for the reapers to read or delete, but the agent is a host process whose liveness can be checked by PID.

The reapers consult it to defer. A Lite reaper does not stop a live agent the way a pod delete does (it stops the orphaned task of a dead one, see OrphanStopper), so it must never fail an attempt whose agent is still alive as an infra loss: the infra re-place that follows keeps the try number, so a second agent would start on the same attempt while the first one could still get its RUNNING report accepted and run user code (#911). The one exception is an attempt past the credential ceiling (see AttemptStopper), which is failed as a task failure, never re-placed, and whose task is then stopped.

type ProvisionalInfraFailure ¶ added in v0.5.1

type ProvisionalInfraFailure struct {
	TaskInstanceID string
	DagRunID       string
	TaskID         string
	TryNumber      int
	AttemptEpoch   int
}

ProvisionalInfraFailure is one infra mark (agent_lost, pod_lost, dispatch_lost) the reconciler has not confirmed yet (ADR 0052 amendment, part 2), with the attempt it names.

type ReapCandidate ¶

type ReapCandidate struct {
	RunID        string
	DagID        string
	LastActivity time.Time
}

ReapCandidate is one running dag run the reaper is considering, with the timestamp of its most recent observable activity (max of the run's started_at and its task instances' started_at / ended_at). The reaper compares the gap from this stamp to "now" against a stall threshold; a non-zero gap larger than the threshold means the run is orphaned and should be failed.

type ReapCursor ¶ added in v0.5.4

type ReapCursor struct {
	At time.Time
	ID string
}

ReapCursor is a keyset position in a reaper's candidate list: the time the list orders by (started_at for pod-lost, queued_at for dispatch-lost) and the task instance id that breaks ties. The zero cursor is the start of the list.

A list that always returned the oldest ReapPageSize rows let that many live attempts hide a lost one behind them forever (R-07, #1597): every sweep deferred the same live rows and never reached the rest. The reapers now keep a cursor across sweeps and ask for the page after it.

func (c ReapCursor) IsZero() bool

type ReapStore ¶

type ReapStore interface {
	// ListReapCandidates returns every dag_run currently in 'running' state
	// alongside its last-activity timestamp. The query is the authority on what
	// "running" means and how to compute the timestamp; the reaper only decides
	// whether each one has been quiet for too long.
	ListReapCandidates(ctx context.Context) ([]ReapCandidate, error)
	// ReapRun transitions a run to 'failed' with an "orphaned" note and fails
	// any still-active task instances, but only if the run is still orphaned
	// at that moment: the store re-checks the orphan predicate atomically,
	// with quietBefore as the last-activity cutoff, because the list is only a
	// snapshot. It reports false, with no error, when the run is no longer an
	// orphan and nothing was touched. It is idempotent: a second call on the
	// same run reports false.
	ReapRun(ctx context.Context, runID string, quietBefore time.Time) (bool, error)
}

ReapStore is the slice of scheduler.Store the reaper needs. The full scheduler.Store embeds this interface so production wires through one type; the unit tests fake just this surface.

type Reaper ¶

type Reaper struct {
	// contains filtered or unexported fields
}

Reaper is the execution-side backstop that fails stuck runs and task instances the scheduler dispatched but that then went dark. It bundles the five independent reapers behind one ReapOnce entrypoint the leader's maintenance loop drives once per cycle, after the pod reconciler's sweep, so the caller depends on a single capability rather than wiring each reaper itself.

func NewReaper(store ReaperStore, pods PodManager, cache PodPresenceCache, warmPods WarmPodLister, rec DecisionRecorder, logger *slog.Logger, cfg ReaperConfig, inStepDown func() bool) *Reaper

func (r *Reaper) ReapOnce(ctx context.Context) error

func (r *Reaper) SetAttemptLifetimeCeiling(d time.Duration)

func (r *Reaper) SetAttemptTokenTTL(ttl time.Duration)

func (r *Reaper) SetInformerSynced(fn func() bool)

func (r *Reaper) SetLastSweepCompleted(fn func() time.Time)

func (r *Reaper) SetLeaderSince(fn func() time.Time)

func (r *Reaper) SetLeading(fn func() bool)

func (r *Reaper) SetLogSink(s logSink)

func (r *Reaper) SetProcessLiveness(p ProcessLiveness)

type ReaperConfig ¶

type ReaperConfig struct {
	OrphanThreshold       time.Duration
	AgentLostThreshold    time.Duration
	DispatchLostThreshold time.Duration
	// PodLostGrace is the per-task liveness floor measured from the TI's running
	// transition, below which the pod-lost reaper does not consult pod liveness.
	PodLostGrace time.Duration
	// SettlingGrace is the minimum time after this instance acquires leadership
	// before ANY reaper may act — one rung of the leader-settling gate, which
	// also requires a synced pod informer and a completed reconciler sweep. See
	// defaultSettlingGrace and Reaper.settling. Zero disables the whole gate.
	SettlingGrace time.Duration
}

ReaperConfig holds the idle thresholds and the post-leadership settling grace the reapers apply. Zero values are legal but reap aggressively; callers pass DefaultReaperConfig unless a test or load harness deliberately overrides them.

func DefaultReaperConfig() ReaperConfig

type ReaperStore ¶

type ReaperStore interface {
	ReapStore
	HeartbeatReapStore
	DispatchLostReapStore
	PodLostReapStore
	WarmWorkerLostReapStore
}

ReaperStore is the full store surface the five reapers need, composed from each reaper's own slice. The metadatabase-backed SchedulerStore satisfies all five, so production wires through one type; a unit test fakes just the slice its reaper touches.

type Reconciler ¶

type Reconciler struct {
	// contains filtered or unexported fields
}

Reconciler detects task pods whose task instance was never settled by the agent (a pod killed before or during its report) and records the true outcome — from the pod's durable outcome record where present, else its phase — so retries and run finalization proceed instead of stranding the task. It also garbage-collects finished pods once they age out.

Outcome recovery is best-effort (ADR 0052 is a School-A optimization on the School-B re-drive floor): the settle guards on the active states. If a reaper marked the attempt as an infra failure first (agent_lost, pod_lost, dispatch_lost), a durable SUCCESS record still wins while the mark is provisional (ADR 0052 amendment): the reconciler falls back to SucceedTaskOverInfraMark, which settles success only for that exact attempt, before the mark is confirmed or the planner's confirmation valve opens, on a running run. Past that window the success is dropped and the task degrades to the safe retry path, a correctness-safe re-run, not a wrong result. A FAILED record never overrides a mark: the reap's own teardown makes the agent write one.

func NewReconciler(clientset kubernetes.Interface, namespace string, reporter OutcomeReporter) *Reconciler

func (r *Reconciler) LastSweepCompletedAt() time.Time

func (r *Reconciler) Reconcile(ctx context.Context) error

func (r *Reconciler) SetInfraConfirmer(c InfraConfirmer)

func (r *Reconciler) SetInfraOverrideObservers(rec InfraOverrideRecorder, sink logSink)

func (r *Reconciler) SetPodSnapshotter(s PodSnapshotter)

func (r *Reconciler) SetSettledRunCollection(c SettledRunChecker)

type Request ¶

type Request struct {
	TaskInstanceID string
	TenantID       string
	DagID          string
	RunID          string
	TaskID         string
	TryNumber      int
	// AttemptEpoch is the execution of TryNumber the dispatcher claimed for this
	// request (ADR 0051 amendment). It is stamped into the pod's identity
	// annotation so the exchanged token names this execution.
	AttemptEpoch int

	Image           string
	ImagePullPolicy string
	Operator        string
	Entrypoint      string
	Env             map[string]string
	Resources       domain.Resources
	Execution       domain.Execution
	// DefaultServiceAccount is the operator's default task ServiceAccount, the
	// one the dispatcher fills into Execution.ServiceAccount when the task names
	// none. The execution allowlist does not govern it, so the executor's
	// re-check of the allowlist skips exactly this account and checks any other
	// (issue #537 C2).
	DefaultServiceAccount string
	TimeoutSeconds        int
	// AttemptLifetimeCeilingSeconds is the operator's attempt credential ceiling
	// (auth.max_attempt_credential_lifetime) in whole seconds. The Kubernetes
	// executor floors the task pod's ActiveDeadlineSeconds with it when the task
	// declares no TimeoutSeconds, so no task pod is immortal: the agent's reports
	// retry for as long as the control plane is unreachable, and without a floor
	// a total control-plane outage would leave every such pod Running forever,
	// holding its requests and blocking node scale-down. Past the ceiling the pod
	// can do nothing useful — heartbeat renewal stops and its bearer lapses. A
	// user-declared TimeoutSeconds always wins, even when longer. Non-positive
	// means "no ceiling" (the same convention as the warm worker's attempt
	// watchdog for the same knob) and applies no floor. Ignored by the subprocess
	// executor, which has no pod.
	AttemptLifetimeCeilingSeconds int64

	// PodSecurity carries the two hardening choices that can change how a task
	// runs. Everything else BuildPod applies is unconditional, because dropping
	// capabilities, blocking privilege escalation and setting a seccomp profile
	// cost a normal task nothing. Zero value is the secure default.
	PodSecurity PodSecurity

	// ExecutionPolicy is the operator's allowlist of Execution fields (issue
	// #629). The dispatcher refuses a task that breaks it before building the
	// request; the Kubernetes executor checks it again before it builds the pod,
	// so no path that builds a Request can place a pod the operator did not
	// allow. Nil allows everything.
	ExecutionPolicy *domain.ExecutionPolicy

	// Source is the dag.py text captured at compile time. The SubprocessExecutor
	// materializes it to a per-TI temp dir so `python -m leoflow_runtime
	// dag:<task>` can importlib it from there — this is how multi-DAG Lite setups
	// avoid the ModuleNotFoundError that hit Lima 2026-06-01 when the agent's
	// global workdir didn't carry the user's dag.py. The K8s executor ignores it
	// unless SourceMode is set (the container image carries the source).
	Source string
	// SourceMode tells the K8s executor to ship Source into the pod as dag.py
	// (ADR 0067 §3). The dispatcher sets it only for a version on the operator's
	// runtime image that carries a source. Ignored by the subprocess executor,
	// which always materializes Source.
	SourceMode bool

	// Agent connection details injected into the worker environment.
	ControlPlaneAddr string
	AgentToken       string

	// AgentTokenTransport selects how the agent's bearer credential reaches the
	// pod (ADR 0055 Fix #3): "" / "envvar" (the default) sets AgentToken as a
	// plaintext LEOFLOW_AGENT_TOKEN env var — today's behavior, byte-identical;
	// "exchange" keeps the plaintext token OFF the pod object and instead projects
	// a ServiceAccount token the agent exchanges for a task-scoped JWT. Ignored by
	// the subprocess executor (Lite has no pod/SA).
	AgentTokenTransport string
	// AgentTokenAudience is the audience of the projected ServiceAccount token
	// under the exchange transport (the control plane's audience). Empty falls back
	// to the default control-plane audience.
	AgentTokenAudience string
	// AgentTokenExpirationSeconds is the projected token's expiration under the
	// exchange transport. Floored to a safe minimum so a very short task's
	// bootstrap token has not already expired at exchange time.
	AgentTokenExpirationSeconds int64
	// AgentTokenSecretName / AgentTokenSecretKey select the SecretKeyRef fallback
	// for the exchange transport (a cluster that cannot project an SA token): when
	// AgentTokenSecretName is set, LEOFLOW_AGENT_TOKEN is sourced from that Secret
	// via SecretKeyRef rather than projected — still off the plaintext pod spec.
	AgentTokenSecretName string
	AgentTokenSecretKey  string

	// StagingClaim, when set, is the name of the per-DAG-run RWX PVC mounted at
	// /staging in the task pod for large intermediate data shared across the run
	// (ADR 0022). Empty means no staging volume. StagingSize/StagingStorageClass
	// are used to provision the claim on first use.
	StagingClaim        string
	StagingSize         string
	StagingStorageClass string
	// StagingAccessMode is the PVC access mode (default ReadWriteMany; single-node
	// dev uses ReadWriteOnce). Empty means ReadWriteMany.
	StagingAccessMode string

	// AgentTLSCAConfigMap, when set, is the name of a ConfigMap holding the CA
	// (key ca.crt) the agent uses to verify the control plane's gRPC TLS cert
	// (issue #58). It is mounted into the task pod and selects TLS for the agent.
	AgentTLSCAConfigMap string

	// TaskSecretName, when set, is a Kubernetes Secret mounted read-only into the
	// task pod at TaskSecretMountPath. It carries a credential a task references by
	// path (e.g. a GCP service-account key via the connection's key_path), keeping
	// the key in the cluster's secret store rather than in Dexaflow (ADR 0035).
	TaskSecretName      string
	TaskSecretMountPath string

	// SecretsBackend / SecretsBackendKwargs, when set, are the operator's external
	// secrets backend (ADR 0060): the provider class the in-pod resolver drives and
	// its raw kwargs JSON. Injected as LEOFLOW_SECRETS_BACKEND[_KWARGS] pod env, in
	// the leoflow-owned group (operator-sourced — an author's task env cannot set
	// LEOFLOW_ keys, #828). Empty = no external backend (chain stays vault-only).
	SecretsBackend       string
	SecretsBackendKwargs string
}

Request bundles everything an executor needs to run a single task instance.

type ResilienceLadder ¶

type ResilienceLadder struct {
	HeartbeatInterval  time.Duration
	AgentLostThreshold time.Duration
	// SettlingGrace is the post-leadership window during which no reaper fires
	// (ReaperConfig.SettlingGrace); the one grace every reaper shares.
	SettlingGrace   time.Duration
	AttemptTokenTTL time.Duration
	// ReconcileInterval is the period of the leader's maintenance loop: one
	// reconciler sweep followed by one reaper pass per cycle.
	ReconcileInterval time.Duration
	// OrphanThreshold is how long a running dag run may sit with no activity
	// before the orphan-run reaper fails it (ReaperConfig.OrphanThreshold).
	OrphanThreshold time.Duration
	// InfraReplaceMaxDelay is the longest the scheduler may park an infra-failed
	// task before re-placing it: the backoff before the final permitted re-place
	// plus the whole de-synchronizing jitter window. The scheduler owns and
	// computes it; it is passed in because the scheduler package depends on this
	// one, so the validator cannot read it directly.
	InfraReplaceMaxDelay time.Duration
	// MaxAttemptCredentialLifetime is the operator's auth.max_attempt_credential_lifetime:
	// the ceiling on how long heartbeat renewal keeps an attempt's bearer alive.
	// A non-positive value is the documented "no ceiling" setting; the rung that
	// depends on it is then trivially satisfied and skipped.
	MaxAttemptCredentialLifetime time.Duration
	// TaskTerminationGrace is the default termination grace of a task pod: how
	// long a pod stopped in place may take for its task container to exit.
	TaskTerminationGrace time.Duration
	// InfraConfirmMaxWait is the scheduler's infra confirmation valve
	// (scheduler.InfraConfirmMaxWait), passed in for the same reason as
	// InfraReplaceMaxDelay.
	InfraConfirmMaxWait time.Duration
	// DispatchLostThreshold is how long an attempt may sit queued before the
	// dispatch-lost reaper looks at it (ReaperConfig.DispatchLostThreshold).
	DispatchLostThreshold time.Duration
}

ResilienceLadder is the set of timing knobs whose relative ORDER the control-plane-restart recovery depends on. Each is owned by a different package (agent, executor, scheduler, server main, operator config) and tuned for its own reason, so nothing but this type states that they must line up. The invariants:

heartbeat interval  <  agent-lost threshold  <  settling grace  <  attempt token TTL
2 × maintenance interval  <  settling grace
longest infra re-place delay  <  orphan threshold
task termination grace + 2 × maintenance interval  <  infra confirmation valve  <  orphan threshold
attempt token TTL  <  max attempt credential lifetime   (when the ceiling is enabled)
dispatch-lost threshold  <  attempt token TTL - start credential margin
2 × maintenance interval + heartbeat interval  <  start credential margin

Why each rung matters:

• heartbeat < threshold: a healthy agent must beat well inside the silence window or the agent-lost reaper fails live tasks.
• threshold < settling grace: after a (re-)election the reapers wait longer than the silence they punish, so the whole fleet can re-heartbeat.
• settling grace < token TTL: the grace ends while a re-heartbeat can still authenticate and renew the bearer; otherwise the fleet is unreapable AND unable to report — every in-flight task is lost.
• 2×maintenance < settling grace: the leader's maintenance loop runs the reconciler's sweep and then the reapers as ONE ordered cycle, so a reap structurally follows a completed sweep — the settling gate additionally requires that a sweep completed under the new leader. This rung is what makes the grace meaningful on top of that ordering: at least two whole cycles fit inside it, so a settle that failed transiently on the first sweep (a DB hiccup) is retried before the gate can open, and the liveness valve at 2×grace has seen at least four cycles before it declares the sweep broken. A single cycle is not enough: the first tick under a new leader may land anywhere in the interval, so "one interval below the grace" can mean zero completed cycles.
• infra re-place delay < orphan threshold: a run whose only live task is parked in its longest infra re-place backoff has no activity to show the orphan-run reaper; the threshold must outlast that parking or the reaper eats a run that is still recovering from the very fault being retried.
• termination grace + 2×maintenance < infra confirmation valve (ADR 0052 amendment): a reaped pod is stopped in place, its task container exits within the termination grace, and the reconciler needs a sweep after that to read the record and confirm or override the mark. A valve that opens sooner lets the planner re-place on the guess before the evidence could be read. A DAG that declares a longer termination grace gets the valve, not the record, for that task.
• infra confirmation valve < orphan threshold: a run whose only live task is a provisional mark shows no activity to the orphan-run reaper.
• token TTL < credential lifetime: heartbeat renewal keeps an attempt's bearer alive only while the attempt is younger than the ceiling. A ceiling below the TTL means the first renewal is already refused, the bearer lapses at the TTL, and every task longer than one TTL is unreapable AND unable to report — the restart recovery is silently disabled. This is the ONLY operator-tunable rung; every other value is a build-time constant.
• dispatch-lost threshold < token TTL - start margin (#1126): a pod still Pending as its dispatch-minted credential nears its end is re-placed by the dispatch-lost reaper, which only considers attempts queued past its threshold; a threshold past the start deadline would leave the pod to start with a dead credential and spend a user retry.
• 2×maintenance + heartbeat < start margin: a pod that starts just after a reaper pass saw it inside the deadline is not re-placed, so its agent must register and reach its first heartbeat, which renews the credential, before the credential lapses.

Warm pools share the same ladder: a warm worker's per-attempt bearer is renewed by the same heartbeat under the same ceiling, and the warm-worker-lost reaper reuses the pod-lost mark, so no warm-specific rung exists.

type RunRef ¶ added in v0.5.1

type RunRef struct {
	Tenant string
	Run    string
}

RunRef names a run in its tenant, as the pod labels carry them.

type ServiceAccounts ¶ added in v0.5.5

type ServiceAccounts struct {
	// contains filtered or unexported fields
}

ServiceAccounts reads the ServiceAccounts of the namespace task pods run in, for the register dry run (#1889, ADR 0085): whether the account a task pod would run as exists, and how many pull secrets it carries. It reads the account only, never the Secrets it names, with the one `get` on serviceaccounts the chart's Role grants.

func NewServiceAccounts(clientset kubernetes.Interface, namespace string) *ServiceAccounts

func (s *ServiceAccounts) ImagePullSecrets(ctx context.Context, name string) (count int, found bool, err error)

type SettledRunChecker ¶ added in v0.5.1

type SettledRunChecker interface {
	SettledRuns(ctx context.Context, refs []RunRef) (map[RunRef]bool, error)
}

SettledRunChecker reports which of the given runs are settled in their tenant. Runs it leaves out are not collected.

type StagingStore ¶

type StagingStore interface {
	RecordStagingVolume(ctx context.Context, tenantID, dagID, runID, pvcName, size string) error
	MarkStagingDeleted(ctx context.Context, pvcName, reason string) error
	ListActiveStagingVolumes(ctx context.Context) ([]domain.StagingVolumeState, error)
}

StagingStore persists the per-run staging-volume lifecycle in the metadatabase (ADR 0022): provisioning records an active row, GC marks it deleted with a reason, and GC reads the active set joined with each run's state. Identified by the deterministic PVC name (unique per namespace).

type StaleQueuedCandidate ¶

type StaleQueuedCandidate struct {
	TaskInstanceID string
	DagRunID       string
	// TenantID names the attempt's log stream for a start_expired marker.
	TenantID string
	DagID    string
	TaskID   string
	// TryNumber is the attempt the queued row is on, so a best-effort pod
	// delete after the mark targets exactly that attempt's pod (#474).
	TryNumber int
	// AttemptEpoch is the epoch the row's current execution was dispatched
	// with (ADR 0051 amendment). The mark and the pod teardown are pinned to it
	// as well as to TryNumber, so a row re-placed or re-dispatched between the
	// list and the write is a different attempt and is left alone.
	AttemptEpoch int
	QueuedAt     time.Time
	// WarmWorkerID is the warm pod durably bound to this attempt (ADR 0058
	// N1d-a2), or "" for a dedicated task or a warm attempt not yet acked. When
	// set AND the worker is in the live warm-pod set, the dispatch-lost reaper
	// DEFERS: the warm worker holds this attempt and is merely slow to transition
	// queued->running, so failing it would double-run the task (review finding
	// H3). A warm attempt has no task pod, so the existing pod-presence gate
	// cannot protect it — this warm check is what does.
	WarmWorkerID string
}

StaleQueuedCandidate is one task instance in `queued` whose dispatch may have been lost — typically because the scheduler crashed mid-tick between committing the scheduled→queued transition and actually dispatching the TI to an executor. The reaper compares the gap from QueuedAt to "now" against a dispatch-lost threshold; a non-zero gap larger than the threshold means the dispatch is presumed gone and the TI is failed with reason `dispatch_lost`. This unblocks the orphan-run reaper, which keeps stuck runs out of its candidate set as long as any TI looks active (#202).

type SubprocessExecutor ¶

type SubprocessExecutor struct {
	// contains filtered or unexported fields
}

SubprocessExecutor runs the agent as a host subprocess with no isolation. It is for dev mode only and logs a prominent warning on construction.

func NewSubprocessExecutor(agentPath string, logger *slog.Logger) *SubprocessExecutor

func (e *SubprocessExecutor) AttemptProcessAlive(_ context.Context, runID, taskID string, tryNumber int) (bool, error)

func (e *SubprocessExecutor) Execute(ctx context.Context, req Request) (Disposition, error)

func (e *SubprocessExecutor) SetPIDDir(dir string)

func (e *SubprocessExecutor) SetWorkDir(dir string)

func (e *SubprocessExecutor) StopAttempt(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

func (e *SubprocessExecutor) StopOrphanedTask(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

type WarmBoundTI ¶

type WarmBoundTI struct {
	TaskInstanceID string
	DagRunID       string
	TaskID         string
	TryNumber      int
	// AttemptEpoch pins the failover mark to the bound execution (ADR 0051
	// amendment).
	AttemptEpoch int
	WarmWorkerID string
	// TenantID and DagID locate the attempt's log stream for the killed marker
	// (#861).
	TenantID string
	DagID    string
}

WarmBoundTI is one `running` task instance durably bound to a warm worker (ADR 0058 N1d-a2): WarmWorkerID is the warm pod that acked and is serving this attempt. The failover reaper matches WarmWorkerID against the live warm-pod set to find attempts a dead warm pod held.

type WarmPodCache ¶ added in v0.5.4

type WarmPodCache interface {
	CachedWarmPods() ([]WarmPodInfo, bool)
	// BeginCreate is called before a create is issued, EndCreate after it with
	// the created pod's name, or "" when the create failed.
	BeginCreate(t WarmTarget)
	EndCreate(t WarmTarget, podName string)
	CreatesPending(dagVersionID string) bool
	PendingCreatesByTenant() map[string]int
	// ExpectationExpired reports, once, that an expectation timed out without
	// its pod being observed: the cache may be missing pods.
	ExpectationExpired() bool
	// Covers reports whether the cache holds, or saw deleted, every named pod:
	// false means it is behind the last live LIST and must not be trusted.
	Covers(podNames []string) bool
}

WarmPodCache is the reconciler's cached read of the warm fleet (the WarmPodInformer), plus create expectations keyed by the created pod's name: a create the cache has not observed yet holds back further creates for its version and counts toward its tenant's budget, so a lagging cache cannot push a pool past its target, MaxPoolSize or MaxWarmPodsPerTenant.

type WarmPodClient ¶

type WarmPodClient interface {
	ListWarmPods(ctx context.Context) ([]WarmPodInfo, error)
	// CreateWarmPod returns the created pod's name, which keys the event-driven
	// refill's create expectation.
	CreateWarmPod(ctx context.Context, t WarmTarget, anchorName, anchorUID string) (podName string, err error)
	DeleteWarmPod(ctx context.Context, name string) error
	EnsureWarmAnchor(ctx context.Context, dagVersionID string) (uid string, err error)
	DeleteWarmAnchor(ctx context.Context, dagVersionID string) error
	// ListWarmAnchors returns the dag_version of every GC anchor in the
	// namespace, so the reconciler can delete the anchor of an inactive version
	// that never had a pod (every create refused), which the drain never visits.
	ListWarmAnchors(ctx context.Context) ([]string, error)
}

WarmPodClient is the cluster side of warm-pool reconciliation: list the warm fleet, create a new warm worker for a target (which mints the bootstrap token, builds the pod via BuildWarmPod, and Creates it — the auth/config-aware half, wired in main.go), delete one by name, and manage the per-dag-version GC-anchor ConfigMap (ADR 0058 D11). Kept as a narrow seam so the reconciler is unit-tested with a fake and the executor imports neither auth nor config. KubernetesWarmPods is the production implementation.

EnsureWarmAnchor creates (idempotently) the version's anchor ConfigMap and returns its UID; every warm pod created for the version is stamped with an ownerReference to it, so on control-plane loss / namespace teardown the pods are cascade-GC'd. CreateWarmPod threads that anchor name+UID onto the pod. The reconciler ensures the anchor before any create and deletes it (DeleteWarmAnchor) ONLY once the version has fully drained to zero live pods — so the cascade is always a no-op and never kills a live attempt.

type WarmPodInfo ¶

type WarmPodInfo struct {
	Name         string
	DagVersionID string
	Terminal     bool
	TenantID     string
	// TerminationReason is the warm-worker container's terminated reason as the
	// kubelet reported it (for example OOMKilled), or "" while it runs. The
	// warm-worker-lost reaper names an OOM kill from it (#1218).
	TerminationReason string
}

WarmPodInfo identifies one existing warm-worker pod and the dag_version it serves (read from its labels). The reconciler counts these per version.

Terminal marks a warm pod that has reached a terminal phase (Succeeded or Failed). Warm pods are RestartPolicy:Never and the agent has no reconnect, so a crashed/drained/finished worker lingers as a terminal pod that can never serve again. The reconciler must NOT count terminal pods toward the target (or a dead worker never gets replaced) and always reaps them. TenantID is the tenant that owns the dag_version this pod serves, read from the pod's leoflow.io/tenant-id label (M4). The reconciler counts a tenant's live pods by it — including pods of a draining/inactive version, which are absent from the active targets — so the per-tenant cap sees the tenant's whole warm footprint. A pre-label pod (rolling upgrade) carries "" here; the reconciler attributes it via its version when resolvable and NEVER deletes it for the cap.

type WarmPodInformer ¶ added in v0.5.4

type WarmPodInformer struct {
	// contains filtered or unexported fields
}

WarmPodInformer is a dedicated shared-informer read path over the warm worker fleet (pods labeled leoflow.io/warm-worker=true), which the task-pod PodInformer does not cover: it selects on the run-id label warm pods do not carry. It replaces the warm-pool reconciler's per-tick LIST and drives event-driven refill: losing a worker (deleted, or reaching a terminal phase) signals Changes so the reconciler replaces it at once rather than on its next tick.

Its readings are trusted only where staleness is harmless (ADR 0058, #461): the reconciler uses the cache to decide what to CREATE and which idle workers to trim, and confirms with a live LIST before the one cascading delete it performs (a drained version's GC anchor). The failover reapers keep their own live LIST; nothing here authorizes failing an attempt.

func NewWarmPodInformer(clientset kubernetes.Interface, namespace string) (*WarmPodInformer, error)

func (w *WarmPodInformer) BeginCreate(t WarmTarget)

func (w *WarmPodInformer) CachedWarmPods() ([]WarmPodInfo, bool)

func (w *WarmPodInformer) Changes() <-chan struct{}

func (w *WarmPodInformer) Covers(names []string) bool

func (w *WarmPodInformer) CreatesPending(dagVersionID string) bool

func (w *WarmPodInformer) EndCreate(t WarmTarget, podName string)

func (w *WarmPodInformer) ExpectationExpired() bool

func (w *WarmPodInformer) Kick()

func (w *WarmPodInformer) PendingCreatesByTenant() map[string]int

func (w *WarmPodInformer) RecordTerminationsTo(k *KubernetesWarmPods)

func (w *WarmPodInformer) Start(ctx context.Context)

func (w *WarmPodInformer) WaitForCacheSync(ctx context.Context) bool

type WarmPodLister ¶

type WarmPodLister interface {
	ListWarmPods(ctx context.Context) ([]WarmPodInfo, error)
}

WarmPodLister is the narrow read-only seam onto the live warm fleet the failover paths need: just the warm-pod LIST, no create/delete. Both the warm-worker-lost reaper and the dispatch-lost reaper's H3 defer depend on this capability rather than the full WarmPodClient, so a unit test fakes only the LIST. KubernetesWarmPods (via WarmPodClient) already satisfies it — production reuses that one type, it is not duplicated.

type WarmPodSpec ¶

type WarmPodSpec struct {
	DagVersionID    string
	Image           string
	ImagePullPolicy string
	Namespace       string

	// TenantID is the tenant that owns this dag_version. It is stamped onto the pod
	// as the leoflow.io/tenant-id label so the reconciler can attribute the pod to
	// its tenant for the per-tenant aggregate warm-pod cap (M4), even after the
	// version stops being active (the label outlives the active target).
	TenantID string

	// ControlPlaneAddr / AgentTLSCAConfigMap mirror the task-pod connection knobs:
	// where the agent dials and (when set) the CA ConfigMap it verifies the server
	// cert against. Reused verbatim so a warm worker connects exactly as a task pod.
	ControlPlaneAddr    string
	AgentTLSCAConfigMap string

	// ServiceAccount is the ServiceAccount the warm pod runs as — the operator's
	// default task ServiceAccount (executor.task_service_account). A warm worker is
	// pre-created and generic, so it can only carry the operator-wide default, not a
	// per-DAG execution.service_account; empty leaves it on the namespace default SA.
	// Without this a task placed on a warm worker runs as the default SA and keyless
	// secret resolution breaks (#2, warm-pool path).
	ServiceAccount string

	// Bootstrap-credential transport (ADR 0055 Fix #3), mirroring Request's token
	// fields: BootstrapToken rides plaintext on the env-var default; the exchange
	// transport keeps it off the pod and projects a ServiceAccount token instead.
	// The credential authorizes only Register + AwaitAssignment (no secret access —
	// secrets resolve per-attempt against the in-band attempt token), so a warm
	// worker never stamps a task-instance identity annotation.
	BootstrapToken              string
	AgentTokenTransport         string
	AgentTokenAudience          string
	AgentTokenExpirationSeconds int64
	AgentTokenSecretName        string
	AgentTokenSecretKey         string

	// Self-lifecycle caps the warm agent enforces on ITSELF (ADR 0058 D9/D10/D6/H3),
	// carried in-band as env so a worker can drain, idle-recycle, and hard-bound a
	// wedged attempt without any control-plane round trip. Each is zero when the
	// operator disables that bound; the agent treats zero/unset as "no bound".
	//
	//   - MaxAttemptsPerWorker: drain after this many completed attempts (D9/D10).
	//   - MaxWorkerLifetimeSeconds: drain past this wall-clock age (D9/D10).
	//   - WorkerIdleTTLSeconds: idle-recycle after this long awaiting work (D6).
	//   - AttemptWatchdogSeconds: hard per-attempt ceiling, independent of the task's
	//     execution_timeout, so a no-timeout wedge cannot pin the slot (H3). main.go
	//     sets it to auth.max_attempt_credential_lifetime — an attempt can never
	//     validly outlive its credential ceiling.
	MaxAttemptsPerWorker     int
	MaxWorkerLifetimeSeconds int64
	WorkerIdleTTLSeconds     int64
	AttemptWatchdogSeconds   int64

	// PodSecurity carries the same container/pod hardening choices as a task pod.
	PodSecurity PodSecurity

	// Resources sizes the warm container. With an operator resource unit it is
	// one unit (ADR 0066 §3), and only tasks of that size are placed on the
	// worker. Nil leaves the container unsized, as before.
	Resources *domain.Resources

	// ReadOnlyRootFilesystem is the warm isolation mode (X3.2,
	// execution.warm_read_only_root_filesystem). It forces a read-only root on the
	// warm container whatever PodSecurity says, mounts the writable /tmp emptyDir,
	// and tells the agent (warmAttemptHomeEnv) to give each attempt its own HOME
	// and XDG dirs inside the scratch it wipes, and to sweep /tmp and /dev/shm
	// around every attempt. On a writable root a file one attempt plants on the image would be
	// executed by the next attempt on this worker.
	ReadOnlyRootFilesystem bool

	// AnchorName / AnchorUID identify the per-dag-version GC-anchor ConfigMap this
	// warm pod is owned by (ADR 0058 D11). When BOTH are set, BuildWarmPod stamps an
	// ownerReference to the anchor, so on control-plane loss / namespace teardown the
	// pod is cascade-GC'd with the anchor — the one orphan class the reconciler (as
	// deleter) cannot cover. When either is empty (off-cluster/pre-anchor builds and
	// tests) the pod is built bare, exactly as before D11.
	AnchorName string
	AnchorUID  types.UID

	// Labels / Annotations are operator-declared metadata overlaid onto the pod;
	// keys under domain.ReservedMetadataPrefix are dropped and Dexaflow's own
	// warm-worker labels always win a collision (see mergeMetadata).
	Labels      map[string]string
	Annotations map[string]string
}

WarmPodSpec is everything BuildWarmPod needs to build one long-lived warm-worker pod: which dag_version pool it serves, the image (the DAG's image — a warm worker runs the agent in warm mode and forks a child per attempt), the control-plane connection, and how its BOOTSTRAP credential reaches it (the same transport a task pod uses). It carries NO task instance: the per-attempt token and task identity arrive in-band over AwaitAssignment, never on this spec.

type WarmPodSpecFunc ¶

type WarmPodSpecFunc func(t WarmTarget) (WarmPodSpec, error)

WarmPodSpecFunc mints the per-pod bootstrap credential and fills the transport / connection / security fields for a warm worker of the target, returning a spec ready for BuildWarmPod. It is the auth- and config-aware half of warm-pod creation, injected from main.go so the executor package imports neither auth nor config. It is called once per warm worker the reconciler needs to create.

type WarmPoolReconciler ¶

type WarmPoolReconciler struct {
	// contains filtered or unexported fields
}

WarmPoolReconciler maintains the IDLE warm-worker buffer per active dag_version (ADR 0058 N1b2b + N1d-b, model A2). Each tick it reads the active targets, the existing warm fleet, and the busy set (pods serving a running attempt), then per version partitions the live pods into BUSY and IDLE and:

• CREATEs enough workers to restore EffectiveMinIdle IDLE workers, capped by MaxPoolSize (the total ceiling) — so the pool grows past min_idle under load and never exceeds the cap;
• DELETEs only EXCESS IDLE workers (idle over the target), never a busy one, so a scale-down or a drain never kills an in-flight attempt (review M1). Each delete re-reads the busy set first, and every delete of the tick runs before any create, so a worker that takes an attempt during the tick is kept;
• reaps terminal pods unconditionally (H1);
• drains a no-longer-active version (target 0) down to its idle workers + terminal pods, LEAVING busy workers to finish (they are deleted a later tick once idle).

It is leader-gated (run on a gated ticker, like the pod reconciler) so at replicaCount>1 only the leader mutates the fleet. Idempotent (it reconciles to the idle target, so re-running converges), panic-safe and per-version isolated (one bad dag_version never blocks the others), and O(active versions) per tick. It is only constructed and started when warm pools are enabled; with them off the reconciler never runs, every warm pool stays empty, and dispatch is byte-for-byte today's dedicated pod-per-task.

The busy set is a hard safety input: without it every worker looks idle and a busy worker could be deleted, so a nil busy source or a busy-list error aborts the tick with zero mutations (do-no-harm).

GC anchor (D11): before creating any pod for a version the reconciler ensures a per-dag-version anchor ConfigMap and stamps every warm pod with an ownerReference to it, so on control-plane loss / namespace teardown the fleet is cascade-GC'd. The anchor is create-only during a version's active life and deleted only once an inactive version has fully drained to zero pods (the footgun guard in Reconcile), so the cascade never kills a live attempt.

func NewWarmPoolReconciler(targets WarmTargetSource, pods WarmPodClient, busy BusyWarmWorkerSource, maxWarmPodsPerTenant int, logger *slog.Logger, rec DecisionRecorder) *WarmPoolReconciler

func (r *WarmPoolReconciler) Reconcile(ctx context.Context) error

func (r *WarmPoolReconciler) SetCache(c WarmPodCache, parallelism int)

type WarmTarget ¶

type WarmTarget struct {
	DagVersionID     string
	DagID            string
	Image            string
	EffectiveMinIdle int
	MaxPoolSize      int
	TenantID         string
	Prewarm          bool
	// contains filtered or unexported fields
}

WarmTarget is one active dag_version the warm-pool reconciler keeps warm workers ready for (model A2, ADR 0058 N1d-b). Image is the DAG's image the warm worker runs the agent from.

EffectiveMinIdle is the number of IDLE workers to keep ready for this version — the warm buffer, already resolved through the clamp/fallback/flag gate (see config.ExecutionSection.EffectiveMinIdle). It is NOT a total pool size: busy workers do not count against it, so under load the pool grows past min_idle to keep the idle buffer available.

MaxPoolSize is the TOTAL ceiling of live workers (idle + busy) the version may hold at once (config.ExecutionSection.MaxPoolSize). The reconciler never creates past it, so the pool breathes between EffectiveMinIdle-idle + busy and this cap. The N1a boot validation guarantees max_pool_size >= 1 when warm pools are on and EffectiveMinIdle is clamped to it upstream, so MaxPoolSize >= EffectiveMinIdle in practice; the reconciler still handles MaxPoolSize < EffectiveMinIdle defensively by taking the effective ceiling as max(EffectiveMinIdle, MaxPoolSize) — the idle buffer must always be creatable. TenantID is the tenant that owns this dag_version, threaded through so the per-tenant aggregate warm-pod cap (M4) can sum a tenant's promised idle floors and ration its shared budget across versions. It comes from the active run's tenant_id (or the DAG's, for a prewarm target); the reconciler groups targets and live pods by it.

Prewarm marks a target held only because the version is its DAG's current version (#1707, ADR 0082): no active run needs it yet, and its pool is started so the first run finds warm workers. Under a per-tenant cap its floor shares what the cap leaves and never raises the budget (boundPrewarm); otherwise the reconciler treats it like any other target and only logs and meters when a prewarm pool starts.

DagID is the DAG's id, which orders a tenant's prewarm targets when the per-tenant cap cannot cover all of their floors, so the same DAGs keep their pools from one reconcile to the next.

type WarmTargetSource ¶

type WarmTargetSource interface {
	ActiveWarmTargets(ctx context.Context) ([]WarmTarget, error)
}

WarmTargetSource yields the dag_versions to keep warm and their effective warm target: the current version of each unpaused DAG that declares min_idle_workers, and the versions of active runs that are still current (ADR 0082). A version absent from it drains. It is implemented on the storage/scheduler side (it reads active runs, the DAGs' current versions and their cached specs and applies the operator's clamp/fallback), and is defined HERE so the executor's reconciler depends on a narrow capability rather than importing the scheduler or storage package (which would be a dependency cycle: storage already imports executor).

type WarmTerminationLookup ¶ added in v0.5.5

type WarmTerminationLookup interface {
	LastTerminationReason(podName string) string
}

WarmTerminationLookup is the optional seam a WarmPodLister offers to name the cause of a worker the LIST no longer returns: the warm-worker container's terminated reason it last observed for that pod, or "" (#1835). KubernetesWarmPods implements it.

type WarmWorkerLostReapStore ¶

type WarmWorkerLostReapStore interface {
	ListWarmBoundRunningTIs(ctx context.Context) ([]WarmBoundTI, error)
	// MarkTaskPodLost is reused verbatim from the pod-lost reaper: a warm worker
	// vanishing is the same infra failure as a task pod vanishing, so it routes
	// through the same infra path with the same idempotency guard (WHERE
	// state='running'). applied=false means a late settle raced the mark — a
	// benign skip, never a false reap.
	MarkTaskPodLost(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
	// MarkTaskPodLostOOMKilled is MarkTaskPodLost, the same infra mark with the
	// same guards, for an attempt whose warm worker's container the kubelet
	// reported OOMKilled (#1218): only the failure message differs, naming the
	// OOM kill. The attribution stays infra (ADR 0051, ADR 0058 D7).
	MarkTaskPodLostOOMKilled(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)
}

WarmWorkerLostReapStore is the slice of the store the warm-worker-lost reaper needs: list the warm-bound running TIs, and reuse the pod-lost mark to route a lost attempt to infra (bumps infra_attempts, NOT try_number). The full scheduler store satisfies it; a unit test fakes just this surface.