Help us shape the future of Go! Take the 2026 Go Developer Survey

domain

package
v0.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 8, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

Package domain defines the core Dexaflow types (DAG, Task, project config) and validates them against the canonical JSON Schemas in docs/api.

Index

Constants

View Source 
const (
	// UnitEnforceRefuse refuses a task that does not fit its size: registration
	// fails and a DAG registered earlier fails at dispatch. The default.
	UnitEnforceRefuse = "refuse"
	// UnitEnforceWarn accepts such a task and runs it with its own resources,
	// logging and counting it, so an operator can roll a unit out without
	// failing tasks first.
	UnitEnforceWarn = "warn"
)

Enforcement modes of executor.unit.enforce (ADR 0066 §3).

View Source 
const DefaultPoolName = "default_pool"

DefaultPoolName is the implicit pool a task with no declared pool draws from, so the admission gate is always well-defined. Matches Airflow's default_pool.

View Source 
const DefaultUnitMaxSize = 64

DefaultUnitMaxSize is the largest pool_slots a task may have while a unit is set and executor.unit.max_size is not (ADR 0066 §3).

View Source 
const ReservedMetadataPrefix = "leoflow.io/"

ReservedMetadataPrefix is the label and annotation key prefix Dexaflow owns on the pods it creates. Components decide what a pod is from these keys (a task attempt's identity, a warm worker and the pool it serves, the owning tenant), so a DAG-declared key under it would let a DAG change how its own pod is treated.

Variables

View Source 
var (
	// ErrCyclicDAG reports a task graph with no valid execution order.
	ErrCyclicDAG = errors.New("cyclic task graph")
	// ErrUnknownDependency reports a depends_on entry naming no declared task.
	ErrUnknownDependency = errors.New("unknown task dependency")
	// ErrDuplicateTaskID reports two tasks declaring the same task_id.
	ErrDuplicateTaskID = errors.New("duplicate task_id")
)
View Source 
var ErrConflict = errors.New("resource already exists")

ErrConflict is returned when a write conflicts with an existing resource (e.g. a duplicate dag run for the same logical date). The API maps it to 409.

View Source 
var ErrInvalidDbtProject = errors.New("invalid dbt.project")

ErrInvalidDbtProject reports a dbt.project path the compiler cannot use.

View Source 
var ErrInvalidRunID = errors.New("invalid run_id")

ErrInvalidRunID reports a dag_run_id a caller may not use.

View Source 
var ErrLimitExceeded = errors.New("tenant limit exceeded")

ErrLimitExceeded is returned when a write would take a tenant past one of the limits the operator set for it (TenantLimits). The API maps it to 403: the request is well formed, and repeating it will not succeed until the limit or the tenant's usage changes.

View Source 
var ErrNotFound = errors.New("resource not found")

ErrNotFound is returned when a requested resource does not exist.

View Source 
var ErrSchemaNotCurrent = errors.New("database schema is not current")

ErrSchemaNotCurrent is returned when the database is reachable but its migration state is not one the running binary can serve. It lives here, next to the other cross-cutting sentinels, because two layers need to agree on it: storage produces the verdict, and the health endpoints must tell it apart from "the database could not be read at all". Both answers are a 503, but one sends an operator to the migration Job and the other to the connection pool, and a probe that conflates them costs an on-call hour.

View Source 
var ErrUnknownAlertPlaceholder = errors.New("unknown alert placeholder")

ErrUnknownAlertPlaceholder reports an alert message template referencing a substitution Dexaflow does not perform.

View Source 
var ErrValidation = errors.New("invalid input")

ErrValidation is returned when input fails a business-rule check that the caller can fix (e.g. creating a user with a role that does not exist). The API maps it to 400.

Functions

func IsCronlessSchedule 

func IsCronlessSchedule(expr string) bool

IsCronlessSchedule reports whether expr is empty (manual-only) or a recognized non-cron Airflow schedule. Such a schedule is valid but is never run on a cron, so callers skip cron handling for it without treating it as an error.

func IsOnceSchedule 

func IsOnceSchedule(expr string) bool

IsOnceSchedule reports whether expr is Airflow's "@once" — a DAG that runs exactly one time (on first scheduler sight) and never again.

func IsReservedMetadataKey added in v0.5.2 

func IsReservedMetadataKey(key string) bool

IsReservedMetadataKey reports whether a label or annotation key falls under ReservedMetadataPrefix and therefore may only be set by Dexaflow itself.

func MinScheduleInterval added in v0.5.1 

func MinScheduleInterval(expr string) (time.Duration, bool)

MinScheduleInterval returns the shortest gap between two consecutive fire times of a DAG schedule: a 5-field cron, a preset such as @hourly, or "@every <duration>". ok is false when the schedule never fires on a cron (empty, @once, @continuous, a date that does not exist such as 31 February) or does not parse.

For a cron it is exact over the wall clock. The minutes and hours a cron fires at are the same on every day it fires, so the gaps inside a day are fixed; the only other gaps run from the last time of one firing day to the first time of the next, which is shortest across the closest two firing days. Those days come from a scan of one 28-year calendar cycle that walks firing days only, so even "* * * * *" costs a few thousand steps at most. The scan is in UTC, so the gaps are wall-clock gaps. With CRON_TZ the two days a year the clocks change differ: a gap that spans the change is an hour shorter or longer, a time inside the repeated hour fires twice, one hour apart, and a time inside the skipped hour does not fire that day.

func NewTenantSchemaCompiler added in v0.5.1 

func NewTenantSchemaCompiler() *jsonschema.Compiler

NewTenantSchemaCompiler returns a JSON Schema compiler for a schema a tenant wrote, such as a DAG param schema. Add the schema under TenantSchemaURL so errors never name a local path. It resolves references into the schema itself (its "$defs") and the standard draft meta-schemas the library ships, and refuses every other "$ref": no local file and no remote URL is loaded.

func Safef 

func Safef(class error, format string, a ...any) error

Safef builds a SafeError in class with the formatted message. The message is shown to API clients verbatim, so it must contain only facts the caller supplied or Dexaflow chose — never a driver, filesystem, or network error.

func SupportedPythonVersions 

func SupportedPythonVersions() ([]string, error)

SupportedPythonVersions returns the python_version values the project publishes a task base image for, in schema order (ascending). The slice is a copy: callers may sort or filter it.

This is the one list. scripts/check-python-runtime-matrix.sh asserts that the release matrix, the Makefile target, the runtime Dockerfile and the configuration reference all agree with it, because a published image nobody can select — and a selectable version nobody publishes — both surface as a `docker pull` 404 inside the user's own build.

func TenantSchemaURL added in v0.5.2 

func TenantSchemaURL(name string) string

TenantSchemaURL is the location a tenant schema is added to its compiler under. A bare name such as "param.json" is resolved against the control plane's working directory, so every compile and validation error that names the schema, or a relative "$ref" inside it, would reveal that directory to the tenant. Under this non-file base a relative "$ref" resolves to another dexaflow:// URL, which noRemoteRefs refuses like any other.

func ValidateRunID 

func ValidateRunID(v string) error

ValidateRunID checks a caller-supplied dag_run_id.

The trigger endpoint accepts dag_run_id verbatim from the request body, and a run id ends up as a path segment in the log sink — so a value carrying a separator or a parent reference steers the control plane's own writes outside the log root, with content the caller also controls.

The sink refuses such a value independently; this exists so the request fails as a readable 400 instead of creating a run that appears to work and then silently produces no logs. Two gates, different jobs: this one is UX, the sink's is the security boundary and must never be relaxed to match this.

Separators are banned, punctuation is not: Airflow-generated ids embed an RFC3339 timestamp ("manual__2026-07-30T12:00:00+00:00"), so rejecting ':' or '+' would reject every run the scheduler creates.

Types

type AlertRule 

type AlertRule struct {
	// Type is the channel: "slack" (Slack incoming webhook) or "webhook" (a generic
	// HTTP POST, e.g. PagerDuty/Opsgenie/Teams). Validated by the schema enum.
	Type string `json:"type" yaml:"type"`
	// Conn is the managed Dexaflow connection id holding the endpoint (and secret).
	Conn string `json:"conn" yaml:"conn"`
	// Message is the optional notification body; it is templated at fire time with
	// run context ({{dag}}, {{run_id}}, {{task}}, …). Empty uses a default summary.
	Message string `json:"message,omitempty" yaml:"message,omitempty"`
}

AlertRule is one channel to notify on an alert event. The endpoint and its secret always come from a managed connection (Conn), never a literal URL or token in dexaflow.yaml — that keeps credentials out of the compiled dag.json and mirrors the env-ref secret discipline.

type AlertsConfig 

type AlertsConfig struct {
	// OnFailure lists the rules dispatched when a DagRun reaches failed.
	OnFailure []AlertRule `json:"on_failure,omitempty" yaml:"on_failure,omitempty"`
}

AlertsConfig groups alert rules by the lifecycle event that fires them. Only on_failure is wired today (#424); on_success/on_retry are reserved for a later increment so the surface can grow without a breaking change.

type AuditLogEntry 

type AuditLogEntry struct {
	ID           int64
	When         time.Time
	Action       string
	ResourceType string
	ResourceID   string
	Owner        string
	Extra        string
}

AuditLogEntry is one recorded action against a resource — the source for the UI's Audit Log table. ResourceID carries the DAG id for dag-scoped events.

type BuildConfig 

type BuildConfig struct {
	Dockerfile string            `json:"dockerfile,omitempty" yaml:"dockerfile,omitempty"`
	Context    string            `json:"context,omitempty" yaml:"context,omitempty"`
	Platforms  []string          `json:"platforms,omitempty" yaml:"platforms,omitempty"`
	Labels     map[string]string `json:"labels,omitempty" yaml:"labels,omitempty"`
}

BuildConfig controls how the container image is built from the project.

type ClearOptions 

type ClearOptions struct {
	ResetDagRun        bool
	RunOnLatestVersion bool
}

ClearOptions carries the two independent decisions a clear makes about the run it belongs to.

ResetDagRun re-opens the run so the scheduler looks at it again. Without it a terminal run stays terminal, and the cleared task instance sits at `none` in a run nothing schedules.

RunOnLatestVersion decides WHICH version the re-run executes: the DAG's current one, or the one the run was pinned to when it was created.

These were a single boolean. Folding the version decision into the re-open meant there was no way to re-run a task against the image that produced it — "clear last week's task" always executed today's code — and no way to re-open a run without also moving it forward a version. Apache Airflow separates them the same way, calling the second `run_on_latest_version`.

type ConfigDefaults 

type ConfigDefaults struct {
	Retries                 int `json:"retries,omitempty" yaml:"retries,omitempty"`
	RetryDelaySeconds       int `json:"retry_delay_seconds,omitempty" yaml:"retry_delay_seconds,omitempty"`
	ExecutionTimeoutSeconds int `json:"execution_timeout_seconds,omitempty" yaml:"execution_timeout_seconds,omitempty"`
	// Size is the default pool_slots of every task that sets none, in dag.py or
	// in tasks.<id>.size (ADR 0066).
	Size      *int              `json:"size,omitempty" yaml:"size,omitempty"`
	Resources *DefaultResources `json:"resources,omitempty" yaml:"resources,omitempty"`
	// NodeSelector is the DAG-wide pod placement fallback applied to every task
	// that declares no execution.node_selector of its own. Like Resources it is a
	// default, so the most-specific per-task value always wins. Consumed at
	// compile time by the overlay, which bakes it onto each task in dag.json.
	NodeSelector map[string]string `json:"node_selector,omitempty" yaml:"node_selector,omitempty"`
}

ConfigDefaults holds task defaults applied to every task generated from the project at compile time.

type Connection 

type Connection struct {
	ConnID      string
	ConnType    string
	Host        string
	Schema      string
	Login       string
	Password    string
	Port        *int
	Extra       string
	Description string
}

Connection is an Airflow-style connection: credentials/endpoints for operators, managed from the Admin UI. Password and Extra are encrypted at rest (ADR 0019); Password is write-only and never returned by the API.

func ParseConnectionURI 

func ParseConnectionURI(connID, uri string) (Connection, error)

ParseConnectionURI reads an Airflow connection URI — the AIRFLOW_CONN_<ID> form a task consumes — back into a Connection.

It is the inverse of the URI the control plane delivers to tasks, and it follows Airflow's own from_uri semantics wherever the two could differ, because the value being parsed is one the user wrote for Airflow to read:

  • the scheme carries `-` where the conn_type has `_` (google_cloud_platform and friends are not legal URI schemes), so it is reversed here exactly as the emitter applies it;
  • userinfo is percent-decoded into login and password;
  • the path becomes the schema with ONE leading slash removed, which is what Airflow does — note this is lossy for an absolute path like sqlite's `sqlite:///tmp/db`, whose schema comes back as `tmp/db`. Matching Airflow is the right loss to take: the task resolves the connection through Airflow, so Airflow's reading is the meaning;
  • `__extra__` carries the JSON blob, as the emitter writes it.

No error returned here ever contains the URI: the value is a credential, and an error is the most likely thing to be logged, wrapped or printed. Errors name the connection and the problem.

An empty conn_type is refused. A connection with no type cannot be delivered to a task in a form Airflow will accept, so storing one would replace a missing connection with a broken one.

type ConnectionPatch 

type ConnectionPatch struct {
	ConnID      string
	ConnType    string
	Host        *string
	Schema      *string
	Login       *string
	Password    *string
	Port        *int
	Extra       *string
	Description *string
}

ConnectionPatch is a tri-state write to a Connection (#887). Each nullable field is one of three states the write path must keep distinct:

  • nil pointer -> absent: preserve the stored value (the upsert passes NULL and COALESCE keeps the current column).
  • non-nil "" -> present and empty: clear the field to empty.
  • non-nil value -> present: set the field.

A plain string cannot carry "absent" vs "present and empty", which is why the safe-merge upsert alone (v0.4.4) could neither clear a field nor stop a round-tripped mask from overwriting a secret. ConnType is always written (required on every upsert). Secret-mask handling (the mask means "unchanged") is resolved by the caller before it builds the patch, so the repository only ever sees the three states above.

type DAG 

type DAG struct {
	DagID          string
	Description    string
	Owner          string
	Tags           []string
	Schedule       *string
	ScheduleTZ     string
	StartDate      *time.Time
	IsPaused       bool
	IsActive       bool
	MaxActiveRuns  int
	Catchup        bool
	LastParsedTime *time.Time
	// CurrentVersion is the LABEL of the DAG's current registered version, empty
	// when it cannot be resolved. Paired with DagRun.Version by the UI's clear
	// dialog, which offers "Run with latest bundle version" only when the two
	// differ — see that field.
	CurrentVersion string
}

DAG is a registered DAG with its scheduling metadata (distinct from DAGSpec, which is the compiled artifact).

type DAGSpec 

type DAGSpec struct {
	SchemaVersion string   `json:"schema_version"`
	DagID         string   `json:"dag_id"`
	DagVersion    string   `json:"dag_version"`
	Image         string   `json:"image"`
	Description   string   `json:"description,omitempty"`
	Owner         string   `json:"owner,omitempty"`
	Tags          []string `json:"tags,omitempty"`
	Schedule      *string  `json:"schedule,omitempty"`
	ScheduleTZ    string   `json:"schedule_timezone,omitempty"`
	StartDate     string   `json:"start_date,omitempty"`
	EndDate       *string  `json:"end_date,omitempty"`
	MaxActiveRuns int      `json:"max_active_runs,omitempty"`
	// MaxActiveTasks caps how many of this DAG's task instances may be
	// concurrently non-terminal (queued or running) across all of its active
	// runs — Airflow's per-DAG max_active_tasks (ADR 0053 Stage 1). Zero (the
	// default) means unlimited: a DAG that never sets it, and all of Lite, plans
	// exactly as before. The scheduler enforces it in PlanRun's scheduled→queued
	// admission gate.
	MaxActiveTasks int `json:"max_active_tasks,omitempty"`
	// MinIdleWorkers is a DORMANT seam for per-DAG author-declared warmth: the
	// number of warm workers an author would want kept ready for this DAG version
	// so its tasks skip cold-pod startup (ADR 0058, warm pools model A2). It is NOT
	// author-settable today. There is no author entry point: the field is absent
	// from the authoring schema (leoflow-yaml-schema.json, which is
	// additionalProperties:false) and the parser never emits it, so after the
	// parse→compile path this value is ALWAYS 0. The intended split is that the
	// operator gates IF warmth happens at all (execution.warm_pools_enabled) while
	// the author would only tune HOW MANY, with the operator clamping the request;
	// whether a pod may be reused across attempts stays the operator's security
	// decision. The downstream is intentionally pre-wired around this field —
	// config.ExecutionSection.EffectiveMinIdle already clamps it and the scheduler
	// store already reads it — so exposing it to authors later is a schema+parser
	// change only (add the key to the authoring schema and have the parser emit
	// it), with no domain/scheduler rework. Until then it stays 0 and every DAG,
	// and all of Lite, behaves exactly as before.
	MinIdleWorkers int          `json:"min_idle_workers,omitempty"`
	Catchup        bool         `json:"catchup,omitempty"`
	DefaultArgs    *DefaultArgs `json:"default_args,omitempty"`
	// Staging, when enabled, requests an ephemeral RWX volume shared by the run's
	// tasks at /staging (ADR 0022). nil/disabled means no staging volume.
	Staging *StagingConfig `json:"staging,omitempty"`
	// Alerts declares native on-failure alerting (#424), overlaid from dexaflow.yaml
	// at compile time so the scheduler fires it from the artifact without re-reading
	// the project config. nil means no alerting.
	Alerts *AlertsConfig `json:"alerts,omitempty"`
	// Variables and Connections are the secret names this DAG declares (ADR 0045,
	// ADR 0055). A task receives a variable or connection only if the DAG declared
	// it; TaskSpec may narrow the set further per task. Absent (empty) is always
	// valid and means the DAG declares nothing — the additive, back-compatible
	// default. These carry the declaration only; secret delivery still ships the
	// whole tenant vault until enforcement lands on a later increment.
	Variables   []string `json:"variables,omitempty"`
	Connections []string `json:"connections,omitempty"`
	// Params are the DAG's author-declared run parameters (Airflow's params=),
	// keyed by name. Each carries a Default (materialized into a run's conf when
	// the trigger omits that key) and an optional JSON Schema the trigger-time
	// conf value is validated against. Absent (empty) means the DAG declares no
	// params — the additive, back-compatible default, so the compiled shape of a
	// param-free DAG is unchanged. Part of the immutable spec (CanonicalHash), so
	// changing a default or schema produces a new DAG version.
	Params map[string]ParamSpec `json:"params,omitempty"`
	Tasks  []TaskSpec           `json:"tasks"`
	// Source is the original dag.py text, captured at compile time so the UI's
	// Code tab can show the Python a human wrote (not the compiled spec). It is
	// part of the artifact: changing it produces a new version.
	Source string `json:"source,omitempty"`
}

DAGSpec is the canonical serialized representation of a DAG consumed by the control plane. It mirrors docs/api/dag-schema.json.

func (*DAGSpec) CanonicalHash 

func (d *DAGSpec) CanonicalHash() (string, error)

CanonicalHash returns the SHA-256 of the spec's canonical JSON encoding. Go's struct marshaling is deterministic (fixed field order, sorted map keys), so identical specs hash identically — used to deduplicate DAG versions.

func (*DAGSpec) Validate 

func (d *DAGSpec) Validate() error

Validate checks the DAGSpec against the canonical dag.json schema and returns a joined error describing every schema violation, or nil when valid.

func (*DAGSpec) ValidateSchedule 

func (d *DAGSpec) ValidateSchedule() error

ValidateSchedule checks that a DAG's cron schedule is parseable. An empty or absent schedule (manual-only) and the recognized non-cron Airflow schedules (@once, @continuous) are valid. A malformed cron expression — a 4-field cron, a typo — is rejected here so it fails loudly at compile time; otherwise the scheduler silently can't parse it and the DAG simply never runs, with no error surfaced anywhere (the worst failure mode). The parser is robfig/cron's ParseStandard, the same one the scheduler uses, so what validates here is exactly what the scheduler can run (see scheduler/cron.go).

type DagRun 

type DagRun struct {
	DagID       string
	RunID       string
	LogicalDate time.Time
	State       DagRunState
	RunType     string
	QueuedAt    time.Time
	StartedAt   *time.Time
	EndedAt     *time.Time
	Note        string
	// Conf is the run's configuration as a JSON object, supplied at trigger time
	// and exposed to tasks as params. Empty means no configuration; storage
	// persists the empty-object default so downstream readers never see NULL.
	Conf json.RawMessage
	// Version is the LABEL of the dag_version this run is pinned to (the
	// `dag_version` from its dag.json), empty when it cannot be resolved.
	//
	// It exists for the UI: the SPA's clear dialog renders its "Run with latest
	// bundle version" control only when the run's bundle_version differs from the
	// DAG's, and hides it when either is null. With no label the operator cannot
	// ask for the current version at all, whatever the API accepts.
	Version string
}

DagRun is an execution of a DAG, identified by dag_id + run_id.

type DagRunState 

type DagRunState string

DagRunState is the lifecycle state of a DagRun. The values mirror the dag_run_state enum in the database (migration 003). 

const (
	// DagRunStateQueued means the run has been created but not started.
	DagRunStateQueued DagRunState = "queued"
	// DagRunStateRunning means at least one task instance is active.
	DagRunStateRunning DagRunState = "running"
	// DagRunStateSuccess means every leaf task reached a successful terminal state.
	DagRunStateSuccess DagRunState = "success"
	// DagRunStateFailed means the run finished with at least one failure.
	DagRunStateFailed DagRunState = "failed"
)

DAG run lifecycle states.

func (DagRunState) IsTerminal 

func (s DagRunState) IsTerminal() bool

IsTerminal reports whether the dag run state is final.

type DagStats 

type DagStats struct {
	Active  int
	Failed  int
	Running int
	Queued  int
}

DagStats holds the home dashboard's DAG counters: the number of active DAGs and how many have a latest run in each state.

type DagVersion 

type DagVersion struct {
	ID            string
	VersionNumber int
	CreatedAt     time.Time
	// Version is the deployment label that produced this snapshot: a git describe
	// (tag/SHA) in production, or "dev-<timestamp>" in dev. It is the stable
	// per-deployment identifier under a stable dag_id.
	Version string
}

DagVersion is a registered version of a DAG. VersionNumber is the 1-based ordinal the UI uses (the stored version label is free-form).

type DbtConfig 

type DbtConfig struct {
	// Project is the directory containing dbt_project.yml.
	Project string `json:"project,omitempty" yaml:"project,omitempty"`
	// Granularity is the task partition strategy: node, level or folder (ADR 0042
	// §5). Empty means node. Tag and selector grouping are not implemented — the
	// schema enum is the three above; see #398.
	Granularity string `json:"granularity,omitempty" yaml:"granularity,omitempty"`
	// Manifest optionally points to a pre-built manifest.json (the Pro/CI baked
	// path); empty means run `dbt parse` to generate it at compile time.
	Manifest string `json:"manifest,omitempty" yaml:"manifest,omitempty"`
	// Schedule is the DAG's cron expression or preset (e.g. "@daily",
	// "0 6 * * *"). dbt carries no schedule, so it is declared here; empty means
	// an unscheduled DAG (run on demand).
	Schedule string `json:"schedule,omitempty" yaml:"schedule,omitempty"`
	// Connection is a managed Dexaflow connection id (ADR 0043 #2). When set, the
	// dbt task generates its profiles.yml from the connection delivered to the pod
	// instead of a profiles.yml baked into the image — use one or the other.
	Connection string `json:"connection,omitempty" yaml:"connection,omitempty"`
	// Schema overrides the dbt target schema in the generated profile (where models
	// materialize); empty uses the connection's or dbt's default.
	Schema string `json:"schema,omitempty" yaml:"schema,omitempty"`
}

DbtConfig declares a dbt project as the DAG source (ADR 0042). The compiler reads the project's manifest.json and renders one task per dbt node (or per group), so a dbt project becomes a Dexaflow DAG with no Cosmos or Airflow.

type DefaultArgs 

type DefaultArgs struct {
	Retries                 int `json:"retries,omitempty"`
	RetryDelaySeconds       int `json:"retry_delay_seconds,omitempty"`
	ExecutionTimeoutSeconds int `json:"execution_timeout_seconds,omitempty"`
}

DefaultArgs holds retry and timeout defaults applied to every task in a DAG.

type DefaultResources 

type DefaultResources struct {
	CPU    string `json:"cpu,omitempty" yaml:"cpu,omitempty"`
	Memory string `json:"memory,omitempty" yaml:"memory,omitempty"`
}

DefaultResources expresses default CPU and memory for generated tasks.

func (*DefaultResources) AsResources 

func (d *DefaultResources) AsResources() *Resources

AsResources expands the simplified default cpu/memory into a full Resources with requests == limits (the QoS story of #725). This mirrors how the per-cluster platform default is built at dispatch. Returns nil when the receiver is nil or declares no quantity, so callers can treat a missing default as "leave the task untouched".

Guaranteed QoS is reached only when BOTH cpu and memory are set (and thus equal across requests and limits). A partial default — only cpu, or only memory — leaves the other dimension unset, so Kubernetes classifies the pod as Burstable, not Guaranteed.

type Execution 

type Execution struct {
	NodeSelector    map[string]string `json:"node_selector,omitempty" yaml:"node_selector,omitempty"`
	Tolerations     []map[string]any  `json:"tolerations,omitempty" yaml:"tolerations,omitempty"`
	ServiceAccount  string            `json:"service_account,omitempty" yaml:"service_account,omitempty"`
	ImagePullPolicy string            `json:"image_pull_policy,omitempty" yaml:"image_pull_policy,omitempty"`

	// PriorityClassName ranks this task pod against its neighbors on a shared
	// cluster; the named PriorityClass is a platform-owned, cluster-scoped object,
	// so under genuine contention the scheduler preempts Dexaflow's ETL rather than
	// production services (ADR 0054).
	PriorityClassName string `json:"priority_class_name,omitempty" yaml:"priority_class_name,omitempty"`
	// TerminationGracePeriodSeconds is how long the pod is given to shut down after
	// a delete/preempt before SIGKILL. Nil leaves the cluster default (30s).
	TerminationGracePeriodSeconds *int64 `json:"termination_grace_period_seconds,omitempty" yaml:"termination_grace_period_seconds,omitempty"`
	// RuntimeClassName selects an alternate container runtime (e.g. a sandboxed or
	// GPU runtime) registered as a RuntimeClass. Nil uses the cluster default.
	RuntimeClassName *string `json:"runtime_class_name,omitempty" yaml:"runtime_class_name,omitempty"`
	// TopologySpreadConstraints spread a DAG's task pods across failure domains
	// (zones, nodes). Untyped []map[string]any carried verbatim from the DAG spec;
	// the executor round-trips it to []corev1.TopologySpreadConstraint.
	TopologySpreadConstraints []map[string]any `json:"topology_spread_constraints,omitempty" yaml:"topology_spread_constraints,omitempty"`
	// Affinity pins or repels a task pod relative to nodes and other pods
	// (node/pod affinity and anti-affinity). Untyped map[string]any carried verbatim
	// from the DAG spec; the executor round-trips it to *corev1.Affinity.
	Affinity map[string]any `json:"affinity,omitempty" yaml:"affinity,omitempty"`
	// ResourceClaims declares the pod-level ResourceClaims (Dynamic Resource
	// Allocation, GA in Kubernetes 1.34) an accelerator DAG needs — e.g. a GPU from
	// a claim template. Untyped []map[string]any carried verbatim from the DAG spec;
	// the executor round-trips it to []corev1.PodResourceClaim. A container consumes
	// one by naming it in Resources.Claims.
	ResourceClaims []map[string]any `json:"resource_claims,omitempty" yaml:"resource_claims,omitempty"`
	// Labels and Annotations are operator-declared pod metadata merged onto the task
	// pod. Keys under ReservedMetadataPrefix belong to Dexaflow (the reconciler,
	// terminate path, warm-pool reconciler and token exchange select on them), so
	// Validate refuses them and the executor drops any that still arrive.
	Labels      map[string]string `json:"labels,omitempty" yaml:"labels,omitempty"`
	Annotations map[string]string `json:"annotations,omitempty" yaml:"annotations,omitempty"`
}

Execution carries executor-specific placement and scheduling hints for a task. Every field beyond NodeSelector/Tolerations/ServiceAccount is applied only by the Kubernetes executor; Lite (subprocess, no pods) ignores them.

type ExecutionMode 

type ExecutionMode string

ExecutionMode selects how a task runs. Every task runs inside a worker pod; the field is retained for forward compatibility and defaults to pod. 

const (
	// ExecutionModePod runs a task inside a worker pod via the agent.
	ExecutionModePod ExecutionMode = "pod"
)

Supported execution modes. See docs/api/dag-schema.json.

type HistoricalMetrics 

type HistoricalMetrics struct {
	RunStates map[string]int
	TIStates  map[string]int
}

HistoricalMetrics holds run- and task-instance counts grouped by state over a time window, keyed by the Dexaflow state name (e.g. "success", "up_for_retry").

type ImportError 

type ImportError struct {
	// ID is the stable identifier of the error record.
	ID string
	// Filename is the DAG source path that failed to import.
	Filename string
	// StackTrace is the human-readable parse/compile error (traceback).
	StackTrace string
	// BundleName is the originating bundle (empty when unknown).
	BundleName string
	// Timestamp is when the error was recorded.
	Timestamp time.Time
}

ImportError is a DAG parse/compile failure surfaced as Airflow's "Import Errors" banner on the home dashboard. It is keyed by Filename; a successful re-import of the same file clears it. The `dexaflow lite` watcher writes these on a failed compile and removes them on the next good compile.

type LeoflowConfig 

type LeoflowConfig struct {
	SchemaVersion string   `json:"schema_version,omitempty" yaml:"schema_version,omitempty"`
	DagID         string   `json:"dag_id" yaml:"dag_id"`
	Description   string   `json:"description,omitempty" yaml:"description,omitempty"`
	Owner         string   `json:"owner,omitempty" yaml:"owner,omitempty"`
	Tags          []string `json:"tags,omitempty" yaml:"tags,omitempty"`
	PythonVersion string   `json:"python_version,omitempty" yaml:"python_version,omitempty"`
	// PythonVersionDefaulted reports that ApplyDefaults supplied PythonVersion
	// because the config declared none. It is not part of the authoring surface
	// and never serialized — it exists so a consumer can tell "the author chose
	// 3.11" from "nobody said anything", which are the same string afterwards.
	// `dexaflow lite` needs the distinction: enforcing a version the CLI invented
	// refuses to boot over a choice the user never made (#1092 follow-up).
	PythonVersionDefaulted bool     `json:"-" yaml:"-"`
	BaseImage              string   `json:"base_image,omitempty" yaml:"base_image,omitempty"`
	Dependencies           []string `json:"dependencies,omitempty" yaml:"dependencies,omitempty"`
	Connectors             []string `json:"connectors,omitempty" yaml:"connectors,omitempty"`
	// Connections and Variables are the per-DAG declared secret sets (ADR 0045,
	// ADR 0055). Carried verbatim to the parser, which emits them into dag.json.
	// Distinct from Connectors (pip provider packages, ADR 0038) — a different key
	// one letter away. Empty declares nothing.
	Connections    []string        `json:"connections,omitempty" yaml:"connections,omitempty"`
	Variables      []string        `json:"variables,omitempty" yaml:"variables,omitempty"`
	SystemPackages []string        `json:"system_packages,omitempty" yaml:"system_packages,omitempty"`
	DagSource      string          `json:"dag_source,omitempty" yaml:"dag_source,omitempty"`
	IncludePaths   []string        `json:"include_paths,omitempty" yaml:"include_paths,omitempty"`
	ExcludePaths   []string        `json:"exclude_paths,omitempty" yaml:"exclude_paths,omitempty"`
	Build          *BuildConfig    `json:"build,omitempty" yaml:"build,omitempty"`
	Registry       *RegistryConfig `json:"registry,omitempty" yaml:"registry,omitempty"`
	Defaults       *ConfigDefaults `json:"defaults,omitempty" yaml:"defaults,omitempty"`
	// Staging requests the opt-in per-DAG-run shared volume (ADR 0022). It is a
	// Dexaflow deployment concern (not an Airflow DAG attribute), so it lives in
	// dexaflow.yaml and the compiler overlays it onto the produced dag.json.
	Staging *StagingConfig `json:"staging,omitempty" yaml:"staging,omitempty"`
	// Dbt declares a dbt project as the DAG source (ADR 0042). Its presence routes
	// `dexaflow compile` to the dbt renderer instead of the Python parser.
	Dbt *DbtConfig `json:"dbt,omitempty" yaml:"dbt,omitempty"`
	// DbtGroups configures dbt projects embedded as task groups in a dag.py (ADR
	// 0043), keyed by the name passed to `dbt_group(name)`. Schedule does not apply
	// to a group (the DAG owns the schedule).
	DbtGroups map[string]*DbtConfig `json:"dbt_groups,omitempty" yaml:"dbt_groups,omitempty"`
	// Tasks holds per-task overrides bound by task_id (ADR 0023). Each entry's
	// key must match a task_id in the compiled DAG; the compiler errors on an
	// unknown id rather than silently dropping it.
	Tasks map[string]*TaskConfig `json:"tasks,omitempty" yaml:"tasks,omitempty"`
	// Alerts declares native on-failure alerting (#424): the scheduler fires the
	// listed rules when a DagRun reaches the terminal failed state, in Go, with no
	// task pod and no Python in the hot path. A Dexaflow deployment concern (not an
	// Airflow DAG attribute), so it lives in dexaflow.yaml and the compiler overlays
	// it onto the produced dag.json.
	Alerts *AlertsConfig `json:"alerts,omitempty" yaml:"alerts,omitempty"`
}

LeoflowConfig is the developer-facing project configuration parsed from dexaflow.yaml. It mirrors docs/api/leoflow-yaml-schema.json and is consumed by `dexaflow compile` to build an image and emit a DAGSpec.

func (*LeoflowConfig) ApplyDefaults 

func (c *LeoflowConfig) ApplyDefaults()

ApplyDefaults fills zero-valued fields with the defaults declared in the canonical JSON Schema (internal/domain/schemas/leoflow-yaml-schema.json). Explicit user-set values are preserved; nested structs (Build, Registry) are instantiated when nil so their own defaults can be applied. The method is idempotent: a second call after the first is a no-op.

Centralizing defaults here (instead of scattered `if x == ""` fallbacks at each consumer) is what lets the multi-DAG workspace synthesize a working config when a subdir ships no dexaflow.yaml, while keeping the resolved values debuggable from one place.

func (*LeoflowConfig) EffectiveDependencies 

func (c *LeoflowConfig) EffectiveDependencies() ([]string, error)

EffectiveDependencies resolves the full pip install list the image/venv needs: the `connectors:` short names expanded to their apache-airflow-providers-* packages (ADR 0038's sugar), followed by the explicit `dependencies:` verbatim. Providers come first so a transitive driver pinned in dependencies resolves against the provider declared via the sugar.

An unknown connector name is a compile error, not a silent drop: a typo that slipped through would otherwise surface as a ModuleNotFoundError inside the task pod, far from its cause. The message names the offender, lists the known types, and points at the dependencies: escape hatch.

func (*LeoflowConfig) Validate 

func (c *LeoflowConfig) Validate() error

Validate checks the LeoflowConfig against the canonical dexaflow.yaml schema and returns a joined error describing every violation, or nil when valid.

type PageCursor added in v0.5.1 

type PageCursor struct {
	At  time.Time
	Key string
}

PageCursor is a keyset position in a list ordered newest first: the next page holds the rows strictly before (At, Key). Key breaks ties between rows with the same At: the run id for dag runs, the entry id for the audit log.

type ParamSpec 

type ParamSpec struct {
	// Default is the value merged into a run's conf when the trigger omits this
	// key. Absent (omitempty, len 0) means the param is REQUIRED — the trigger
	// must supply it — as distinct from an explicit JSON null default.
	Default json.RawMessage `json:"default,omitempty"`
	Schema  json.RawMessage `json:"schema,omitempty"`
}

ParamSpec is one author-declared DAG-run parameter: a default value and the JSON Schema its trigger-time conf value is validated against. Both are carried as raw JSON so an arbitrary default and an arbitrary schema round-trip verbatim. Schema is {} (or absent) when the author declared a bare default with no constraints, in which case any conf value for that key is accepted.

type Pool 

type Pool struct {
	Name        string
	Slots       int
	Description string
	IsDefault   bool
}

Pool is a named, tenant-scoped, cross-DAG task-concurrency budget (Airflow's pool). Slots is the cap: a task in the pool is admitted to `queued` only while the pool has a free slot, counting queued+running task instances across every DAG (ADR 0053 Stage 3). IsDefault marks the implicit default_pool a task with no declared pool falls back to. Pools are a Pro-only concept.

type PoolUsage 

type PoolUsage struct {
	Running   int
	Queued    int
	Scheduled int
	Deferred  int
}

PoolUsage is a pool's per-state occupancy, feeding the Airflow PoolResponse slot fields. The slots admission actually spends are queued+running; scheduled and deferred are reported for the UI but do not hold a slot.

type PythonDeprecation 

type PythonDeprecation struct {
	// Version is the deprecated python_version value, e.g. "3.10".
	Version string `json:"-"`
	// EOL is the upstream CPython end-of-life date (YYYY-MM-DD). After it,
	// docker-library/python stops rebuilding the line and the base image stops
	// receiving OS security updates.
	EOL string `json:"eol"`
	// RemoveAfter is the date after which the release stops publishing a base
	// image for this line (YYYY-MM-DD).
	RemoveAfter string `json:"remove_after"`
	// Replacement is the version an author should move to.
	Replacement string `json:"replacement"`
	// Reason is the human-readable justification, quoted verbatim in the CLI
	// warning so the author gets the argument and not just a verdict.
	Reason string `json:"reason"`
}

PythonDeprecation records why a published Python line is on its way out and when it stops being published. It is read from the authoring schema rather than declared in Go so that the enum and the deprecation notes cannot drift apart: they are two fields of the same object.

func DeprecatedPythonVersion 

func DeprecatedPythonVersion(v string) (PythonDeprecation, bool)

DeprecatedPythonVersion reports the deprecation record for v, if the schema carries one. The second result is false for a version that is fine.

type RegistryConfig 

type RegistryConfig struct {
	URL         string `json:"url,omitempty" yaml:"url,omitempty"`
	AuthMethod  string `json:"auth_method,omitempty" yaml:"auth_method,omitempty"`
	ImageName   string `json:"image_name,omitempty" yaml:"image_name,omitempty"`
	TagStrategy string `json:"tag_strategy,omitempty" yaml:"tag_strategy,omitempty"`
}

RegistryConfig describes where the built image is pushed and how it is tagged.

type ResourceQuantity 

type ResourceQuantity struct {
	CPU    string `json:"cpu,omitempty" yaml:"cpu,omitempty"`
	Memory string `json:"memory,omitempty" yaml:"memory,omitempty"`
	// EphemeralStorage bounds the node-local scratch (writable layer, emptyDir,
	// logs) a task may use. Setting it keeps a runaway task from filling a shared
	// node's disk and evicting its neighbors under disk pressure (ADR 0054).
	// Kubernetes quantity, e.g. "2Gi".
	EphemeralStorage string `json:"ephemeral_storage,omitempty" yaml:"ephemeral_storage,omitempty"`
}

ResourceQuantity expresses CPU, memory, and ephemeral-storage in Kubernetes notation.

type ResourceUnit added in v0.5.2 

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

ResourceUnit is the operator's resource unit (executor.unit, ADR 0066 §3): the CPU and memory one pool slot stands for. A task of pool_slots N may use at most N units, and a task that declares no CPU or memory gets N units. A nil *ResourceUnit means no unit is configured and every method is a no-op, so callers need not branch on it.

func ParseResourceUnit added in v0.5.2 

func ParseResourceUnit(c ResourceUnitConfig) (*ResourceUnit, error)

ParseResourceUnit parses the configured unit. CPU and memory both empty means no unit (nil, nil), whatever enforce and max_size say. Otherwise both must be valid, positive Kubernetes quantities: a unit with only one dimension would leave the other unbounded, which is the opposite of what a unit is for.

func (*ResourceUnit) Apply added in v0.5.2 

func (u *ResourceUnit) Apply(t TaskSpec) *Resources

Apply returns the resources a task's pod gets. With no unit it returns the task's resources unchanged (nil when it declares none). With a unit:

  • A task that declares no cpu or memory gets requests = limits = pool_slots x unit (Guaranteed QoS), as ADR 0066 §3 says.
  • A value the task declares is never changed: the engine does not clamp or rewrite an author's resources.
  • A missing request follows the declared limit of the same dimension, the value Kubernetes itself would give it, else pool_slots x unit.
  • A missing limit is pool_slots x unit, or the declared request when that is larger, which only a misfit tolerated under enforce: warn can be. That task then runs with its own resources, as warn promises.

So every task ends up with a cpu and memory limit, and requests are never above limits, which the API server would reject. The task is not modified.

func (*ResourceUnit) Check added in v0.5.2 

func (u *ResourceUnit) Check(t TaskSpec) (warned, refused error)

Check decides a task against the unit. refused is a misfit that must stop it (registration fails, dispatch refuses). warned is a misfit tolerated under enforce: warn, which the caller logs and counts and then lets through. A task over max_size is always refused: pool_slots is new with the unit, so no task that ran before can trip it. Nil unit: both nil.

func (*ResourceUnit) CheckSpec added in v0.5.2 

func (u *ResourceUnit) CheckSpec(d *DAGSpec) (warned []error, refused error)

CheckSpec runs Check on every task of the DAG: refused is the first refusal, warned every tolerated misfit. Nil unit: both nil.

func (*ResourceUnit) WarmEligible added in v0.5.2 

func (u *ResourceUnit) WarmEligible(t TaskSpec) bool

WarmEligible reports whether a task may be placed on a warm pod. A warm pod is one unit, so with a unit set only a size-1 task that declares no resources fits on it; any other task takes a dedicated pod. Without a unit every task is eligible, as before.

func (*ResourceUnit) WarmResources added in v0.5.2 

func (u *ResourceUnit) WarmResources() *Resources

WarmResources is what a warm pod (ADR 0058) gets while a unit is set: one unit as requests and limits. Nil without a unit, leaving warm pods as before.

type ResourceUnitConfig added in v0.5.2 

type ResourceUnitConfig struct {
	CPU     string
	Memory  string
	Enforce string // "" means UnitEnforceRefuse
	MaxSize int    // 0 means DefaultUnitMaxSize
}

ResourceUnitConfig is the executor.unit section as configured.

type Resources 

type Resources struct {
	Requests *ResourceQuantity `json:"requests,omitempty" yaml:"requests,omitempty"`
	Limits   *ResourceQuantity `json:"limits,omitempty" yaml:"limits,omitempty"`
	// Claims lists the ResourceClaims (declared in Execution.ResourceClaims) this
	// task's container consumes — the container half of Dynamic Resource Allocation
	// (DRA, GA in Kubernetes 1.34). Untyped []map[string]any carried verbatim from
	// the DAG spec; the executor round-trips it to []corev1.ResourceClaim. Each
	// entry names a claim (and optionally a specific request within it) that makes
	// an accelerator available inside the container.
	Claims []map[string]any `json:"claims,omitempty" yaml:"claims,omitempty"`
}

Resources holds Kubernetes-style resource requests and limits for a task.

type SafeError 

type SafeError struct {
	// Message is the client-facing phrase, without the class's own wording.
	Message string
	// Class is the sentinel this failure belongs to (ErrValidation,
	// ErrConflict, ErrNotFound, ...). errors.Is sees it through Unwrap.
	Class error
}

SafeError carries a message Dexaflow composed itself, attached to one of the sentinel classes above. It exists so the API boundary can tell "a phrase we wrote for the caller" apart from "whatever text the failure happened to carry".

The distinction is a security boundary, not a style choice. A repository error usually wraps a driver error, and a *pgconn.PgError renders itself as "severity: message (SQLSTATE code)" with the constraint, table and column names of our schema inside the message. Echoing that to an authenticated tenant of a multi-tenant control plane is schema disclosure (CWE-209). The API therefore returns Message verbatim and replaces everything else with a constant phrase, logging the real error instead.

That inverts the safe/unsafe default: a new error is opaque to clients until someone deliberately writes it as a SafeError, so a newly wrapped driver error cannot leak by omission.

Message is returned to API clients verbatim. Never interpolate a driver, filesystem, or network error into it — pass the caller-facing facts (a role name, a declared variable, a configured cap) and let the underlying error reach the log through the wrap chain.

func (*SafeError) ClientMessage 

func (e *SafeError) ClientMessage() string

ClientMessage returns the phrase that may be shown to an API client.

func (*SafeError) Error 

func (e *SafeError) Error() string

Error renders "<message>: <class>", identical to what fmt.Errorf("<message>: %w", class) produced before, so logs and error strings are unchanged.

func (*SafeError) Unwrap 

func (e *SafeError) Unwrap() error

Unwrap returns the sentinel class so errors.Is keeps working.

type StagingConfig 

type StagingConfig struct {
	Enabled      bool   `json:"enabled" yaml:"enabled"`
	Size         string `json:"size,omitempty" yaml:"size,omitempty"`
	StorageClass string `json:"storage_class,omitempty" yaml:"storage_class,omitempty"`
}

StagingConfig is the opt-in per-DAG-run shared staging volume (ADR 0022). Size is a Kubernetes quantity (e.g. "5Gi"); StorageClass empty uses the cluster default RWX class.

type StagingVolumeState 

type StagingVolumeState struct {
	// PVCName is the staging PersistentVolumeClaim's name.
	PVCName string
	// RunState is the DAG run's state ("success", "failed", "running", …), or
	// empty when the run no longer exists.
	RunState string
	// RunEndedAt is when the run reached a terminal state, if known.
	RunEndedAt *time.Time
	// CreatedAt is when the volume was provisioned. The GC never deletes a volume
	// younger than the TTL when its run cannot be resolved, so a lookup miss can
	// never reclaim an active run's fresh volume.
	CreatedAt time.Time
}

StagingVolumeState is a tracked per-run staging volume joined with its DAG run's state, used by the GC to decide deletion (ADR 0022). RunState is empty when the run row is gone (orphan); RunEndedAt is the run's terminal time, used for the post-terminal TTL on failed runs.

type TaskConfig 

type TaskConfig struct {
	Retries                 *int              `json:"retries,omitempty" yaml:"retries,omitempty"`
	RetryDelaySeconds       *int              `json:"retry_delay_seconds,omitempty" yaml:"retry_delay_seconds,omitempty"`
	ExecutionTimeoutSeconds *int              `json:"execution_timeout_seconds,omitempty" yaml:"execution_timeout_seconds,omitempty"`
	Env                     map[string]string `json:"env,omitempty" yaml:"env,omitempty"`
	// Connections and Variables narrow the DAG-level declared secret set to this
	// task (ADR 0045 §Settled #1, ADR 0055). Empty means the task inherits the
	// DAG-level declaration.
	Connections []string   `json:"connections,omitempty" yaml:"connections,omitempty"`
	Variables   []string   `json:"variables,omitempty" yaml:"variables,omitempty"`
	Resources   *Resources `json:"resources,omitempty" yaml:"resources,omitempty"`
	Execution   *Execution `json:"execution,omitempty" yaml:"execution,omitempty"`
	// Size is the task's pool_slots (ADR 0066): how many slots of its pool it
	// takes while queued or running. It overrides pool_slots from dag.py.
	Size *int `json:"size,omitempty" yaml:"size,omitempty"`
}

TaskConfig holds the dexaflow.yaml per-task overrides bound by task_id (ADR 0023). Every field is optional; a set field overrides the value compiled from the DAG (most specific wins: task override > DAG default_args). These are Dexaflow deployment concerns, not Airflow operator attributes.

type TaskInstance 

type TaskInstance struct {
	DagID     string
	RunID     string
	TaskID    string
	MapIndex  int
	TryNumber int
	MaxTries  int
	State     TaskState
	Operator  string
	// ScheduledAt and QueuedAt record when the instance first entered the
	// scheduled and queued states (Airflow's scheduled_when / queued_when).
	ScheduledAt *time.Time
	QueuedAt    *time.Time
	StartedAt   *time.Time
	EndedAt     *time.Time
	Duration    *float64
	Hostname    string
	// Note is operational context shown in the UI's task panel — e.g. why a task
	// is queued but not running (no executor available).
	Note string
	// FailureReason is a short, human-readable cause for a terminal failure,
	// recorded by whichever component observed it: the agent's own report, the
	// reconciler reading the pod (image pull, OOM, exit code), a reaper declaring
	// the pod or agent lost, or the agent's pre-registration classification. It is
	// the answer to "why did this fail?" for an attempt that streamed no logs
	// because its agent never started — the case where the only remaining source
	// of truth used to be `kubectl logs` against the cluster.
	//
	// It is best-effort and often empty: a healthy instance has none, and a cause
	// nobody observed cannot be invented. It carries a classification, never a
	// credential or a raw internal error.
	FailureReason string
}

TaskInstance is an execution of a task within a DagRun.

type TaskSpec 

type TaskSpec struct {
	TaskID      string      `json:"task_id"`
	Type        TaskType    `json:"type"`
	DependsOn   []string    `json:"depends_on,omitempty"`
	TriggerRule TriggerRule `json:"trigger_rule,omitempty"`
	// Pool is the named task pool this task draws a slot from (Airflow's `pool`),
	// the cross-DAG concurrency budget admission enforces (ADR 0053 Stage 3). Empty
	// (the default) means the implicit default_pool, so every task is always in a
	// well-defined pool. The pool gate is Pro-only; Lite ignores this field, so a
	// DAG that sets it plans identically on Lite.
	Pool string `json:"pool,omitempty"`
	// PoolSlots is how many slots of its pool the task takes while it is queued
	// or running (Airflow's pool_slots, ADR 0066). Zero means 1, so a DAG that
	// never sets it weighs exactly what it did before. dexaflow.yaml sets it as
	// `size`. Ignored where Pool is (Lite).
	PoolSlots               int               `json:"pool_slots,omitempty"`
	Retries                 *int              `json:"retries,omitempty"`
	RetryDelaySeconds       *int              `json:"retry_delay_seconds,omitempty"`
	ExecutionTimeoutSeconds *int              `json:"execution_timeout_seconds,omitempty"`
	ExecutionMode           ExecutionMode     `json:"execution_mode,omitempty"`
	Entrypoint              string            `json:"entrypoint,omitempty"`
	Env                     map[string]string `json:"env,omitempty"`
	// Variables and Connections narrow the DAG's declared secret set to this task
	// (ADR 0045 §Settled #1, ADR 0055). Absent (empty) means the task inherits the
	// DAG-level declaration. Carries the declaration only; delivery is unchanged.
	Variables   []string            `json:"variables,omitempty"`
	Connections []string            `json:"connections,omitempty"`
	Resources   *Resources          `json:"resources,omitempty"`
	Execution   *Execution          `json:"execution,omitempty"`
	XComInput   map[string][]string `json:"xcom_input,omitempty"`
	XComSchema  map[string]any      `json:"xcom_schema,omitempty"`
	// CallArgs carries TaskFlow literal call arguments captured at compile time
	// (#115). The agent serializes the whole map as a single env var
	// LEOFLOW_CALL_ARGS_JSON; the runtime decodes and delivers each value to
	// the user function. XCom upstreams take precedence at runtime over a
	// same-name literal (the deterministic merge owned by leoflow_runtime).
	// Named call_args (not params) to leave the term free for Airflow's
	// DAG-run params semantic (#148).
	CallArgs map[string]any `json:"call_args,omitempty"`
	// OperatorClass is the dotted Airflow operator/sensor class for an
	// airflow_operator task (ADR 0040), e.g.
	// "airflow.providers.snowflake.operators.snowflake.SQLExecuteQueryOperator".
	OperatorClass string `json:"operator_class,omitempty"`
	// OperatorArgs are the operator's constructor kwargs captured at compile time.
	// The agent serializes them as the env var LEOFLOW_OPERATOR_ARGS; the runtime
	// instantiates the operator with them.
	OperatorArgs map[string]any `json:"operator_args,omitempty"`
	// OnFailureCallback marks that the task declares an Airflow on_failure_callback
	// (#424). The callable itself is not carried (it can't be serialized); the
	// runtime re-imports dag.py and runs it in the task process on failure. The
	// flag lets the agent/UI know a callback will run without importing user code.
	OnFailureCallback bool `json:"on_failure_callback,omitempty"`
}

TaskSpec describes a single unit of work within a DAG.

func (TaskSpec) EffectiveExecutionMode 

func (t TaskSpec) EffectiveExecutionMode() ExecutionMode

EffectiveExecutionMode returns the task's execution mode, defaulting to pod when unset. Every task runs in a worker pod.

func (TaskSpec) EffectivePoolSlots added in v0.5.2 

func (t TaskSpec) EffectivePoolSlots() int

EffectivePoolSlots is the number of pool slots the task takes: its PoolSlots, or 1 when unset (ADR 0066).

type TaskState 

type TaskState string

TaskState is the lifecycle state of a TaskInstance. The values mirror the task_state enum in the database (migration 003). 

const (
	// TaskStateNone is the initial state: the task has not been considered yet.
	TaskStateNone TaskState = "none"
	// TaskStateScheduled means dependencies are satisfied and the task is queued for dispatch.
	TaskStateScheduled TaskState = "scheduled"
	// TaskStateQueued means the executor has been asked to start the task.
	TaskStateQueued TaskState = "queued"
	// TaskStateRunning means the task is executing.
	TaskStateRunning TaskState = "running"
	// TaskStateSuccess means the task finished successfully.
	TaskStateSuccess TaskState = "success"
	// TaskStateFailed means the task finished with an error.
	TaskStateFailed TaskState = "failed"
	// TaskStateSkipped means the task was deliberately not run.
	TaskStateSkipped TaskState = "skipped"
	// TaskStateUpstreamFailed means a required upstream failed, so the task cannot run.
	TaskStateUpstreamFailed TaskState = "upstream_failed"
	// TaskStateUpForRetry means the task failed but has retries remaining.
	TaskStateUpForRetry TaskState = "up_for_retry"
	// TaskStateUpForReschedule means a reschedule-mode sensor poked not-ready and
	// released its pod; the scheduler re-dispatches it once reschedule_at is reached,
	// without consuming retry budget (ADR 0040 Phase B, #380). Non-terminal.
	TaskStateUpForReschedule TaskState = "up_for_reschedule"
)

Task lifecycle states.

func (TaskState) IsTerminal 

func (s TaskState) IsTerminal() bool

IsTerminal reports whether the task state is final (no further automatic transitions occur from it).

type TaskType 

type TaskType string

TaskType enumerates the kinds of work a task can perform. 

const (
	// TaskTypePython runs a Python callable identified by an entrypoint.
	TaskTypePython TaskType = "python"
	// TaskTypeBash runs a shell command supplied as the entrypoint.
	TaskTypeBash TaskType = "bash"
	// TaskTypeAirflowOperator runs a captured Airflow provider operator/sensor in
	// the task pod via the generic executor (ADR 0040): the runtime instantiates
	// OperatorClass with OperatorArgs and calls execute(). The provider is
	// installed in the image via connectors:/dependencies:.
	TaskTypeAirflowOperator TaskType = "airflow_operator"
	// TaskTypeDbtGroup is a transient placeholder for a dbt project embedded in a
	// DAG (ADR 0043). The compiler expands it into one task per dbt node and the
	// type never appears in a finished dag.json.
	TaskTypeDbtGroup TaskType = "dbt_group"
)

Supported task types. See docs/api/dag-schema.json.

type TenantLimits added in v0.5.1 

type TenantLimits struct {
	// MaxDags caps how many DAGs the tenant may register. New versions of a DAG
	// it already has are always accepted.
	MaxDags int
	// MaxRunsPerDay caps how many DAG runs, manual and scheduled alike, the
	// tenant may create in one UTC calendar day.
	MaxRunsPerDay int
	// MinScheduleIntervalSeconds is the shortest gap, in seconds, a DAG's
	// schedule may leave between two consecutive runs (MinScheduleInterval).
	MinScheduleIntervalSeconds int
	// MaxTaskPoolSlots caps a task's pool_slots (its size, ADR 0066): a DAG
	// with a larger task is refused at registration.
	MaxTaskPoolSlots int
}

TenantLimits are optional caps on what one tenant may do, set by the operator through the service API. A zero field means unlimited, which is also what a tenant created without limits has.

type TenantLimitsUpdate added in v0.5.1 

type TenantLimitsUpdate struct {
	MaxDags                    *int
	MaxRunsPerDay              *int
	MinScheduleIntervalSeconds *int
	MaxTaskPoolSlots           *int
}

TenantLimitsUpdate changes some of a tenant's limits: a nil field leaves the stored value as it is, and a pointer to 0 sets that limit back to unlimited.

func (TenantLimitsUpdate) IsZero added in v0.5.1 

func (u TenantLimitsUpdate) IsZero() bool

IsZero reports whether the update changes nothing.

type TriggerRule 

type TriggerRule string

TriggerRule decides whether a task runs based on its upstreams' states. 

const (
	// TriggerRuleAllSuccess runs when every upstream succeeded (default).
	TriggerRuleAllSuccess TriggerRule = "all_success"
	// TriggerRuleAllFailed runs when every upstream failed.
	TriggerRuleAllFailed TriggerRule = "all_failed"
	// TriggerRuleAllDone runs once every upstream finished, regardless of state.
	TriggerRuleAllDone TriggerRule = "all_done"
	// TriggerRuleOneSuccess runs as soon as one upstream succeeds.
	TriggerRuleOneSuccess TriggerRule = "one_success"
	// TriggerRuleOneFailed runs as soon as one upstream fails.
	TriggerRuleOneFailed TriggerRule = "one_failed"
)

Supported trigger rules for the MVP. See docs/api/dag-schema.json.

type UnitMisfitError added in v0.5.2 

type UnitMisfitError struct {
	TaskID  string
	Size    int // the task's pool_slots
	Need    int // the size its resources need; 0 when over max_size
	MaxSize int // set when the task is over executor.unit.max_size
	// contains filtered or unexported fields
}

UnitMisfitError is a task that does not fit the unit: it declares more than pool_slots x unit, or its pool_slots is above executor.unit.max_size.

func (*UnitMisfitError) Error added in v0.5.2 

func (e *UnitMisfitError) Error() string

type User 

type User struct {
	ID        string
	Email     string
	Roles     []string
	IsActive  bool
	CreatedAt time.Time
}

User is a control-plane account as returned by the admin user-management API. It never carries the password or its hash — those are write-only. Roles is the full set of role names the user holds: the list path aggregates every role grant, and the create path echoes back the roles it granted (empty when none were requested).

type Variable 

type Variable struct {
	Key         string
	Value       string
	Description string
}

Variable is a tenant-scoped key/value setting consumed by DAGs and managed from the Admin UI. Value is stored as-is (plaintext for the MVP); the API masks values of secret-ish keys.

type VariablePatch 

type VariablePatch struct {
	Key         string
	Value       *string
	Description *string
}

VariablePatch is a tri-state write to a Variable (#887). Description mirrors domain.ConnectionPatch: nil preserves the stored value (COALESCE), non-nil "" clears, and a value sets. Value is subtly different because the `value` column is NOT NULL: the caller resolves an omitted or masked ("***" for a sensitive key) value to the stored value BEFORE building the patch, so Value is expected non-nil here — a non-nil "" still clears. Key is always written.

type XComEntryMeta 

type XComEntryMeta struct {
	Key       string
	Timestamp time.Time
	MapIndex  int
}

XComEntryMeta is the metadata for one stored XCom value (without the value payload) — the source for a task instance's XCom list. Dexaflow XComs are unmapped, so MapIndex is -1.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.