From 010ec117001446935602c77ee3d03b3be9fe3103 Mon Sep 17 00:00:00 2001 From: athvin Date: Sun, 12 Jul 2026 21:01:17 -0400 Subject: [PATCH] Make repo documentation-only Remove the Rust workspace (crates, Cargo manifests, examples, migrations, build output, and CI) and keep only the docs. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/architecture.md | 672 ++++++++++++++++++++++++++++++++++++++++ docs/authoring-guide.md | 431 ++++++++++++++++++++++++++ docs/data-model.md | 363 ++++++++++++++++++++++ docs/decisions.md | 40 +++ docs/deployment.md | 235 ++++++++++++++ 5 files changed, 1741 insertions(+) create mode 100644 docs/architecture.md create mode 100644 docs/authoring-guide.md create mode 100644 docs/data-model.md create mode 100644 docs/decisions.md create mode 100644 docs/deployment.md diff --git a/docs/architecture.md b/docs/architecture.md new file mode 100644 index 0000000..d02f23a --- /dev/null +++ b/docs/architecture.md @@ -0,0 +1,672 @@ +# Maestro — Architecture + +> A Rust-native DAG scheduler for one team. Airflow-shaped, but with **zero runtime parsing** +> (compilation *is* the parse — §1.1), **no dynamic tasks**, **compile-time-checked data passing**, and +> **one binary** that runs a DAG either in-process (dev) or one-pod-per-task on Kubernetes (prod). +> +> The production footprint is the pitch: **one scheduler pod + ephemeral task pods + a bucket.** + +This document explains how Maestro is put together and *why*. Companion docs: + +- **[authoring-guide.md](./authoring-guide.md)** — how you write a DAG in Rust. +- **[data-model.md](./data-model.md)** — the state schema, state machines, and indexes. +- **[deployment.md](./deployment.md)** — running it locally and on Kubernetes. + +--- + +## 1. What Maestro is (and the three departures from Airflow) + +Airflow lets you `import` a Python SDK and build DAGs in code. Maestro keeps that *feel* — you add a +crate to `Cargo.toml`, import the SDK, and write DAGs in Rust — but diverges in three ways that shape +everything else here. + +| # | Departure | Why | Consequence | +|---|-----------|-----|-------------| +| 1 | **Zero runtime parsing.** The compiled binary already "knows" every DAG and task. | Airflow re-parses because a Python DAG file only *produces* a DAG when executed, and the product can depend on runtime data — so its parser can never finish. That's its single biggest source of latency, flakiness, and anti-patterns. | No `DagFileProcessor`. No file scanning. The scheduler reads a **compiled-in registry**. | +| 2 | **No dynamic task creation — strict.** Tasks are hard-coded. | Predictability. You can look at the source and know the exact task set. | A task is a **Rust type**, not a value you can build from runtime data. The graph is fixed per-process the moment the registry is built. | +| 3 | **One binary, two runtimes.** The same image runs a DAG in-process (dev) or one-pod-per-task (prod). | You shouldn't need a Kubernetes cluster to run a DAG. Dev must be a faithful rehearsal of prod. | The scheduler and every task pod are the *same compiled artifact*; only the state/executor backends swap. | + +Departures #1 and #2 are not two features — they are **one decision seen from two sides**: forbidding +dynamic tasks (#2) is exactly what makes a DAG fully determined by source code, which is exactly what +lets the compiler do the entire parse at build time (#1). §1.1 walks the chain. + +One more line of positioning, because it's the decision that keeps Maestro an order of magnitude lighter +than the durable-execution engines: **Maestro is state-machine rows + at-least-once task retries — not +event-sourced replay.** (Temporal is the explicit anti-goal.) + +### 1.1 Compilation is the parse + +An Airflow DAG file does not *contain* a DAG. It contains Python that *produces* one when executed — and +because that code can consult runtime data (dynamic task mapping, a DAG factory looping over a database +query), the product is undetermined until execution and can differ between executions. The only way to +know what such a file means is to run it — and keep running it, because the answer can change. That is +the `DagFileProcessor`: **a parser that is never allowed to finish.** Not an implementation wart; the +logical price of an undetermined artifact. + +Maestro deletes the reason, and the component follows. Forbidding dynamic tasks (§5.3, +[authoring-guide §8](./authoring-guide.md#8-the-no-dynamic-tasks-rule)) means there is no path from +runtime data to a task — so a DAG is **fully determined by source code**, and a fully determined artifact +can be discovered and verified once, before deploy. Rust already ships the machinery: + +| Job | Airflow (at runtime, forever) | Maestro (once, before deploy) | +|-----|-------------------------------|-------------------------------| +| **Discovery** — what DAGs and tasks exist | `DagFileProcessor` executes `.py` files on an interval | the linker: `inventory` collects every `#[maestro::task]` / `register_dag!` at link time (§4) | +| **Verification** — do the pieces fit | import errors surface per parse; data typing never (`xcom_pull` returns `Any`) | `rustc`: every referenced task exists, every `.after()`/`join()` edge type-checks, every IO type is `Serialize + Deserialize` (§5.1) | +| **Graph sanity** — whole-graph shape | cycle check re-run on every parse, inside the scheduler | `DagSpec::validate()`: acyclicity, duplicate instance ids, param defaults — as `maestro validate` in CI, re-checked at every boot (§5.2) | +| **Transport** — carry the parse product to scheduler / UI / workers | `serialized_dag` rows in the DB (K8s task pods re-parse anyway) | unnecessary: every role is the same binary, so every role already *has* the parse product (§3) | + +Compilation doesn't *skip* the parse; **compilation *is* the parse** — and `validate` is the lint. Green +build + green validate is the entire pre-deploy proof; nothing about the graph remains to be discovered +at runtime. (Boot re-runs the same deterministic checks on the same binary — it can confirm the answer, +never change it.) The graph errors Airflow surfaces in production — an `import_error` row, a wrong-shape +XCom at 6am in another pod — fail your build or your CI instead. + +So the binary is not code-that-will-describe-DAGs. It is **the finished, verified DAG set with an +execution engine attached**: Airflow ships source to be parsed; Maestro ships the result of parsing. +Hence the operating corollary — **what compiles (and validates) is what runs.** The DAG set is a +compile-time constant of the artifact: read the source and you know the exact graph; run the binary and +you get exactly that graph. (One honest boundary: the guarantee is per-build. Across a redeploy the +writer may be the old binary; that drift is caught at runtime by type fingerprints, loudly — §5.1, §7.1.) + +--- + +## 2. How Airflow works today — what we keep, what we drop + +A quick design ledger. (Sources: Airflow's official docs on the scheduler, DAG file processing, +executors, and XComs.) + +| Airflow mechanism | What it does | Maestro | +|-------------------|--------------|---------| +| **DagFileProcessor** | Re-parses `.py` DAG files on an interval, serializes them to the DB. | **Dropped — the job moved to the build.** A `.py` file only *produces* a DAG when executed, so Airflow must re-parse forever; a Maestro DAG is fully determined by source, so `rustc` and the linker did the parse once, at build time (§1.1). | +| **Scheduler loop** | Creates DagRuns from timetables, evaluates dependencies, hands ready tasks to an executor. | **Kept**, minus the parsing step, plus explicit reconciliation (§7.0) and redeploy semantics (§7.1) Airflow leaves implicit. | +| **Metadata DB as the bus** | Scheduler, webserver, and workers coordinate through Postgres with `SELECT … FOR UPDATE SKIP LOCKED`. | **Adapted.** State lives in embedded Turso; the scheduler is the **single writer** and serves an HTTP API; pods and browsers are clients. No distributed locking. | +| **Executors** (Local/Celery/Kubernetes) | Pluggable execution backends. | **Trimmed to two:** `LocalExecutor` (dev) and `KubernetesExecutor` (prod). | +| **XCom** | Stringly-typed task-to-task values in the DB; large data needs a custom backend. | **Kept, upgraded:** wiring is **type-checked at compile time**; inline values are self-describing JSON; large payloads go to object storage behind a pointer. | +| **Pools / priority / task caps** | Five overlapping concurrency mechanisms. | **Dieted to two:** global `--slots` + per-DAG `max_active_runs`. Kubernetes requests/limits arbitrate shared resources. The rest can return as additive migrations. | +| **KubernetesExecutor** | One pod per task; the pod re-parses the DAG. | **Kept the pod-per-task shape, dropped the re-parse:** the pod resolves the task from its compiled-in registry. | +| **Webserver** | A separate service rendering serialized metadata. | **Embedded.** The scheduler serves the UI and the API from one axum server, rendering the graph from the live registry. | + +The theme: **keep Airflow's control-plane shape, move the parsing/interpretation layer to build time +(compilation *is* the parse — §1.1), and cash in the single-writer design for simplicity** everywhere +Airflow needs distributed coordination. + +--- + +## 3. One binary, four roles + +There is exactly **one** compiled artifact, built from *your* crate — the one that defines your DAGs — +which depends on the `maestro` SDK and calls `maestro::run()` from `main`: + +```rust +use maestro::prelude::*; + +// ... your #[maestro::task] fns and dag definitions ... + +fn main() -> anyhow::Result<()> { + maestro::run() // parse the subcommand, discover the registry, dispatch the role +} +``` + +| Role | Lifetime | Responsibility | +|------|----------|----------------| +| `maestro scheduler` | long-running service | Owns the Turso DB (auto-applies pending migrations at boot), runs the loop, and serves the **web UI + HTTP/JSON control-plane API** from one axum server on one port. `--drain` stops new work for clean deploys (§7.1). | +| `maestro execute-task --dag-id X --task-id Y --run-id Z --try N` | short-lived | The **Kubernetes pod entrypoint**. Runs exactly one task instance, then exits. | +| `maestro dev --dag-id X [--trigger]` | single process | Everything at once: in-process scheduler + `LocalExecutor` + embedded Turso (auto-migrated). No cluster. | +| `maestro migrate` / `maestro validate` / `maestro run-task` | one-shot | Ops & dev tools: apply schema standalone; validate all DAGs for CI (cycles, duplicate ids, param defaults); run a single task in-process for debugging. | + +There is no separate web service: the UI is a read-mostly view for a handful of humans, and the scheduler +is already the single writer and the API server. Embedding it deletes a Deployment, a serialization +indirection, and version-drift handling — and makes prod match dev, which always worked this way. If the +scheduler is restarting, the UI is briefly down too; that is honest. + +Because all roles are the same binary, they build the **same in-memory DAG registry** (§4): + +- The scheduler ships only **identifiers** to a task pod (`dag-id`, `task-id`, `run-id`, `try`). It never + serializes or ships code. +- The pod's compiled-in registry resolves the identifier back to the actual closure. +- **Same image tag = same git SHA ⇒ the scheduler and the pods it launches run identical DAG code.** + (What happens when a *deploy changes the SHA* while runs are in flight is a real question with a real + answer — §7.1.) + +> **Why not separate binaries per role?** You'd have to build, tag, and version-match N images and hope +> the scheduler's view of a DAG matches the executor's. One binary makes code-consistency a compile-time +> fact. The cost is a slightly larger binary in short-lived pods — a good trade. + +--- + +## 4. The compiled-in registry (no parsing, ever) + +`#[maestro::task]` emits a hidden `inventory::submit!` **keyed by implementation id** (the fn name). The +author writes no registration boilerplate and there is no central list. At process start, +`Registry::discover()` collects the link-time set: + +```rust +// Emitted by #[maestro::task] (you never write this). Note: no `dag` field — +// a task implementation is DAG-agnostic and reusable. +inventory::submit! { + maestro::TaskRegistration { impl_id: "transform", fingerprint: TYPE_FP, make: || Arc::new(transform::Handle) } +} +inventory::submit! { maestro::DagRegistration { id: "sales_etl", build: sales_etl } } + +pub struct Registry { + dags: HashMap<&'static str, DagSpec>, // built once from DagRegistration + impls: HashMap<&'static str, fn() -> Arc>, // impl_id -> closure factory +} +``` + +**Implementation vs. instance.** A `DagSpec` node carries `(instance_id, impl_id)`: the *instance* id is +the task's name within that DAG (defaults to the fn name, overridable with `.with_id("extract_eu")`), the +*impl* id names the compiled function. One implementation can back many instances — across DAGs, or twice +in one DAG. This is Maestro's equivalent of Airflow's operator-vs-task distinction. + +So `maestro execute-task --dag-id sales_etl --task-id extract_eu …` does two lookups, both in memory: +instance → impl via the compiled `DagSpec`, impl → closure via the registry. **No file to read, no Python +to import, nothing to deserialize.** The lookup is cheap because the expensive step already happened: the +registry *is* the parse product, produced once by the compiler and linker instead of forever by a file +processor (§1.1). + +> `inventory` is the mechanism (purpose-built for "plugin registration without a central list", works +> across crates). `linkme` is the fallback if we ever need `no_std` or want to avoid +> life-before-`main` init. + +--- + +## 5. The SDK model at a glance + +Full walkthrough in **[authoring-guide.md](./authoring-guide.md)**; here is enough to understand the +runtime. The V1 authoring surface is **`#[maestro::task]` + a typed graph builder**. (A TaskFlow-style +proc-macro DSL is deferred — §13 — its lowering is sketched so it can be added without redesign.) + +A task is an `async fn` whose optional second argument is its **typed input** and whose return type is its +**typed output**, both `Serialize + DeserializeOwned`: + +```rust +#[maestro::task] +async fn transform(ctx: &TaskContext, input: RawRows) -> anyhow::Result { … } + +fn sales_etl() -> DagSpec { + Dag::builder("sales_etl") + .schedule(Cron::new("0 6 * * *")) + .add(extract::task()) + .add(transform::task().after(extract::task())) // ✅ compiles: Out(extract) == In(transform) + .add(load::task().after(transform::task())) + .build() +} +``` + +### 5.1 Compile-time-checked data passing (the headline feature — stated honestly) + +Airflow's `xcom_pull` returns `Any`; a wrong shape blows up at runtime in another pod. In Maestro each +task's handle carries its IO in the type system, and the edge methods only compile when the types line up: + +```rust +pub trait TaskIo { type In; type Out; const ID: &'static str; } + +impl TaskNode { + pub fn after(self, upstream: TaskNode) -> Self + where U: TaskIo::In> { … } // the whole trick +} + +load::task().after(extract::task()); // ❌ compile error: expected CleanRows, found RawRows +``` + +At runtime the value is serialized through the state store and deserialized in another process; because +the same type `T` is named on both sides of the edge in source, writer and reader agree — **within one +build**. Two honest caveats, both handled: + +- **Across a redeploy**, the writer may be the old binary and the reader the new one. Inline XCom is + therefore **self-describing JSON** by default, and every task carries a **type fingerprint** (a hash of + a stable schema descriptor of its `Out` type); `GetTaskInputs` checks the stored fingerprint against the + consumer's expectation and fails loudly on drift instead of mis-deserializing (§7.1). +- **Type-checking is not identity-checking.** Two upstreams with the *same* output type can be wired into + a join crosswise and the compiler can't object. Prefer distinct newtype outputs for join inputs. + +### 5.2 Type-checked while typed, homogeneous at runtime + +Typed handles (`TaskNode`) exist only during graph construction. `DagSpec` erases them to +`Arc` plus an edge list of instance ids — type-checking happens at every `.after()` call +site while the nodes are still typed; the stored graph is homogeneous and object-safe. The generated glue +for each task pulls its upstream values by **instance id from the DagSpec edge** (not by fn name) and +pushes its output, with the concrete types baked in by monomorphization. + +Cycles, duplicate instance ids, unknown params, and missing param defaults are validated by +`DagSpec::validate()` — run at every process boot **and** by `maestro validate` in CI, so a broken graph +fails the pipeline, not the scheduler. Division of labor: the **build is the parse** (every task exists, +every edge type-checks, IO types serialize, the registry is complete at link time), **`validate` is the +lint** (the whole-graph facts the type system can't express). Both are green before deploy — nothing +about the graph remains to be discovered at runtime (§1.1). + +### 5.3 Why builder-first + +The builder alone delivers every load-bearing guarantee: typed edges, zero parsing, a graph that is fixed +per-process (the registry is built once at boot — there is no API that turns runtime data into a task; +that prohibition is the premise that makes compilation-as-the-parse (§1.1) true: forbid runtime graph +construction and the whole graph becomes a compile-time fact). +Prior art agrees this is ergonomic (Hatchet, Temporal, and Windmill SDKs are all builder/registration +style). The TaskFlow-style `#[maestro::dag]` macro — call-graph edge inference with `Xcom` placeholders +— is deferred until the builder pinches: it requires a nontrivial AST rewrite (documented in §13) and a +second authoring surface is a permanent maintenance tax. + +--- + +## 6. State: the `StateStore` trait over embedded Turso + +All persisted state lives behind one async trait — **the scheduler's seam to a swappable database**. Task +pods do *not* implement it; they get a narrow `TaskClient` (§8.3). + +```rust +#[async_trait] +pub trait StateStore: Send + Sync { + // DagRun lifecycle (TriggerDagRun writes synchronously through this — no in-memory queue) + async fn create_dag_run(&self, dag_id: &str, logical_date: DateTime, t: RunTrigger) -> Result; + async fn set_dag_run_state(&self, run: &RunId, from: RunState, to: RunState) -> Result; + // TaskInstance state machine — guarded, try_number-fenced transitions are the ONLY way state moves + async fn transition_ti(&self, ti: &TiKey, from: TiState, to: TiState, try_number: u32, m: TransitionMeta) -> Result; + async fn claim_ready(&self, limit: usize) -> Result>; // oldest-first, try_number++ + async fn revert_stale_queued(&self, ti: &TiKey) -> Result; // boot reconcile, try_number-- + async fn append_history(&self, ti: &TaskInstance) -> Result<()>; // freeze a finished attempt (in-txn) + async fn record_heartbeat_ti(&self, ti: &TiKey, try_number: u32, at: DateTime) -> Result; + // XCom: self-describing inline JSON vs pointer to object storage; fingerprint-checked reads + async fn put_xcom(&self, ti: &TiKey, key: &str, is_return: bool, fp: &str, v: XComRef) -> Result<()>; // upsert + async fn get_xcoms_for(&self, run: &RunId, task_ids: &[&str]) -> Result; + async fn clear_xcom_for_ti(&self, ti: &TiKey) -> Result<()>; // retry-clear (same txn) + // …plus sweeps (find_retry_due / find_stuck_queued / find_dead_tis), watermarks, + // liveness, human-action audit, auth, retention. Full list: docs/data-model.md §8 +} + +pub enum XComRef { + Inline(Vec), // serde_json bytes, small (< ~64 KB) + Pointer { uri: String, size: u64, sha256: [u8; 32] }, // object store +} +``` + +Tables (SQLite/Turso): `dag`, `dag_run`, `task_instance`, `task_instance_history`, `xcom`, `job` +(scheduler liveness display), `event_log` (human actions only), `user`/`session`. Full DDL, ERD, state +machines, and index design: **[data-model.md](./data-model.md)**; schema of record: +[`migrations/V1__init.sql`](../migrations/V1__init.sql). + +### 6.1 Why the scheduler owns the DB (the Turso reality) + +Turso — the Rust rewrite of SQLite (the `turso` crate) — is **embedded / in-process only**: no standalone +network server daemon; its beta MVCC buys concurrent writers *within one process*, not across pods. So +**exactly one process may hold the writable DB.** Maestro makes that the organizing principle: + +- The **scheduler owns the embedded Turso DB and is the sole writer.** The DB file lives on a volume only + the scheduler mounts. +- The scheduler exposes an **HTTP/JSON control-plane API** (same axum server as the UI). Task pods and + browsers are clients; **nothing else ever opens the DB.** +- Because there is one writer, Airflow's `SELECT … FOR UPDATE SKIP LOCKED` machinery is unnecessary. + Guarded atomic updates (`UPDATE … WHERE key=? AND state=? AND try_number=?`) remain — for restart + idempotency, zombie fencing (§8.5), and forward-compat with multi-writer backends. + +Since every client is this same binary, the API is plain HTTP/JSON on axum — protobuf/gRPC would buy a +language-neutral contract nobody needs at the cost of a codegen toolchain. XCom bytes travel as +`application/octet-stream` bodies. + +### 6.2 Alternate backends (kept behind the trait) + +| Option | Shape | When | +|--------|-------|------| +| **A. Scheduler owns embedded Turso** | Single writer; pods/browsers are API clients. | **Default.** | +| **B. libSQL `sqld` server** | Network SQLite server; enables scheduler standbys (with a Kubernetes Lease for election). | If you outgrow one writer. | +| **C. Turso Cloud (managed)** | Hosted DB. | Zero DB ops, external dependency. | + +The DDL is deliberately portable — plain-SQLite (`rusqlite`) is also a drop-in `StateStore` impl if you +ever want the boring engine instead of the beta one. Switching is a config change; the loop, executors, +and UI don't change. + +### 6.3 In-process concurrency: the store actor + +"Single writer" must hold *inside* the process too. One **store-actor** task owns the write connection; +the scheduler loop and all HTTP handlers submit commands over a bounded `mpsc` (overflow → HTTP 429, and +clients back off per the retry contract in §8.3). Reads (UI pages, `GetTaskInputs`) go through a small +pool of **read-only WAL connections**, bypassing the writer entirely. Division of labor: the +`ReportTaskState/Result` handlers are the **sole** writers of `running → terminal` transitions; the +executor's pod-phase events are a crash-detection fallback that the loop reconciles (§7.0) — never a +second write path for the same fact. + +--- + +## 7. The scheduler loop (Airflow-shaped, parsing removed) + +``` +every tick (~1s), through the store actor: + + 1. CREATE RUNS for each unpaused dag with a cron schedule: + fires due past last_scheduled_at -> create_dag_run (per §7.2 catchup rules) + (manual runs already exist — TriggerDagRun wrote them synchronously) + 2. ADMIT + SEED for each 'queued' run, if active_run_count(dag) < max_active_runs: + materialize TIs (state='none') from the compiled graph -> run 'queued'->'running' + 3. EVALUATE DEPS for each in-flight run (pure fn over compiled graph + persisted states): + all upstreams success -> none -> scheduled + any upstream failed/upstream_failed -> none -> upstream_failed + 4. ENQUEUE claim_ready(free global slots) (scheduled -> queued, oldest-first, try++) + -> executor.queue(ti) + 5. SWEEPS retry-due (up_for_retry past backoff -> scheduled, clear xcom) + execution-timeout (running past per-task timeout -> kill pod, up_for_retry) + dead-heartbeat reaper (running, stale heartbeat -> kill pod, up_for_retry) + stuck-queued (queued too long, no pod -> revert or retry per §7.0) + retention (purge old terminal runs, trim history, sweep sessions — default-on) + 6. FINALIZE when all leaf TIs terminal: + failed iff any leaf in {failed, upstream_failed}; else success + (skipped / removed leaves count toward success) +``` + +State machines, the lifecycle field-population table, and the claim algorithm are specified in +[data-model.md §5–§6](./data-model.md#5-state-machines). The subsections below cover what a naïve loop +gets wrong. + +### 7.0 Boot & periodic reconciliation + +The scheduler assumes it can crash at any moment, so reality (pods) and record (DB) are reconciled rather +than trusted: + +**At boot:** +1. List all pods labeled `maestro/run-id`. For each `queued`/`running` TI: + - Pod exists → **re-adopt**: re-watch it and keep the state. + - No pod and TI is `queued` → the claim never became a pod (crash between claim and create): + guarded revert `queued → scheduled` with `try_number − 1` — an attempt that never started isn't + burned. + - Pod phase `Succeeded`/`Failed` but TI still `running` and no report arrived → `running → + up_for_retry` (`error_kind='infra'`). **Never trust exit-0 without a report** — the XCom may be + missing. +2. Run redeploy reconciliation (§7.1) for every non-terminal run. + +**Every ~60 s:** re-list pods (watches drop events); delete orphan pods whose TI is terminal; apply the +same phase-vs-state rules. + +**Pod-create failure** (quota, admission webhook): `queued → up_for_retry` (`error_kind='infra'`), normal +backoff. + +### 7.1 Redeploy semantics (new binary, in-flight runs) + +A rollout replaces the scheduler *and* the image future pods run. **In-flight runs adopt the current +graph** (pinning runs to old images is Temporal-grade machinery this project explicitly rejects). On boot, +for each non-terminal run whose `code_version` ≠ the current SHA: + +- Task instances whose `task_id` no longer exists in the graph → `removed` (terminal; best-effort pod + delete). Finalization counts `removed` leaves toward success. +- Tasks in the new graph with no TI row → materialized as `state='none'` so the run can complete under + the new shape (a new leaf can't hang finalization). +- DAGs deleted from code entirely: `dag.is_stale = 1`; live TIs → `removed`; the run → `failed`; one + `event_log` row. + +**Type drift is detected, not guessed at:** each task's compiled **IO fingerprint** travels with its XCom +writes; `GetTaskInputs` compares stored vs. expected and fails the consumer TI (`error_kind='infra'`) +rather than deserializing bytes from a type that no longer exists. Inline JSON keeps even the failure +readable. + +For deploys that change XCom types on purpose, the clean path is **drain-then-deploy**: `maestro +scheduler --drain` stops creating runs and claiming tasks, lets running TIs finish, then exits +([deployment.md](./deployment.md)). `maestro validate` in CI catches accidental task renames/removals +before they strand anything. + +### 7.2 Time semantics + +- **`logical_date` is the cron fire instant, and the run executes at that instant.** This is a loud, + intentional divergence from Airflow's data-interval model (where the run for interval N executes at + N+1) — the single most misunderstood thing in Airflow. A task that needs "the period I cover" derives + it from `logical_date` and its own schedule. +- **`catchup=false`** (default): if the scheduler was down across several fires, only the **latest** + missed fire is created; the watermark advances past the rest. **`catchup=true`**: all missed fires are + created, admission-gated by `max_active_runs` (step 2), so a week of downtime doesn't stampede. +- **DST** (cron evaluated in `dag.timezone` via `chrono-tz`, stored UTC): nonexistent local times are + skipped; ambiguous local times fire at the first occurrence. Pinned by tests. + +--- + +## 8. Execution + +```rust +#[async_trait] +pub trait Executor: Send + Sync { + async fn queue(&self, ti: TaskInstance) -> Result<()>; + fn take_events(&mut self) -> mpsc::Receiver; // pod-phase fallback signals (single receiver) + async fn kill(&self, ti: &TiKey) -> Result<()>; // reaper / timeout / removed + async fn shutdown(&self); +} +``` + +Two impls, no more. + +### 8.1 `LocalExecutor` (dev) + +Resolves the closure from the registry, builds a `TaskContext` over the in-process store, and +`tokio::spawn`s it (CPU-bound → `spawn_blocking`), bounded by `--slots`. XCom still serde-round-trips +through the store, so **dev catches serialization bugs**, not prod. + +### 8.2 `KubernetesExecutor` (prod) + +On `queue(ti)`, via `kube-rs`: + +1. **Create a Pod** — `image: maestro:` (the scheduler's own tag), + `args: [execute-task, --dag-id, …, --task-id, …, --run-id, …, --try, N]`, env: + `MAESTRO_CONTROL_PLANE=maestro-scheduler:8080` and `MAESTRO_TOKEN` (a shared bearer token from a + Secret). Resources / nodeSelector / tolerations come from the task's compiled definition — they are + not persisted per-TI. +2. **Watch** pod phase by label selector — a *fallback* crash signal only. The authoritative result is + the pod's `ReportTaskResult`; phase catches OOMKills and crashes-without-report (§7.0 rules). +3. **GC, asymmetric** (Argo's production lesson): **succeeded** pods are deleted as soon as logs are + shipped; **failed** pods are kept for a configurable delay so `kubectl describe/logs` forensics work. + +### 8.3 Inside a task pod (`maestro execute-task`) + +The pod talks to the scheduler through a narrow **`TaskClient`** — five HTTP verbs, not a `StateStore` +(a pod has no business seeing `purge_terminal_runs` or user management): + +``` +GetTaskInputs · ReportTaskState(Running) · Heartbeat · PutXCom · ReportTaskResult +``` + +``` +1. resolve instance -> impl -> closure from the compiled registry (no parse, no fetch) +2. GetTaskInputs -> upstream return values + inline -> JSON bytes, fingerprint-checked + pointer -> {uri, sha256}; download from object store, verify hash +3. build TaskContext { inputs, run_id, logical_date, try_number, params } +4. ReportTaskState(Running) [carries try_number] +5. spawn the HEARTBEAT loop: Heartbeat(TiKey, try) every 30s for the duration of step 6 + (the reaper threshold is 5x the interval — without this loop, every task longer + than the threshold would be falsely reaped) +6. result = run_fn(ctx).await +7. Ok(v): small -> PutXCom(inline json); large -> upload blob + PutXCom(pointer) + ReportTaskResult(Success) + Err(e): ReportTaskResult(Failed { message, retryable: !e.is::() }) +8. upload the task's tracing output to object storage; exit 0/1 +``` + +**Client retry contract:** every RPC retries with exponential backoff + jitter; `ReportTaskResult` keeps +trying for a bounded window (default 10 min — a scheduler restart is invisible to pods). If reporting +ultimately fails, the pod **exits non-zero** so its phase tells the truth. `ReportTaskResult` is +idempotent for the same `(TiKey, try_number)`. + +### 8.4 XCom across pods, and large payloads + +- **Across pods:** mediated by the control-plane API — pods never touch the DB. A downstream pod's + `GetTaskInputs` returns what the upstream pod's `PutXCom` persisted. +- **Large payloads:** above the inline threshold (default 64 KB), the producer uploads the blob to object + storage (`object_store`: S3/GCS/MinIO/local-fs) under `xcom////` and stores a + pointer row. The consumer downloads and verifies the hash. The DB carries pointers; the bucket carries + bytes. Cleanup is a **prefix delete** of `xcom//` when the run is purged — deterministic, no + reference-sweeping. (The sha256 is for integrity; keys embed the run id, so there is no cross-run dedup + — by design.) + +### 8.5 Fencing stale attempts + +A reaped-but-actually-alive pod (network partition) must not corrupt the attempt that replaced it: + +1. Every report/heartbeat carries `try_number`; the server rejects mismatches (HTTP 409). +2. Every state guard includes `try_number` (`… WHERE state='running' AND try_number=?`). +3. The reaper deletes the pod (best-effort) *before* transitioning `running → up_for_retry`. + +### 8.6 Failure classification & timeouts + +Tasks return `anyhow::Result`; by default an `Err` is **retryable**. `maestro::bail_fatal!` (a `Fatal` +marker in the error chain) makes it **non-retryable** — straight to `failed`, no retry burn. The executor +classifies what tasks can't: `oom` (OOMKilled phase), `infra` (crashed without reporting); the scheduler +classifies `timeout` (the per-task `execution_timeout` sweep kills the pod of a live-but-stuck task — +heartbeats alone can't catch a hang) and `upstream`/`cancelled`. + +--- + +## 9. The web UI (embedded) + +Served by the scheduler's own axum server (`/ui`, plus `/api/v1/…` for programmatic access) — no separate +service, no serialized-snapshot indirection: + +| View | Source | +|------|--------| +| DAG list + **graph** | The **live in-memory registry** (nodes/edges/schedule) — always exactly what this binary runs. | +| DagRuns, TaskInstance grid / Gantt, attempt history | state tables via the read-only connection pool | +| Logs | object storage (per-attempt keys) | +| **XCom viewer** | inline values render directly (they're JSON); pointers show uri + size + hash | +| **Manual trigger** | `TriggerDagRun` — writes the `dag_run` row synchronously; durable before the HTTP response returns | +| **Login** | username + argon2 verify → session cookie (hash stored, not the token). No roles — every user is an operator. | + +Rendering the graph of a *historical* run executed under an older SHA is deferred (§13); the UI badges +`removed` tasks in old runs instead. + +--- + +## 10. Topology + +### 10.1 Production (Kubernetes) + +``` + Browser ── HTTPS ──┐ + ▼ + ┌──────────────────────────────────────────────────────────────────────────────┐ + │ maestro scheduler (StatefulSet, replicas: 1 — crash-restart IS the HA) │ + │ │ + │ ┌──────────────┐ ┌─────────────────────────────┐ ┌─────────────────────┐ │ + │ │ Scheduler │ │ axum :8080 │ │ DagRegistry │ │ + │ │ loop (§7) │──▶│ /ui (web UI) │◀──│ (compiled-in) │ │ + │ │ + sweeps │ │ /api/v1 (TaskClient + │ │ graph · cron · │ │ + │ │ + retention │ │ trigger/reads) │ │ closures · timeouts │ │ + │ └──────┬───────┘ └──────────────┬──────────────┘ └─────────────────────┘ │ + │ │ store actor │ read-only WAL pool │ + │ ▼ (single write │ │ + │ ┌─────────────────── connection) ─▼──────────┐ │ + │ │ Turso EMBEDDED (RW) on a PVC (RWO — only │ │ + │ │ this pod mounts it; the volume IS the lock)│ │ + │ └────────────────────────────────────────────┘ │ + └──────┬────────────────────────────────────────────────┬───────────────────────┘ + │ kube-rs: create / watch / reconcile / GC │ HTTP/JSON (5 verbs, bearer token) + ▼ │ + ┌─────────────────┐ ┌──────────────────────────── ┼──────────────────────┐ + │ Kubernetes API │ │ task pods (one per TaskInstance, ephemeral) │ + └─────────────────┘ │ image=maestro: args=[execute-task …] │ + │ registry lookup -> run -> heartbeat -> report │ + └───────────────┬──────────────────────────────────── ┘ + │ large XCom blobs · logs · DB backups + ▼ + ┌────────────────────────────┐ + │ Object Store (S3/GCS/MinIO) │ + │ xcom//… · logs/… · │ + │ backups/.db │ + └────────────────────────────┘ +``` + +One service, ephemeral pods, a bucket. That's the whole production system. + +### 10.2 Dev / local (single process, no cluster) + +``` + $ maestro dev --dag-id sales_etl --trigger + ┌──────────────────────────────────────────────────────────────┐ + │ ONE OS process (maestro dev) │ + │ DagRegistry -> scheduler loop (§7, same code) -> LocalExecutor│ + │ Turso EMBEDDED (:memory: or ./maestro.db, auto-migrated) │ + │ axum web on localhost:8080 (same UI) │ + │ Large XCom -> object_store LocalFileSystem (./_artifacts) │ + └──────────────────────────────────────────────────────────────┘ +``` + +Dev and prod differ in exactly two injected impls: the `StateStore` and the `Executor`. Everything else — +loop, sweeps, state machine, serialization — is byte-identical. **Dev is a faithful rehearsal of prod.** + +--- + +## 11. End-to-end data flow (one task, prod) + +``` +[1] Upstream `extract` is success. Dep-eval: transform none -> scheduled. +[2] claim_ready: scheduled -> queued (try_number 1), oldest-first within free slots. +[3] KubernetesExecutor creates the pod: + image=maestro: args=[execute-task --dag-id sales_etl --task-id transform --run-id R --try 1] +[4] Pod: registry lookup (instance transform -> impl transform -> closure). No parsing. +[5] Pod: GetTaskInputs -> extract's return value (inline JSON, fingerprint-checked; + or pointer -> download from bucket, sha256-verified). +[6] Pod: ReportTaskState(Running, try=1) -> queued -> running (guarded, fenced). +[7] Pod: heartbeats every 30s while the user fn runs. +[8] Pod: CleanRows produced -> PutXCom (inline json / pointer) -> ReportTaskResult(Success, try=1). + Scheduler: running -> success + history row, same transaction. Pod exits 0. +[9] Next tick: `load` becomes scheduled. Succeeded pod GC'd once logs are in the bucket. +``` + +If the scheduler restarts anywhere in [4–8], the pod's RPCs retry with backoff until it returns; if the +pod dies instead, the reaper or the phase-watch turns the attempt into `up_for_retry` and the next try is +fenced by `try_number`. In dev, [3–8] collapse to `tokio::spawn` against the same store — same +transitions, same serde, no pods. + +--- + +## 12. Crate layout + +| Crate | Contents | +|-------|----------| +| `maestro` | Prelude, `run()`, re-exports. **The one thing in your `Cargo.toml`.** | +| `maestro-macros` | `#[maestro::task]`: typed handle, erased impl, IO fingerprint, `inventory::submit!`. | +| `maestro-core` | `TaskIo`, `TaskNode`, `Dag::builder`/`DagSpec`, `ErasedTask`, `TaskContext`, `StateStore` trait, `Registry`, `TaskClient`. | +| `maestro-scheduler` | Loop + sweeps + reconciliation + the axum server (UI + API) + auth. | +| `maestro-executor` | `Executor` trait, `LocalExecutor`, `KubernetesExecutor` (`kube-rs`). | +| `maestro-store-turso` | The default `StateStore` impl (+ optional `-libsql`, `-rusqlite` siblings). | +| `migrations/` | Versioned SQL schema (`V1__init.sql`, …) — see [data-model.md](./data-model.md). | + +Recommended crates: `clap`, `tokio`, `tracing`; `turso`; `axum` + `tower-http` + `argon2`; `kube` + +`k8s-openapi`; `object_store` + `sha2`; `serde` + `serde_json`; `cron`/`saffron` + `chrono` + `chrono-tz`; +`async-trait`; `inventory`. (No tonic/prost — the control plane is HTTP/JSON because every client is this +same binary.) + +--- + +## 13. Deliberately deferred + +Conscious choices, each with its re-entry path: + +- **The `#[maestro::dag]` TaskFlow-style macro.** Requires an AST rewrite with real design decisions: the + task fn and its DSL form can't share one signature, so the macro must rewrite statement-position calls + `f(args…)` → `f::bind(&mut dag, args…)` against a generated companion in the task's module namespace, + define what non-task calls inside the body mean, and type placeholders (`Wire`) distinct from node + handles. Sketched so it can be built later; the builder is the surface until then. +- **Branching.** `skipped` is reserved in the schema. The static-graph-friendly design: a task returns an + enum; edges are tagged by variant; the scheduler skips the unselected variant's subtree; joins across + branches need a trigger-rule story — which is why it's deferred rather than half-shipped. +- **Pools, priority, per-DAG task caps.** Additive migration if global slots + `max_active_runs` ever + prove insufficient for one team. +- **Event-driven / deferrable tasks.** A later migration adds `trigger`, `task_instance.trigger_id`, and + the `deferred` state. +- **Backfill UX** (and the `backfill` run_type). +- **Historical-graph rendering** (a snapshot table) — the UI renders the live registry until then. +- **Per-TI capability tokens.** V1 uses a shared bearer token + server-side `(state, try_number)` checks; + per-instance tokens are a hardening upgrade. +- **Scheduler standbys.** Only meaningful with a network DB backend (B/C); arrives with a Kubernetes + Lease for election. Under the default backend the RWO volume is the lock and crash-restart is the HA. +- **Turso maturity.** The engine is beta; the docs assume WAL, not beta MVCC. Backups are built in + (periodic `VACUUM INTO` the bucket — [deployment.md](./deployment.md)); `rusqlite` is a drop-in + fallback behind the same trait if the timeline slips. +- **A compiling reference workspace** to prove the SDK ergonomics end-to-end. + +--- + +## 14. How the design honors the requirements + +| Requirement | Mechanism | +|-------------|-----------| +| **No runtime parsing** | DAGs are compiled types collected by `inventory` at link time; every role reads the same in-memory `Registry`; the UI renders the live graph (§4, §9). The build is the parse; `validate` is the lint; both complete before deploy (§1.1, §5.2). | +| **No dynamic task creation** | A task is a type; instances are declared in the builder and fixed at boot; there is no API from runtime data to a task (§5.3). | +| **One binary, dev + Kubernetes** | Same artifact, four roles; dev and prod differ only in the injected `StateStore`/`Executor` (§3, §10). | +| **Lightweight, not a bloated scheduler** | One prod service + pods + a bucket; two concurrency knobs; HTTP/JSON instead of a proto toolchain; state-machine rows instead of event-sourced replay (§1, §6.1). | +| **Typed context passing** | Compile-time-checked edges within a build; fingerprint-checked, self-describing JSON across builds (§5.1, §7.1). | +| **Turso as state backend** | Embedded Turso owned solely by the scheduler; portable DDL keeps `sqld`/cloud/`rusqlite` one impl away (§6). | +| **Survives reality** | Boot/periodic reconciliation, durable triggers, heartbeats + reaper + timeouts, try-number fencing, defined redeploy semantics (§7.0–§7.2, §8.5). | diff --git a/docs/authoring-guide.md b/docs/authoring-guide.md new file mode 100644 index 0000000..6dd59a8 --- /dev/null +++ b/docs/authoring-guide.md @@ -0,0 +1,431 @@ +# Maestro — Authoring Guide + +> How you write a DAG in Rust. For the system design see [architecture.md](./architecture.md); for +> running it, [deployment.md](./deployment.md). + +The mental model: **you write async Rust functions, annotate them with `#[maestro::task]`, and wire them +into a DAG with a typed builder.** The compiler checks that data flowing between tasks type-matches. The +compiled binary "just knows" every DAG because compiling *was* the parsing step: there is no runtime +parse, and there is no way to create tasks from runtime data — the second fact is why the first is +possible. What compiles (and validates) is what runs +([architecture.md §1.1](./architecture.md#11-compilation-is-the-parse)). + +--- + +## 1. Add Maestro to your project + +Your DAGs live in *your own* binary crate: + +```toml +# Cargo.toml +[package] +name = "my-pipelines" +edition = "2021" + +[dependencies] +maestro = "0.1" +serde = { version = "1", features = ["derive"] } +anyhow = "1" +``` + +```rust +// src/main.rs +use maestro::prelude::*; + +// ... your tasks and DAGs (below) ... + +fn main() -> anyhow::Result<()> { + maestro::run() // dispatches scheduler / execute-task / dev / migrate / validate / run-task +} +``` + +That's the whole integration. The resulting binary *is* the scheduler (with the web UI embedded), *is* +the per-task pod entrypoint, and *is* your dev runner — behavior selected by subcommand +([architecture.md §3](./architecture.md#3-one-binary-four-roles)). + +--- + +## 2. A task is an annotated async function + +A task takes a `&TaskContext` first, an optional **typed input** second, and returns a **typed output**. +Inputs and outputs must be `Serialize + Deserialize` (they cross process boundaries in prod). + +```rust +use maestro::prelude::*; +use serde::{Serialize, Deserialize}; + +#[derive(Serialize, Deserialize, Clone)] +pub struct RawRows { pub csv: String, pub source_uri: String } + +#[derive(Serialize, Deserialize, Clone)] +pub struct Record { pub id: u64, pub amount_cents: i64 } + +#[derive(Serialize, Deserialize, Clone)] +pub struct CleanRows { pub rows: Vec, pub dropped: u64 } + +#[derive(Serialize, Deserialize, Clone)] +pub struct LoadReport { pub inserted: u64, pub table: String } + +/// EXTRACT — no upstream input; produces RawRows. +#[maestro::task] +async fn extract(ctx: &TaskContext) -> anyhow::Result { + let p: SalesEtlParams = ctx.params()?; // typed params (see §6) + if !p.source_uri.starts_with("https://") { + maestro::bail_fatal!("bad source_uri: {}", p.source_uri); // don't burn retries on config errors + } + let csv = reqwest::get(&p.source_uri).await?.text().await?; // plain user code — bring your own clients + Ok(RawRows { csv, source_uri: p.source_uri }) +} + +/// TRANSFORM — consumes RawRows; produces CleanRows. +#[maestro::task] +async fn transform(ctx: &TaskContext, input: RawRows) -> anyhow::Result { + ctx.info(format!("parsing {} bytes from {}", input.csv.len(), input.source_uri)); + let (rows, dropped) = parse_and_clean(&input.csv)?; + Ok(CleanRows { rows, dropped }) +} + +/// LOAD — consumes CleanRows; produces LoadReport. +#[maestro::task] +async fn load(ctx: &TaskContext, input: CleanRows) -> anyhow::Result { + let p: SalesEtlParams = ctx.params()?; + let inserted = insert_rows(&p.target_table, &input.rows).await?; + Ok(LoadReport { inserted, table: p.target_table }) +} +``` + +Notes: + +- **One input type per task.** Multiple upstreams → tuple input via `join` (§5). One input keeps the + type-checked edge relation clean. +- **`Err` = retryable failure by default;** `maestro::bail_fatal!` (or wrapping any error in + `maestro::Fatal`) marks it non-retryable — the task goes straight to `failed` without burning retries. + Use it for config/validation errors that a retry can't fix. +- **Tasks execute at least once.** A task can run to completion and *still* be retried (its success + report can be lost to a crash). Write side effects idempotently — e.g. upsert by key, or make the + target write conditional on `(run_id, task_id)`. +- The `#[maestro::task]` macro generates a typed handle module (`extract::task()`), the erased runtime + wrapper, a schema **fingerprint** of the IO types (used to detect type drift across deploys — + [architecture.md §7.1](./architecture.md#71-redeploy-semantics-new-binary-in-flight-runs)), and the + registry entry. You never write registration code. + +--- + +## 3. Define the DAG with the typed builder + +```rust +fn sales_etl() -> DagSpec { + Dag::builder("sales_etl") + .schedule(Cron::new("0 6 * * *")) // 06:00 daily, in .timezone(...) if set + .catchup(false) + .default_retries(3) + .params::() // typed param schema (see §6) + .add(extract::task()) + .add(transform::task().after(extract::task())) + .add(load::task().after(transform::task()) + .retries(1) + .execution_timeout(mins(30))) + .build() +} +maestro::register_dag!(sales_etl); +``` + +- `.add(node)` declares a task **instance**; `.after(upstream)` declares a type-checked edge. Fan-out is + just two `.after(...)` calls naming the same upstream (§5). +- Per-instance knobs chain on the node: `.retries(n)`, `.retry_delay(secs(30))`, + `.execution_timeout(mins(30))`, `.with_id("...")` (§4). +- **Sinks are all out-degree-0 tasks** — multiple sinks are normal; the run finalizes from all of them. +- `Dag::builder(...).root(a).then(b).then(c)` exists as sugar for plain chains; it lowers to the same + `add/after` graph. +- Validation (cycles, duplicate instance ids, unknown/defaultless params, schedule sanity) runs at every + process boot **and** in CI via `maestro validate` (§7). + +### The compile-time check + +Each task's handle carries its input/output types; `.after()` only compiles when they line up: + +```rust +transform::task().after(extract::task()) // ✅ extract::Out (RawRows) == transform::In (RawRows) +load::task().after(extract::task()) // ❌ compile error: + // expected `TaskIo`, found `Out = RawRows` +``` + +Change a task's output type and every downstream consumer stops compiling until you fix the wire. That's +the point. (Two honest limits: the guarantee is per-build — across a *deploy*, drift is caught at runtime +by the fingerprint check, loudly; and type-checking isn't identity-checking — two upstreams with the same +output type can be crossed in a join, so prefer distinct newtype outputs. See +[architecture.md §5.1](./architecture.md#51-compile-time-checked-data-passing-the-headline-feature--stated-honestly).) + +Read the ❌ above with Airflow eyes: it is a **parse error** — surfaced by `rustc` at your desk instead of +as a wrong-shape XCom at 6am in another pod. What the compiler proves: every task exists, every edge +type-matches, every IO type serializes. What it can't see as types — cycles, duplicate instance ids, +missing param defaults — `maestro validate` proves (§7). Green build + green validate is the entire +pre-deploy proof. + +--- + +## 4. Task reuse: implementations vs. instances + +A `#[maestro::task]` fn is a reusable **implementation**; each `.add()` creates an **instance** with an id +(default: the fn name). `.with_id()` lets one implementation appear twice in a DAG — or in many DAGs: + +```rust +fn multi_region_etl() -> DagSpec { + let eu = extract::task().with_id("extract_eu"); + let us = extract::task().with_id("extract_us"); + Dag::builder("multi_region_etl") + .schedule(Cron::new("0 6 * * *")) + .params::() + .add(eu.clone()) + .add(us.clone()) + .add(transform::task().with_id("transform_eu").after(eu)) + .add(transform::task().with_id("transform_us").after(us)) + .build() +} +``` + +Instance ids must be unique within a DAG (`validate()` enforces it). In the database and the UI, the +instance id is `task_id` and the implementation is `impl_id` +([data-model.md §4](./data-model.md#4-tables)). Reusable tasks should take their configuration from their +**input type**, not by reaching into DAG-level params by name — that keeps them DAG-agnostic. + +--- + +## 5. Fan-out and fan-in + +```rust +#[derive(Serialize, Deserialize, Clone)] +pub struct AuditReport { pub anomalies: u64 } + +#[derive(Serialize, Deserialize, Clone)] +pub struct Summary { pub ok: bool } + +/// Fan-in: tuple input, one slot per joined upstream (order matches join(...)). +#[maestro::task] +async fn reconcile(ctx: &TaskContext, input: (LoadReport, AuditReport)) -> anyhow::Result { + let (load_report, audit) = input; + Ok(Summary { ok: audit.anomalies == 0 && load_report.inserted > 0 }) +} + +fn sales_etl_v2() -> DagSpec { + Dag::builder("sales_etl_v2") + .schedule(Cron::new("0 6 * * *")) + .params::() + .add(extract::task()) + .add(transform::task().after(extract::task())) + .add(load::task().after(transform::task())) // fan-out: transform feeds + .add(audit_rows::task().after(transform::task())) // both load and audit_rows + .add(reconcile::task().join(load::task(), audit_rows::task())) // fan-in, type-checked per slot + .build() +} +``` + +`join(a, b)` requires `In = (A::Out, B::Out)` — each slot is type-checked. Higher arities (`join3`, …) +exist. Because tuple slots are positional, two upstreams with the *same* output type can be swapped +silently — use distinct newtypes for join inputs. + +There is **no branching construct** in V1 — no conditional skipping of subtrees. The graph you declare is +the graph that runs; put conditional logic *inside* a task. (A static-graph-friendly branching design is +sketched in [architecture.md §13](./architecture.md#13-deliberately-deferred).) + +--- + +## 6. Params, the context, and errors + +### Typed params + +Declare a params struct once; defaults are required for any DAG with a schedule (cron runs have no +trigger to supply values — boot validation enforces this): + +```rust +#[derive(maestro::DagParams, Serialize, Deserialize, Clone)] +pub struct SalesEtlParams { + #[param(default = "https://example.com/sales.csv")] + pub source_uri: String, + #[param(default = "sales")] + pub target_table: String, +} +``` + +Read them **as the struct** — field access, so a typo or a type drift is a compile error, not a runtime +failure in a pod: + +```rust +let p: SalesEtlParams = ctx.params()?; // one deserialize; values validated at trigger time +let uri = &p.source_uri; +``` + +Manual triggers supply overrides as JSON (UI or CLI), validated against the schema before the run is +created. `ctx.param::("key")` exists as an escape hatch; an undeclared key is a hard error, not a None. + +### `TaskContext` + +| Method | Purpose | +|--------|---------| +| `run_id()` / `logical_date()` | Run identity. `logical_date` is the cron fire instant — Maestro runs *at* the fire time, unlike Airflow's interval model ([architecture.md §7.2](./architecture.md#72-time-semantics)). | +| `try_number()` / `is_last_retry()` | Retry bookkeeping (attempt 1 is the first run). | +| `params::

()` / `param::(key)` | Typed params / escape hatch. | +| `info/warn/error(msg)` | Structured logging, shipped to object storage per attempt. | +| `pull::()` | Typed escape-hatch read of an upstream's output (id + type from the handle). | +| `xcom_push::(key, &v)` / `xcom_pull` | Keyed extra values. Attempt-scoped: cleared if the attempt fails and retries. | + +You normally never pull/push yourself — the generated glue moves your declared inputs/outputs. There is +deliberately no HTTP client, DB pool, or cloud SDK on the context: bring your own clients in task code. + +### Errors, retries, timeouts + +```rust +.retries(5).retry_delay(secs(30)) // per-instance; default_retries(n) on the builder +.execution_timeout(mins(30)) // scheduler kills the pod past this → up_for_retry(timeout) +``` + +- Plain `Err(e)` → **retryable**: `up_for_retry` with backoff until retries are exhausted, then `failed`. +- `maestro::bail_fatal!(...)` / `Err(maestro::fatal(e))` → **non-retryable**: straight to `failed`. +- Infrastructure failures (OOMKill, pod crash without report, dead heartbeat, timeout) are classified by + the platform and retried like ordinary failures — your task doesn't handle them. +- Outputs are addressed by `(dag, task, run)` and each retry starts from a clean slate (the failed + attempt's outputs are cleared; its record is preserved in attempt history). + +--- + +## 7. Testing + +### Unit-test a task — it's just an async fn + +`#[maestro::task]` leaves your fn callable under its own name. `TaskContext::test()` builds a context with +no scheduler, no DB: + +```rust +#[tokio::test] +async fn transform_drops_bad_rows() -> anyhow::Result<()> { + let ctx = TaskContext::test() + .params(SalesEtlParams::default()) + .logical_date("2026-07-11T06:00:00Z") + .build(); + let out = transform(&ctx, RawRows { csv: FIXTURE.into(), source_uri: "test".into() }).await?; + assert_eq!(out.dropped, 2); + Ok(()) +} +``` + +### Integration-test a DAG — in-process, real serde, `:memory:` store + +The dev runner is exposed as a library harness (same `LocalExecutor` + embedded store as `maestro dev`): + +```rust +#[tokio::test] +async fn sales_etl_end_to_end() -> anyhow::Result<()> { + let run = maestro::test::harness() // :memory: Turso, LocalExecutor + .run("sales_etl") + .param("source_uri", "file://fixtures/sales.csv") + .await?; + assert_eq!(run.state("load"), TiState::Success); + let report: LoadReport = run.output("load")?; // typed, serde-round-tripped exactly like prod + assert!(report.inserted > 0); + Ok(()) +} +``` + +### Validate graphs in CI + +```bash +cargo run -- validate # cycles, duplicate instance ids, params without defaults on + # scheduled DAGs, schedule sanity — non-zero exit on any failure +``` + +Run it in CI so a broken graph fails the pipeline, not the scheduler at 6am. It also catches accidental +task renames/removals that would strand in-flight runs' tasks as `removed` on deploy +([architecture.md §7.1](./architecture.md#71-redeploy-semantics-new-binary-in-flight-runs)). + +The build is the parse; `validate` is the lint. Together they do the whole discovery-and-verification job +Airflow performs at runtime, forever — moved to before deploy, run once. + +--- + +## 8. The "no dynamic tasks" rule + +Tasks are hard-coded. A task is a *type* generated by `#[maestro::task]` — there is no API that turns a +`String` or a runtime list into a task, and the registry is built once at boot, so the graph is fixed for +the life of the process. The Airflow pattern of generating 1,000 DAGs from a database query at parse time +is impossible by construction: there is no parse time. More precisely, **parse time *is* compile time** — +and that identity is exactly what this rule buys. The moment a task could be built from a runtime +`String`, the graph would be knowable only by executing code at runtime, and the runtime parser — the +re-parse loop, the import-error table, the 6am surprises — would have to come back. + +If you need N similar branches, declare them (`.with_id("shard_a")`, `.with_id("shard_b")`, … — they're +few and known), or make it *one* task that processes a batch internally. The strictness is the feature: +the source is the complete, honest description of what runs. **We must know what we get — so the compiler +makes sure we do** ([architecture.md §1.1](./architecture.md#11-compilation-is-the-parse)). + +--- + +## 9. Run it locally + +No cluster, no setup — dev mode auto-migrates its own database: + +```bash +# run one DAG end-to-end, in-process, right now +cargo run -- dev --dag-id sales_etl --trigger \ + --param source_uri=file://fixtures/sales.csv \ + --param target_table=sales_test + +# keep a dev scheduler running on its cron, web UI on :8080 +cargo run -- dev --web + +# iterate on ONE task without rerunning the whole DAG: +# inputs come from the last local run's stored outputs (or --input fixture.json) +cargo run -- run-task --dag-id sales_etl --task-id transform +cargo run -- run-task --dag-id sales_etl --task-id transform --input fixtures/raw_rows.json + +# tight loop +cargo watch -x 'run -- dev --dag-id sales_etl --trigger' +``` + +`--param k=v` values are parsed as JSON first (numbers, bools, arrays), falling back to plain strings, +then validated against the declared schema — a type mismatch fails *before* the run starts, naming the +expected type. + +Dev serde-round-trips every XCom exactly as prod does, so a type that "works in memory" but fails to +serialize is caught here, not in Kubernetes. + +--- + +## 10. Cheat sheet + +```rust +use maestro::prelude::*; +use serde::{Serialize, Deserialize}; + +#[derive(Serialize, Deserialize, Clone)] +struct Foo { n: u64 } + +#[maestro::task] +async fn a(ctx: &TaskContext) -> anyhow::Result { Ok(Foo { n: 1 }) } + +#[maestro::task] +async fn b(ctx: &TaskContext, input: Foo) -> anyhow::Result<()> { + if input.n == 0 { maestro::bail_fatal!("n must be nonzero"); } // no retries for config errors + Ok(()) +} + +fn demo() -> DagSpec { + Dag::builder("demo") + .schedule(Cron::new("@hourly")) + .default_retries(2) + .add(a::task()) + .add(b::task().after(a::task()).execution_timeout(mins(5))) + .build() +} +maestro::register_dag!(demo); + +fn main() -> anyhow::Result<()> { maestro::run() } +``` + +- Task = `#[maestro::task] async fn(ctx, [input]) -> Result`; `Out: Serialize + Deserialize`. +- Wire with `.add(node.after(upstream))`; mismatched types don't compile; `join(a, b)` for fan-in. +- Reuse an implementation via `.with_id(...)`; instance ids unique per DAG. +- Params: typed struct + defaults (required if scheduled); read via `ctx.params::

()`. +- `Err` retries; `bail_fatal!` doesn't. Tasks run **at least once** — make side effects idempotent. +- Test: call the fn with `TaskContext::test()`; run DAGs with `maestro::test::harness()`; + `maestro validate` in CI. +- `cargo run -- dev --dag-id demo --trigger` runs it with zero infrastructure. diff --git a/docs/data-model.md b/docs/data-model.md new file mode 100644 index 0000000..ab4cf99 --- /dev/null +++ b/docs/data-model.md @@ -0,0 +1,363 @@ +# Maestro — State Data Model + +> The relational schema that tracks the state of every DAG run and task. The literal, runnable schema of +> record is [`migrations/V1__init.sql`](../migrations/V1__init.sql); this document explains the tables, +> the state machines, the indexes, and how the Rust `StateStore` trait maps onto them. +> +> See also: [architecture.md §6–§8](./architecture.md#6-state-the-statestore-trait-over-embedded-turso) +> (where this store sits) and [deployment.md](./deployment.md) (migrations & retention in ops). + +--- + +## 1. Principles + +- **Portable SQLite / libSQL / Turso.** Only `TEXT` / `INTEGER` / `REAL` / `BLOB`; every table is `STRICT`. + No engine-specific column types, so the same DDL runs on plain SQLite for local dev and CI. +- **Enums are `TEXT` + `CHECK (col IN (...))`.** Booleans are `INTEGER 0/1` (STRICT disallows the + `BOOLEAN`/`DATETIME` aliases). +- **Timestamps are ISO-8601 UTC `TEXT` with millisecond precision** (`'2026-07-11T06:00:00.123Z'`) — + lexically sortable, friendly to Rust `chrono`/`serde`, and precise enough that claim/reap sweeps have a + deterministic order. Durations are `INTEGER` milliseconds. +- **Natural composite keys**, with the cascade chain `dag → dag_run → task_instance → xcom`. +- **Single writer, WAL, no distributed locking.** The scheduler is the sole writer + ([architecture.md §6.1](./architecture.md#61-why-the-scheduler-owns-the-db-the-turso-reality)); state + changes are **guarded atomic updates** (`UPDATE … WHERE key=? AND state=? AND try_number=?`, checking + rows affected) — the `try_number` in the guard fences out stale attempts (a reaped-but-still-alive pod + from a previous try cannot corrupt the current one). +- **Small on purpose.** V1 gates concurrency on exactly two knobs — global `--slots` and per-DAG + `max_active_runs`. Airflow's `pool`/`pool_slots`/`priority_weight`/`max_active_tasks` are deliberately + deferred (§9): Kubernetes requests/limits already arbitrate shared resources for one team, and priority + is a symptom of too-few slots. + +## 2. What we keep vs. drop from Airflow + +Maestro's schema is modeled on Airflow's metadata DB but adapted to the project's constraints: + +- **No runtime parsing ⇒ the parsing tables are gone.** Airflow's `serialized_dag`, `dag_code`, + `import_error`, `dag_pickle`, and `dag_warning` are the fossil record of runtime parsing — each stores + a parse-stage artifact or failure. Maestro's parse product is the binary itself + ([architecture.md §1.1](./architecture.md#11-compilation-is-the-parse)): discovery happened at link + time, every role carries the parse product compiled in, and the UI renders the graph straight from the + live in-memory registry — nothing to serialize, store, or invalidate. +- **No dynamic task mapping ⇒ no `map_index`.** Task identity is exactly `(dag_id, task_id, run_id)`. +- **Impl vs. instance.** `task_id` is the *instance* id (unique within a DAG, defaults to the task fn + name, overridable via `.with_id()`); `impl_id` names the compiled *implementation* the registry resolves + to. One implementation can back many instances, across DAGs and within one DAG — Maestro's equivalent of + Airflow's operator-vs-task distinction. +- **Concurrency diet.** Only `max_active_runs` survives from Airflow's five mechanisms (plus the global + `--slots` flag, which is config, not schema). + +Everything Airflow uses to *know the state of a task* — runs, task instances, retry history, XCom, +liveness, audit — is kept. + +## 3. ERD + +``` + ┌───────────────────────────┐ + │ dag │ registry metadata + │ PK dag_id │ (schedule, catchup, is_stale, ...) + └─────────────┬─────────────┘ + │ ON DELETE CASCADE + ▼ + ┌────────────────────────────┐ + │ dag_run │ + │ PK (dag_id, run_id) │ + │ UX (dag_id, logical_date) │ + │ WHERE run_type='scheduled'│ + └───────┬────────────────┬────┘ + ON DELETE │ │ FK (dag_id,run_id) + CASCADE │ │ + ▼ ▼ + ┌──────────────────────────┐ ┌──────────────────────────────┐ + │ task_instance │ │ task_instance_history │ + │ PK (dag_id,task_id,run_id)│ │ PK (…,run_id, try_number) │ + │ impl_id → registry │ │ append-only per attempt │ + └────────────┬─────────────┘ └──────────────────────────────┘ + ON DELETE │ CASCADE + ▼ + ┌──────────────────────────┐ + │ xcom │ + │ PK (dag_id,task_id, │ + │ run_id,key) │ + │ inline(json) | pointer │ + │ + type_fingerprint │ + └──────────────────────────┘ + + Independent / cross-cutting (NOT in the cascade chain): + job — scheduler liveness display only (no election; the RWO volume is the lock) + event_log — audit of HUMAN actions only [deliberately NO FK: survives deletes] + user → session (ON DELETE CASCADE; session stores token_hash, never the token) +``` + +## 4. Tables + +Column-level DDL lives in [`migrations/V1__init.sql`](../migrations/V1__init.sql). Summaries below. + +### `schema_migration` +Migrator bookkeeping: `version` (PK), `name`, `checksum` (sha256 of the file), `applied_at`. Already-applied +files are immutable (checksum-verified). The **scheduler auto-applies pending migrations at boot** — with a +single writer there is no migration race; `maestro migrate` also runs them standalone for ops/dev. + +### `dag` — registry metadata +`dag_id` (PK), `description`, `schedule` (cron or `@daily`; NULL = manual-only), `timezone`, `start_date`, +`end_date`, `catchup`, `is_paused`, **`is_stale`** (set at boot when the dag disappeared from the compiled +registry — see redeploy semantics), `max_active_runs`, `default_retries`, `last_scheduled_at` (the catchup +high-water mark), `code_version`, `owners`, timestamps. Upserted when the scheduler registers its compiled +DAGs at boot. + +### `dag_run` — one execution of a DAG +PK `(dag_id, run_id)`. **Partial unique** `(dag_id, logical_date) WHERE run_type='scheduled'`: cron-run +creation is idempotent, while manual runs are unconstrained (identity via `run_id = manual__`) — two +manual triggers in the same second, or a manual trigger at a cron fire instant, all coexist. + +Columns: `logical_date`, `run_type` (`scheduled`|`manual`), `state` (`queued`|`running`|`success`|`failed`), +`code_version`, `conf_json`, `triggered_by`, `queued_at`/`start_date`/`end_date`, +`last_scheduling_decision`. **Manual triggers are durable by construction:** the `TriggerDagRun` API +handler writes this row synchronously and returns the `run_id` — there is no in-memory queue to lose. + +### `task_instance` — the current attempt (the heart) +PK `(dag_id, task_id, run_id)` — `task_id` is the instance id; `impl_id` names the compiled implementation. + +| Group | Columns | +|-------|---------| +| identity | `dag_id`, `task_id` (instance), `run_id`, `impl_id` (implementation) | +| **state** | `state` (`none`,`scheduled`,`queued`,`running`,`success`,`failed`,`up_for_retry`,`upstream_failed`,`skipped`*,`removed`), `try_number`, `max_tries` | +| lifecycle | `scheduled_at`, `queued_at`, `start_date`, `end_date`, `duration_ms` | +| placement | `worker` (pod name in prod / host label in dev), `exit_code` | +| retry | `retry_delay_secs`, `next_retry_at` | +| outcome | `error_kind` (`user`/`timeout`/`oom`/`infra`/`cancelled`/`upstream`/`unknown`), `error_message` | +| liveness | `last_heartbeat_at`, `updated_at` | + +\* `skipped` is **reserved** (kept in the CHECK so enabling branching later is additive); no V1 code path +produces it. Per-task config that is compiled in — pod resources, tolerations, execution timeout — is +**not** persisted per-TI: the scheduler reads it from the registry. + +### `task_instance_history` — every finished attempt +PK `(dag_id, task_id, run_id, try_number)`. A frozen snapshot appended on each finished attempt (§5), +including `max_tries` (it can change across deploys — the live row won't tell you why attempt 3 was final). +FK references **`dag_run`**, not `task_instance`, so history survives a TI being re-materialized or +`removed`, while still cascading when the run/dag is purged. + +### `xcom` — task outputs / passed values +PK `(dag_id, task_id, run_id, key)` (`key = 'return_value'` for the primary output). A **tagged union** on +`value_kind`: +- `inline` → `inline_value` BLOB. **Default encoding: `serde_json`** — self-describing, so type drift + across a redeploy fails detectably (not silent garbage) and the web UI can render values without + compiled types. The < ~64 KB inline ceiling makes JSON's size penalty irrelevant. +- `pointer` → `uri` + `size_bytes` + `sha256`; large payloads live in object storage. Compact encodings + (`postcard`/`bincode`) are allowed here, recorded in `serde_format`. + +`is_return` flags the primary return value. **`type_fingerprint`** stores a hash of a stable schema +descriptor of the producer's `Out` type (emitted by `#[maestro::task]`); `GetTaskInputs` compares it with +the consumer's expected fingerprint and **fails the TI loudly** (`error_kind='infra'`) if a redeploy +changed the type mid-run. A `CHECK` enforces the union shape. + +Write semantics: `put_xcom` is an **upsert** on the PK. On `up_for_retry → scheduled`, all xcom rows of +the TI are **deleted in the same transaction** — attempt outputs are attempt-scoped, so a failed attempt's +keyed values never leak into the next attempt's downstream reads. + +### `job` — scheduler liveness (display only) +`id`, `job_type` (`scheduler`), `state`, `hostname`, `latest_heartbeat`, `start_date`, `end_date`. This is +**not** leader election: under the default backend, the DB file sits on an RWO volume only one pod can +mount — *the volume is the lock*. V1 HA = StatefulSet `replicas: 1` + crash-restart; guarded transitions +make recovery idempotent. (A Kubernetes Lease is the failover mechanism if a network-DB backend enables +standbys — that migration reintroduces election.) + +### `event_log` — audit of human actions only +`ts`, `event` (`dag_run.triggered`, `dag.paused`, `user.login`, `run.cleared`, …), nullable +`dag_id`/`run_id`/`task_id`, `actor`, `extra` (JSON). Machine transitions are **not** mirrored here — with +a single writer, the state tables *are* the machine audit trail (`task_instance_history` records every +attempt; `dag_run.triggered_by` records provenance). No FKs: audit rows outlive the entities they describe. + +### `user` / `session` — minimal auth +`user`: `id`, `username` (unique), `password_hash` (argon2id), timestamps — **no roles** (everyone on the +team is an operator). `session`: **`token_hash`** (sha256 of the cookie token — a leaked DB backup must not +yield live sessions), `user_id` (FK, cascade), `expires_at`. + +## 5. State machines + +### TaskInstance + +``` +None ─deps met→ Scheduled ─claim (try_number++)→ Queued ─pod reports Running→ Running + │ ▲ │ │ + │ └──── revert ──────┘ (boot reconcile: claimed but no pod; │ + │ (try_number−−) or pod-create rejected → up_for_retry) │ + │ │ + ├─ any upstream failed → UpstreamFailed (terminal) ┌───────────────────┼──────────────────┐ + │ success failure reaped/timeout + │ │ ┌───────┴────────┐ (dead heartbeat, + ▼ Success retryable & │ exceeded exec +task dropped on redeploy → Removed (terminal) tries left? fatal or │ timeout) + yes │ exhausted│ │ + UpForRetry ─backoff→ Scheduled│ UpForRetry + Failed (terminal) +``` + +Terminal: `success`, `failed`, `skipped`*, `upstream_failed`, `removed` (*reserved). Key rules: + +- **Claiming** (`scheduled → queued`) increments `try_number`. The **boot reconcile** reverts + queued-with-no-pod TIs back to `scheduled` with `try_number − 1` — a claim the executor never acted on + doesn't burn an attempt. +- **Fencing:** every guard includes `try_number`; report/heartbeat calls carry it and are rejected (HTTP + 409) on mismatch, so a zombie pod from try N cannot write into try N+1. The reaper deletes the pod + (best-effort) *before* transitioning to `up_for_retry`. +- **Fatal errors** (`bail_fatal!` in task code → `retryable=false`) short-circuit straight to `failed` + regardless of remaining retries. The executor classifies `oom` (OOMKilled) / `infra` (crashed without + reporting); the scheduler classifies `timeout` (execution-timeout sweep), `upstream`, `cancelled`. + +### DagRun + +``` +queued ──(TIs materialized AND active_run_count < max_active_runs)──▶ running ──▶ success | failed +``` + +- Created in `queued` (by the cron loop or synchronously by `TriggerDagRun`). +- **Finalization** when all leaf TIs are terminal: **failed** iff any leaf ∈ {`failed`, + `upstream_failed`}; otherwise **success** (`skipped`/`removed` leaves count toward success). Sinks = + all out-degree-0 tasks; a run finalizes from all of them. + +### Lifecycle field population + +| Transition | Guard (`WHERE state=… AND try_number=…`) | Columns set | +|------------|------------------------------------------|-------------| +| materialize (insert) | — | `state='none'`, `try_number=0`, `max_tries=retries`, `impl_id`, `retry_delay_secs`, `updated_at` | +| none → scheduled | `none` | `state='scheduled'`, `scheduled_at` | +| none → upstream_failed | `none` | `state`, `end_date`, `error_kind='upstream'` → **append history** | +| scheduled → queued (`claim_ready`) | `scheduled` | `state='queued'`, `queued_at`, `try_number += 1` | +| queued → scheduled (boot revert) | `queued` + no pod | `state='scheduled'`, `try_number −= 1`, clear `queued_at` | +| queued → up_for_retry (pod-create failed) | `queued` | `state`, `error_kind='infra'`, `next_retry_at` → **append history** | +| queued → running (pod reports) | `queued` + try match | `state='running'`, `start_date`, `worker`, `last_heartbeat_at` | +| running → running (heartbeat) | `running` + try match | `last_heartbeat_at` only | +| running → success | `running` + try match | `state`, `end_date`, `duration_ms`, `exit_code=0` → **append history** | +| running → up_for_retry (retryable fail / reaped / timeout) | `running` (+ try for reports) | `state`, `end_date`, `duration_ms`, `exit_code`, `error_*`, `next_retry_at = now + backoff` → **append history** | +| running → failed (fatal, or retries exhausted) | `running` + try match | `state`, `end_date`, `duration_ms`, `exit_code`, `error_*` → **append history** | +| up_for_retry → scheduled (backoff elapsed) | `up_for_retry AND next_retry_at <= now` | `state='scheduled'`, `scheduled_at`; **clear** execution fields + `next_retry_at`; **delete the TI's xcom rows** (same txn) | +| any live → removed (redeploy dropped the task) | any non-terminal | `state='removed'`, `end_date` → **append history**; best-effort pod delete | + +### The history-append invariant + +On **every** transition into a finished attempt (`success`, `failed`, `up_for_retry`, `upstream_failed`, +`removed`), the scheduler — **in the same transaction** as the state change — inserts a snapshot into +`task_instance_history` keyed by the current `try_number`. `task_instance` always holds the *current* +attempt; history holds *every completed* attempt. The `up_for_retry → scheduled` transition then clears the +live row's execution fields and xcom rows so the next attempt starts clean. + +## 6. The claim algorithm + +`claim_ready(limit)` — the hottest write path — in one transaction: + +1. `limit` = free **global slots** (`--slots` minus TIs in `queued`/`running`). +2. Select `scheduled` TIs ordered **oldest `scheduled_at` first** (via the partial `ix_ti_scheduled`), + `LIMIT limit`. No priority tiers, no pool arbitration — deliberately (§9); starvation is fixed by + raising `--slots`. +3. Guarded-update each selected TI `scheduled → queued` (`try_number += 1`). + +Run-level admission is separate: a run leaves `queued` only while `active_run_count(dag) < +max_active_runs`, which is what stops catchup or trigger-spamming from stacking overlapping runs. + +## 7. Indexes + +Full definitions in the migration. Partial indexes keep the per-tick working set proportional to +*in-flight* tasks, not accumulated history. + +| Index | Serves | Notes | +|-------|--------|-------| +| `ix_ti_scheduled` | `claim_ready` | partial `WHERE state='scheduled'`, ordered `scheduled_at ASC` — pre-sorted for the LIMIT | +| `ix_ti_queued` | stuck-queued sweep (claimed but no pod) | partial `WHERE state='queued'` | +| `ix_ti_unfinished` | every-tick dep-eval working set | partial on in-flight states | +| `ix_ti_retry_due` | retry backoff wakeups | partial `WHERE state='up_for_retry'`, by `next_retry_at` | +| `ix_ti_running_heartbeat` | dead-pod reaper + execution-timeout sweep | partial `WHERE state='running'` | +| `ix_ti_run`, `ix_ti_state` | per-run grid; state seeks/metrics | | +| `ix_dr_dag_state`, `ix_dr_active`, `ix_dr_logical` | `max_active_runs`, finalize, catchup | | +| `ix_tih_ti` | history view (newest attempt first) | | +| `ix_xcom_return` | `GetTaskInputs` bundle | partial `WHERE is_return=1` | +| `ix_job_heartbeat` | "scheduler last seen" display | | +| `ix_event_ts`, `ix_event_dag`, `ix_session_*` | audit UI, session sweeps | | + +> Run `ANALYZE` after the first data lands (and periodically) so the planner has statistics; verify hot +> queries with `EXPLAIN QUERY PLAN`. + +## 8. `StateStore` trait mapping + +The trait ([architecture.md §6](./architecture.md#6-state-the-statestore-trait-over-embedded-turso)) is the +**scheduler's** seam to a swappable database — task pods do *not* implement it; they use the narrow +`TaskClient` (5 HTTP verbs) instead. Representative methods (all `async`, all `Result`): + +```rust +// DAG registry +register_dag(&DagMeta) / mark_stale_dags(&[dag_id]) // boot upsert + stale marking + +// DagRun lifecycle +create_dag_run(dag_id, logical_date, RunTrigger) -> DagRun // also called synchronously by TriggerDagRun +set_dag_run_state(&RunId, from, to) -> bool // guarded +list_dag_runs(dag_id, RunFilter) / active_run_count(dag_id) -> usize +advance_schedule_watermark(dag_id, last) + +// TaskInstance state machine +materialize_tis(&RunId, &[TaskSeed]) // seed state='none' from the compiled graph +transition_ti(&TiKey, from, to, try_number, TransitionMeta) -> bool // THE guarded update (+ append_history in-txn) +claim_ready(limit) -> Vec // oldest-first, try_number++ +revert_stale_queued(&TiKey) -> bool // boot reconcile: queued-no-pod, try_number-- +record_heartbeat_ti(&TiKey, try_number, at) -> bool // fenced +find_retry_due(now, limit) / find_stuck_queued(older_than) / find_dead_tis(older_than) +list_task_instances(&RunId) / append_history(&TaskInstance) / list_ti_history(&TiKey) + +// XCom +put_xcom(&TiKey, key, is_return, fingerprint, XComRef) // UPSERT +get_xcom(&TiKey, key) / get_xcoms_for(&RunId, &[task_id]) -> XComBundle // fingerprint-checked +clear_xcom_for_ti(&TiKey) // retry-clear (in the retry txn) + +// Liveness / audit / auth / retention +register_job(host) -> JobId / record_heartbeat(JobId, at) +log_event(&Event) / list_events(EventFilter) // human actions only +create_user / get_user_by_name / create_session / validate_session / delete_session +purge_terminal_runs(older_than) / trim_history(dag_id, keep_runs) / purge_expired_sessions(now) +``` + +The guard-critical methods — `transition_ti`, `claim_ready`, `revert_stale_queued`, `set_dag_run_state` — +are the only ways state moves; each returns whether its guard matched so callers detect lost races and stay +idempotent across restarts. `ReportTaskResult` is idempotent: a duplicate report for the same +`(TiKey, try_number)` whose recorded outcome already matches returns OK. + +## 9. Deliberately deferred + +- **Pools / priority / per-DAG task caps.** `pool`, `pool_slots`, `priority_weight`, `max_active_tasks` + return as an additive migration if two knobs ever prove insufficient. Until then Kubernetes + requests/limits arbitrate shared resources, and `claim_ready` stays oldest-first. +- **Deferrable / event-driven tasks.** A later migration adds a `trigger` table, a + `task_instance.trigger_id` FK, and the `deferred` state. +- **Branching.** The `skipped` state is reserved; the design (a task returns an enum, edges tagged by + variant, unselected subtrees skipped) lives in + [architecture.md §13](./architecture.md#13-deliberately-deferred). +- **Backfill.** `run_type='backfill'` returns with the backfill UX. +- **Historical-graph rendering** (`dag_snapshot`). The UI renders the live registry; drawing the exact + graph of a run executed under an older sha returns with a snapshot table if requested. +- **Config & annotations.** `variable`, `connection`, `dag_run_note` / `task_instance_note` — additive. +- **RBAC.** Roles/permissions if the team ever isn't one team. + +## 10. Operational notes + +**Per-connection PRAGMAs** (the app issues them on every open — they are not persistable in the schema): + +```sql +PRAGMA journal_mode = WAL; -- one writer + many readers +PRAGMA foreign_keys = ON; -- OFF by default in SQLite — MUST enable per connection, + -- and BEFORE any transaction opens (no-op inside one) +PRAGMA busy_timeout = 5000; +PRAGMA synchronous = NORMAL; -- safe with WAL +``` + +Run `PRAGMA foreign_key_check;` after migrating. Stay on **WAL**, not Turso's beta MVCC — the single-writer +scheduler doesn't need it. + +**Migrations.** Versioned `V__.sql`, checksum-recorded in `schema_migration`, auto-applied by the +scheduler at boot (and by `maestro migrate` standalone). Applied files are immutable; changes ship as new +files. SQLite runs DDL transactionally, so a failed migration rolls back. + +**Retention runs inside the scheduler loop** (default-on — see +[deployment.md §5](./deployment.md#5-operations)): purge terminal runs past a retention window +(cascade-drops their TIs, xcom, history), trim per-DAG history, sweep expired sessions, time-purge +`event_log`. Blob cleanup is a **prefix delete** of `xcom//` in object storage when the run is purged +— deterministic, no reference-sweeping. After large purges: `PRAGMA wal_checkpoint(TRUNCATE)` and periodic +`VACUUM`. diff --git a/docs/decisions.md b/docs/decisions.md new file mode 100644 index 0000000..b192452 --- /dev/null +++ b/docs/decisions.md @@ -0,0 +1,40 @@ +# Maestro — Implementation Decision Record + +Decisions made at implementation kickoff (2026-07-12), settled by two deep-research +passes, a fingerprint-verification pass, and design review. Doc sections referenced +are the spec of record; this file records what the docs left open. + +| # | Decision | Rationale | +|---|----------|-----------| +| D1 | **Turso is the V1 default store**; M3 opens with an empirical risk gate (apply `V1__init.sql`, exercise STRICT/partial indexes/transactional DDL/`VACUUM INTO`); any gap pivots to rusqlite behind the same trait. Pinned crate version, WAL only, bucket backups non-negotiable. A rusqlite sibling store lands in M11 as insurance. | Turso is pre-1.0 (Beta 2025-10; v0.6.1 2026-05; vendor recommends independent backups). Owner decided to keep the docs' default and pay for it with the gate + conformance suite. | +| D2 | **Store actor = boxed-closure commands** over a bounded mpsc; typed oneshot reply per command; multi-statement invariants are one closure/one txn. Loop `send().await`; HTTP `try_send` → 429. `:memory:` routes reads through the actor; file mode gets read-only WAL connections. | Avoids a 30-variant command enum that must mirror every `StateStore` method; closures capture typed senders, no downcasting. | +| D3 | **One shared task driver** (inputs → running → heartbeat → run → xcom → result) generic over a `ControlPlane` trait: `InProcess` (dev) vs `Http` (pods). | The keystone of dev/prod parity (architecture.md §10.2): the lifecycle is implemented once, byte-identical in both modes. | +| D4 | **Fingerprint = schemars 1.x extractor → Maestro-owned canonical IR → versioned hash `mfp1:`.** IO types derive `maestro::JsonSchema`. Never hash schemars' raw output. Golden-fingerprint corpus in CI as the upgrade tripwire. Escape hatches: `fingerprint = "opaque"`, `fingerprint_with`, `#[schemars(with)]`. | serde_reflection rejects untagged/flatten/`deserialize_any` (incl. `serde_json::Value`) and is single-maintainer; no extractor's raw output is version-stable, so Maestro owns the canonical form and the algorithm version either way. | +| D5 | **Ordered input slots in `DagSpec`**; glue rule: 1 slot → direct deserialize, n>1 → JSON array in slot order → tuple; macro records `TAKES_INPUT`; `validate()` enforces consistency. Ordering-only edges disallowed in V1 (name `.after_done()` reserved). | Post-erasure, nothing else can distinguish "join of two" from "one tuple-returning upstream". | +| D6 | **`impl_id` = fn name; discovery is eager**; duplicates (impl or dag) fail loudly at boot/validate. | Readable ids in DB/UI; DagSpec construction is pure and cheap. | +| D7 | **Macro hygiene:** generated code uses `::maestro::…` paths; attribute re-exported from the umbrella crate; generated `mod x` beside `fn x` mirrors visibility; macro-consuming tests live in umbrella + reference. | The tokio pattern; core can't consume its own macro (dependency cycle). | +| D8 | **UI = server-rendered askama + htmx, no Node**; read-only dev dashboard early (M8), full UI late (M11). | One-binary ethos; the UI is a read-mostly view for a handful of humans (architecture.md §3, §9). | +| D9 | **Injected `Clock` trait from day one.** | Sweep/cron/backoff/DST tests need controlled time; retrofitting is miserable. | +| D10 | **Fan-out: design slot reserved, no V1 code.** architecture.md §13 gains a deferred map-task construct (static shape, runtime cardinality; additive `map_index` migration). | Flyte's documented history: static-only designs face sustained fan-out demand; a bounded construct preserves "the scheduler already knows the graph". | + +Hard constraints from the owner (2026-07-12): + +- **Exactly two execution modes, forever — local or Kubernetes.** The `Executor` + trait is sealed; no plugin executors, no queue-based executor. +- **Env-var-driven mode selection.** Every flag has a `MAESTRO_*` equivalent; + precedence CLI > env > auto-detected default; `MAESTRO_EXECUTOR` unset + auto-detects in-cluster vs local; explicit `kubernetes` with no reachable + cluster fails fast — mode is deterministic, never guessed at runtime. + +Research-mandated safeguards (see plan for citations): + +- Registration completeness is a **testable invariant**: CI builds the exact + release profile (`release-lto`: fat LTO, codegen-units=1, musl) and runs + `__registry-check --assert-manifest`; `Registry::discover()` asserts counts + at startup. `linkme` is the documented fallback. +- Pod GC is a **bounded concurrent worker pool** with backpressure metrics + (Argo crashed at 4 cleanup workers, needed 32). +- Use `kube_runtime::watcher` (never raw `Api::watch`); the periodic 60s + re-list stays — kube-rs re-lists only on desync, not on a timer. +- The scheduler resolves its image to a **digest** at boot and launches task + pods by digest, not tag. diff --git a/docs/deployment.md b/docs/deployment.md new file mode 100644 index 0000000..cf9eb7b --- /dev/null +++ b/docs/deployment.md @@ -0,0 +1,235 @@ +# Maestro — Deployment + +> How to run Maestro locally and on Kubernetes. Design: [architecture.md](./architecture.md); writing +> DAGs: [authoring-guide.md](./authoring-guide.md); schema: [data-model.md](./data-model.md). + +The same compiled binary runs everywhere. Production is deliberately small: **one scheduler pod + +ephemeral task pods + a bucket.** Dev is the same code in one process. + +--- + +## 1. The one artifact + +Build a single binary from your DAG crate (`maestro::run()` in `main`), package it as one image, and +select the role per pod via `args`: + +```dockerfile +FROM rust:1-slim AS build +WORKDIR /src +COPY . . +ARG GIT_SHA +ENV MAESTRO_CODE_VERSION=$GIT_SHA +RUN cargo build --release + +FROM gcr.io/distroless/cc +COPY --from=build /src/target/release/my-pipelines /usr/local/bin/maestro +ENTRYPOINT ["maestro"] +``` + +```bash +docker build --build-arg GIT_SHA=$(git rev-parse --short HEAD) -t maestro:$(git rev-parse --short HEAD) . +``` + +The binary embeds its git SHA at build time (`MAESTRO_CODE_VERSION` read via `env!`, falling back to +`"dev"` for local builds). The scheduler launches task pods with **its own** image tag, so the scheduler +and every pod it creates run identical DAG code +([architecture.md §3](./architecture.md#3-one-binary-four-roles)); the SHA is also recorded on every +`dag_run` for the redeploy reconciler. + +Note what `cargo build --release` is doing in that Dockerfile: **it is the entire DAG-ingestion +pipeline.** The image does not contain pipeline definitions for the scheduler to interpret; it *is* the +parsed, verified DAG set ([architecture.md §1.1](./architecture.md#11-compilation-is-the-parse)). A green +build plus a green `maestro validate` (§4) is the complete pre-deploy proof that the graph is well-formed +— after `docker build`, there is no step at which the DAG set can change. + +--- + +## 2. Dev / local mode + +One process, zero setup — dev auto-applies migrations to its database (file or `:memory:`): + +```bash +# run one DAG end-to-end, right now +cargo run -- dev --dag-id sales_etl --trigger --param source_uri=file://fixtures/sales.csv + +# dev scheduler on its cron schedule, embedded web UI on :8080 +cargo run -- dev --web --db ./maestro.db + +# debug one task in isolation (inputs from the last local run, or a fixture) +cargo run -- run-task --dag-id sales_etl --task-id transform --input fixtures/raw_rows.json + +# CI graph validation (cycles, duplicate ids, params without defaults) +cargo run -- validate +``` + +| Concern | Dev behavior | +|---------|--------------| +| State store | Embedded Turso file or `:memory:`, auto-migrated at start. | +| Executor | `LocalExecutor` — in-process `tokio::spawn` / `spawn_blocking`, bounded by `--slots`. | +| Large XComs | `object_store` `LocalFileSystem` under `./_artifacts` (same code path as S3). | +| Web UI | Same embedded axum UI as prod, on localhost. | +| Fidelity | Same loop, same state machine, same serde round-trip — a DAG green in dev is a faithful rehearsal. | + +--- + +## 3. Production topology (Kubernetes) + +One service. The scheduler owns the embedded Turso DB (sole writer), runs the loop, and serves the web UI +and the HTTP/JSON control-plane API from a single axum server on one port +([architecture.md §10.1](./architecture.md#101-production-kubernetes)). + +``` + Browser ──HTTPS──▶ maestro scheduler (StatefulSet, replicas: 1) + ├── Turso EMBEDDED (RW) on an RWO PVC — the volume IS the lock + ├── axum :8080 → /ui + /api/v1 (task pods, bearer token) + └── kube-rs: create / watch / reconcile / GC task pods + │ + ▼ + task pods (one per TaskInstance, ephemeral) + │ + ▼ + Object store: xcom blobs · task logs · DB backups +``` + +### 3.1 Scheduler (StatefulSet — the only service) + +```yaml +apiVersion: apps/v1 +kind: StatefulSet +metadata: { name: maestro-scheduler } +spec: + replicas: 1 # crash-restart IS the V1 HA; guarded transitions make recovery idempotent + serviceName: maestro-scheduler + template: + spec: + containers: + - name: scheduler + image: maestro: + args: ["scheduler", "--db", "/data/maestro.db", + "--port", "8080", "--slots", "32", + "--executor", "kubernetes", "--image", "maestro:"] + env: + - { name: MAESTRO_OBJECT_STORE, value: "s3://maestro-artifacts" } + - name: MAESTRO_TOKEN # shared bearer token for task pods + valueFrom: { secretKeyRef: { name: maestro-token, key: token } } + ports: [{ containerPort: 8080 }] + volumeMounts: [{ name: db, mountPath: /data }] + volumeClaimTemplates: + - metadata: { name: db } + spec: { accessModes: ["ReadWriteOnce"], resources: { requests: { storage: 10Gi } } } +``` + +No init container: the scheduler **auto-applies pending migrations at boot** — with a single writer there +is no migration race. (`maestro migrate` still exists as a standalone command for ops/dev.) No separate +web Deployment: the UI is embedded. Note the RWO caveat: on node failure, recovery time includes the PVC +detach/attach — acceptable for a single-team tool; the path to standbys is a network DB backend + a +Kubernetes Lease ([architecture.md §13](./architecture.md#13-deliberately-deferred)). + +### 3.2 Task pods (created by the scheduler; shown for reference) + +```yaml +apiVersion: v1 +kind: Pod +metadata: + labels: { maestro/run-id: "", maestro/task-id: "transform", maestro/try: "1" } +spec: + restartPolicy: Never + containers: + - name: task + image: maestro: # same tag as the scheduler + args: ["execute-task", "--dag-id", "sales_etl", "--task-id", "transform", + "--run-id", "", "--try", "1"] + env: + - { name: MAESTRO_CONTROL_PLANE, value: "http://maestro-scheduler:8080" } + - name: MAESTRO_TOKEN + valueFrom: { secretKeyRef: { name: maestro-token, key: token } } +``` + +The pod resolves its task from the compiled-in registry, fetches inputs, **heartbeats every 30 s while +running**, and reports its result — all over 5 HTTP verbs with retry/backoff (a scheduler restart is +invisible to pods; if reporting ultimately fails the pod exits non-zero so its phase tells the truth). +Auth: the shared bearer token plus server-side checks (the TI exists, is in `queued`/`running`, and the +`try_number` matches — stale zombie attempts get 409). Per-task resources/nodeSelector/tolerations come +from the compiled task definition. + +### 3.3 RBAC + +```yaml +# scheduler ServiceAccount: manage the pods it owns + read their logs +rules: + - apiGroups: [""] + resources: ["pods", "pods/log"] + verbs: ["create", "get", "list", "watch", "delete"] +``` + +Task pods need **no** RBAC — they only call the control-plane API. + +--- + +## 4. Deploys and the drain procedure + +In-flight runs **adopt the new binary** on redeploy; the boot reconciler marks tasks that no longer exist +as `removed`, materializes newly-added tasks, and fails runs of deleted DAGs +([architecture.md §7.1](./architecture.md#71-redeploy-semantics-new-binary-in-flight-runs)). XCom **type +drift across a deploy is detected** (fingerprint check) and fails the affected task loudly rather than +mis-deserializing. + +For deploys that change XCom types on purpose, prefer **drain-then-deploy**: + +```bash +kubectl exec maestro-scheduler-0 -- maestro drain # or: scheduler --drain via a flag flip + restart +# scheduler stops creating runs and claiming tasks; running TIs finish; then roll the image +``` + +And run `maestro validate` in CI on every commit — it catches accidental task renames/removals (which +would strand in-flight TIs as `removed`) before they ship. `validate` is the lint half of the pre-deploy +proof; the build is the parse half — between them, nothing about graph shape is left for the scheduler to +discover. (Cross-deploy XCom type drift remains a runtime concern by design; the fingerprint check +catches it loudly — [architecture.md §7.1](./architecture.md#71-redeploy-semantics-new-binary-in-flight-runs).) + +--- + +## 5. Operations + +- **Migrations.** Versioned `migrations/V__.sql`, checksum-recorded in `schema_migration` + (applied files are immutable — changes ship as new files), auto-applied by the scheduler at boot. + Connection setup (`PRAGMA journal_mode=WAL; foreign_keys=ON; busy_timeout=5000`) is issued per + connection, before any transaction. Details: [data-model.md §10](./data-model.md#10-operational-notes). +- **Retention is built into the scheduler loop** (default-on — an unbounded metadata store is the #1 + production regret of comparable systems): purge terminal runs past `--retention-days` (cascade-drops + their TIs, XComs, and history), trim per-DAG attempt history, sweep expired sessions, time-purge the + audit log. Blob cleanup is a **prefix delete** of `xcom//` when a run is purged. After large + purges the scheduler checkpoints (`wal_checkpoint(TRUNCATE)`) and periodically VACUUMs. +- **Pod GC is asymmetric**: succeeded task pods are deleted as soon as their logs ship; **failed pods are + kept** for `--failed-pod-ttl` (default a few hours) so `kubectl describe/logs` forensics still work. +- **Backups are built in**: the scheduler periodically snapshots the DB (`VACUUM INTO`) straight into the + bucket it already uses — `s3://maestro-artifacts/backups/.db`. No extra infrastructure. (The + `turso` engine is beta — keep the backup cadence honest and pin the crate version; `rusqlite` is a + drop-in `StateStore` fallback behind the same trait.) +- **Logs.** The task process uploads its structured `tracing` output to the bucket per attempt (primary); + before GC'ing a pod the scheduler also fetches `pods/log` and uploads it as a fallback artifact — that + covers OOMKills and panics that never got to upload. Keys: `logs////{task.jsonl,pod.log}`. +- **Scaling knobs.** Exactly two: `--slots` (global concurrent tasks) and per-DAG `max_active_runs` + (stops catchup/trigger-spam from stacking overlapping runs). If these ever pinch, pools/priority return + as an additive migration ([data-model.md §9](./data-model.md#9-deliberately-deferred)). +- **Liveness.** The scheduler heartbeats into the `job` table ("last seen" in the UI) and exposes + `/healthz` for its liveness probe. + +--- + +## 6. Quick reference + +```bash +# --- local --- +cargo run -- dev --dag-id sales_etl --trigger --param source_uri=file://fixtures/sales.csv +cargo run -- dev --web --db ./maestro.db +cargo run -- run-task --dag-id sales_etl --task-id transform --input fixtures/raw_rows.json +cargo run -- validate + +# --- prod roles (container args) --- +maestro scheduler --db /data/maestro.db --port 8080 --slots 32 \ + --executor kubernetes --image maestro: +maestro execute-task --dag-id D --task-id T --run-id R --try N # created by the scheduler +maestro migrate --db /data/maestro.db # standalone (scheduler auto-migrates) +maestro drain # stop new work before a type-changing deploy +```