Tubeless ↔ Airflow

Keep Airflow in charge of scheduling and task delivery while Tubeless handles typed work inside a task. Or let a Tubeless pipeline call an existing Airflow DAG through fromRemote and use its result in later steps.

The examples connect both directions: a local Tubeless pipeline submits an Airflow DAG, the DAG runs another Tubeless pipeline, and the caller validates the result returned through XCom. They use Airflow 3’s public /api/v2 API and airflow.sdk. Airflow 2 requires a different adapter and DAG imports.

Choose who owns execution

These are application examples, not a built-in Airflow provider. They use public Tubeless imports and native fetch; the core runtime gains no dependencies. Use either direction independently. The Python DAG can also be triggered directly from Airflow with {"lines": [" Alpha ", "Beta"]} as its run configuration.

Run the examples together

Prepare the Tubeless project

From a checkout of this repository:

bun ci
bun run build

The Airflow execution workers need Bun 1.3.14 or later and this built project. Set TUBELESS_EXAMPLE_ROOT to its absolute path in the worker environment. For a local Airflow installation sharing the checkout:

export TUBELESS_EXAMPLE_ROOT="$(pwd)"

For distributed workers, install the project and its dependencies in every worker image, or adapt the task to your container operator. A path on the machine running the Airflow UI is insufficient. The Python task uses an argument list with subprocess.run, passes JSON on stdin, and keeps the executable path in deployment configuration.

Register the Airflow DAG

Use an existing Airflow 3 installation or follow the official local quick start. Copy tubeless_example.py into its configured DAGs folder. Start the scheduler and workers with the environment above, verify that tubeless_example imports successfully, and unpause it. The DAG has no schedule and runs only when triggered.

Obtain a JWT through your deployment’s configured auth manager and place it in AIRFLOW_TOKEN in the calling process. See Airflow’s public API authentication. Grant access to trigger this DAG, read its runs, and read the task’s XComs. The example reads the token only during execution; importing and planning need no credentials. Use HTTPS outside local development.

Preview and run from Tubeless

From the repository root, inspect the graph without contacting Airflow:

bun run tubeless -- plan examples/airflow/remote.ts

Run with a stable ID for this business request:

bun run tubeless -- run --trace .tubeless/airflow.ndjson \
  examples/airflow/remote.ts -- \
  --api-url http://localhost:8080/api/v2/ \
  --request-id normalize-demo-001 \
  --source examples/rows.txt

The API base includes /api/v2/; a deployment prefix such as https://example.com/airflow/api/v2/ is preserved. The default total wait is 120 seconds; use --timeout-ms 600000 if the queue and task retries take longer. Add --dry-run after -- to skip submission and downstream consumption without making any HTTP requests. The command is also registered as airflow-remote in the project with custom adapters.

Inspect the local recording afterward:

bun run tubeless -- history --trace .tubeless/airflow.ndjson
bun run tubeless -- ui --trace .tubeless/airflow.ndjson

Tubeless → Airflow: submit, wait, validate

The normalize-in-airflow step maps domain inputs into DAG conf, including the caller’s Tubeless runId. Its adapter:

1. Derives a deterministic dag_run_id from the supplied request ID and submits POST /dags/tubeless_example/dagRuns with logical_date: null.
2. Reads the run back and verifies its identity and business inputs. An HTTP 409 attaches to that same run only when the inputs match.
3. Polls queued and running states and reports progress. A failed DAG or unknown state fails the local step.
4. Reads the normalize task’s return_value XCom after DAG success. The outputSchema validates rows, count, Tubeless run ID, and preview flag before summarize can consume them.

The result is a small native JSON object using the default XCom backend. Custom XCom backends may return storage references and need their own result retrieval. For large datasets, return a small artifact reference and validate the fetched artifact at the consumer boundary. See the Airflow 3 REST API and XCom guidance.

Reusing the same request ID and inputs reattaches to the existing run, including a completed or failed run. It never clears tasks or silently reruns failures. If a submission response is lost, rerun the caller with the same ID to discover the original run. Retain Airflow run records for this deduplication window. Use a new ID only when another execution is intentional; external writes still need business-level idempotency.

The adapter does not retry HTTP failures automatically. HTTP errors retain a machine-readable status code without copying response bodies or credentials into diagnostics. Token refresh belongs to the application.

Airflow → Tubeless: one task, one invocation

The Python task supplies Airflow identities and dag_run.conf to worker.ts. The worker validates the envelope and calls AirflowHostedPipeline.runOrThrow. Normalization is pure and safe to repeat; adapt it into meaningful steps for your application.

Airflow owns two task retries, each of which starts a new Tubeless execution. The worker writes its result to a private temporary file only after successful completion. Python returns that JSON as the task’s return_value XCom. A pipeline failure, nonzero subprocess exit, or 90-second subprocess timeout raises in Python, so Airflow can apply its retry policy.

The worker emits Tubeless NDJSON trace events to stdout, which goes to the Airflow task log. The Python task logs its attempt number. Run correlation uses the Airflow DAG, run, task, and map index and stays stable across retries; Tubeless assigns a fresh runId for every attempt. A Tubeless parentRunId is set only when the caller supplied an actual Tubeless execution identity. Standalone Airflow launches have no Tubeless parent.

The Airflow task log may wrap each event with its own formatting. It is not automatically an importable Tubeless NDJSON artifact. To collect remote runs in Studio, supply a trace exporter that stores the raw events durably. The local caller’s trace records one remote step and the returned child run ID; it does not reconstruct the child’s step history. Logs and traces may contain domain data, so apply your application’s redaction policy before exporting them.

Recovery boundaries

Cancelling or timing out the local caller aborts its HTTP requests and polling. The Airflow job may continue. The example deliberately sends no remote cancel, clear, or state-changing recovery request. A successful HTTP abort does not establish that remote work stopped.

The local parent is not restored after process exit. A fresh execution can reattach to the Airflow run through the same request ID, but this reruns the local pipeline rather than resuming a checkpoint. The remote child remains linked to the original parent’s execution identity.

Choose task boundaries that can be retried together. Replaying an Airflow task repeats the entire hosted pipeline, including completed steps. Use idempotent writes or application-owned checkpoints before introducing side effects. For hosted dry runs, propagate conf.dry_run and mark writes with dryRun: "skip" or provide side-effect-free preview handlers.

This on-demand example supplies no data interval. Scheduled and backfill jobs should pass Airflow’s data interval explicitly as domain input and preserve its meaning when moving execution between hosts. See DAG run intervals.

Verification

The package checks compile the TypeScript examples and exercise submission, reattachment, conflicting inputs, progress, invalid XComs, HTTP failures, dry runs, cancellation, and timeouts against a local HTTP fixture. They also run the real Bun worker and check success-only result publication. These checks do not substitute for testing the DAG in your Airflow deployment with its auth manager, executor, and XCom backend.