Tests, Retries, and Shipping It

The DAG runs. It’s scheduled, it renders a task per model, it only rebuilds what changed. That’s the pipeline working — and “working” is where most tutorials stop and production begins. Production is the set of things you only notice when they’re missing: knowing which assertion failed at 2am, hearing about the test that warned instead of failed, retrying the one model that flaked without double-loading it, seeing the run’s health without SSH-ing anywhere, and stopping a broken model at the pull request instead of in the warehouse. None of it is new dbt or new Airflow. It’s the payoff for having made both graphs one graph.

A failing test is a red cell, not a red run

dbt tests aren’t a separate step you remember to run — they’re nodes in the same graph. Cosmos renders them as their own Airflow tasks, so the not_null on orders.customer_id and the relationships test tying int_order_payments back to stg_customers each become a discrete cell in the grid, wired downstream of the model they guard.

Go back to the naive bridge from post two. There, a failed test looked exactly like a failed model looked exactly like a failed anything: one red bar, and dbt’s output buried in a log you scroll. Here, a broken unique test on orders is a single red square with the assertion’s name on it. You know which model produced bad data, which specific contract it violated, and — because the failed test is downstream of exactly one model — what to fix. You retry that model and its test. Not the world.

That’s the whole argument of this series collapsed into one screen: Airflow can only make a failure legible if it can see the thing that failed, and now it can. But “the test failed and turned the cell red” is only the loud half of the story. The half that bites you in production is the test that found something and didn’t turn anything red at all.

The failure that doesn’t fail: warn-severity tests

dbt tests have a severity. The default is error — the test fails, the node fails, and Cosmos propagates that up as a failed Airflow task, red cell, downstream blocked. But plenty of assertions aren’t “stop the pipeline” conditions; they’re “someone should look at this.” A row count that drifted 3%. A handful of orders with a null shipped_at that’s usually populated. You want those surfaced, not slammed. That’s severity: warn:

models:
  - name: orders
    columns:
      - name: total_amount
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100000
              severity: warn
              warn_if: ">10"
              error_if: ">100"

Here’s the trap. A warn-severity test that finds violating rows does not fail the dbt node, which means Cosmos renders it green. The test ran, it found bad data, it told dbt — and the grid shows you a clean run. If your only observability surface is “is the cell red,” warnings are invisible by construction. You built a signal and then hid it behind the one thing you look at.

Cosmos gives you the hook to un-hide it: on_warning_callback. It fires when a dbt test node emits warnings, and it’s where you turn a silent warn into a Slack message that isn’t a page:

from cosmos import DbtDag, RenderConfig

def warn_handler(context):
    # context carries the dbt test node names that warned
    warned = context.get("test_names", [])
    notify_slack(
        f":large_yellow_circle: {context['dag'].dag_id} warned on "
        f"{len(warned)} test(s): {', '.join(warned)} — non-blocking, but look."
    )

DbtDag(
    project_config=project_config,
    on_warning_callback=warn_handler,          # a DbtDag kwarg — NOT a RenderConfig field
    execution_config=execution_config,
    profile_config=profile_config,
    schedule="@daily",
    dag_id="bookshop",
)

The callback is attached at the Cosmos level, not per-task, because a warning isn’t a task failure — the task succeeded. on_warning_callback is the only thing standing between a warn-severity test and total silence, so if you use warn severity at all (you should), wire this or the severity is a lie you’re telling yourself.

Warnings tell you that something’s off. To triage what, you need the offending rows, and by default dbt throws them away — a test returns a count and discards the failing set. store_failures keeps them:

      - name: total_amount
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              severity: warn
              config:
                store_failures: true
                store_failures_as: table   # or 'view'

Now every run of that test writes the violating rows to a table in a dbt_test__audit schema, named after the test. When the Slack warn lands, you don’t re-run anything to investigate — you select * from dbt_test__audit.accepted_range_orders_total_amount and read the actual bad rows from the last run. store_failures_as: view makes it a view over live data instead of a snapshot, which is what you want for a test you’re actively debugging; table is the durable audit trail. Either way, the triage data is already in the warehouse before you go looking. --store-failures is the CLI flag for the same thing when you want it globally on a run.

Quality gates that fail before bad data exists

not_null and unique are assertions you run after building a model — they catch bad data that already landed. The stronger move is a gate that refuses to build the model at all when its shape is wrong. dbt has two, and Cosmos renders both without you doing anything special.

Model contracts enforce a schema at build time. Declare the columns and types a model must produce, set enforced: true, and dbt builds the model inside a scaffold that fails the run if the output doesn’t match — a renamed column, a type that drifted from int to numeric, a nullable field where you promised non-null:

models:
  - name: orders
    config:
      contract:
        enforced: true
    columns:
      - name: order_id
        data_type: bigint
        constraints:
          - type: not_null
          - type: primary_key
      - name: customer_id
        data_type: bigint
        constraints:
          - type: not_null
      - name: total_amount
        data_type: numeric(12,2)

The difference from a not_null test is when and how loud. A contract violation fails the model’s own build task — the red cell is orders, not orders’s downstream test — and it fails because the column contract broke, before a single downstream model reads the malformed output. It’s a compile-and-build-time gate, so a PR that changes total_amount from numeric to float fails in CI on the model itself, with a message naming the column and the type mismatch. Contracts turn “we test that the data is clean” into “the warehouse refuses to accept a differently-shaped table,” which is the gate you actually want on the models other teams build on.

Unit tests are the other half — they test your SQL logic against fixed inputs, the way application code has unit tests, and they’re a first-class dbt node so Cosmos renders each as its own task:

unit_tests:
  - name: test_order_total_excludes_refunds
    model: orders
    given:
      - input: ref('int_order_payments')
        rows:
          - {order_id: 1, amount: 100, status: 'settled'}
          - {order_id: 1, amount: -30, status: 'refunded'}
    expect:
      rows:
        - {order_id: 1, total_amount: 70}

This runs at build time against mocked inputs — no warehouse data required — and asserts that your aggregation logic actually excludes refunds the way you think it does. It catches the class of bug a data test never can: the SQL is wrong even when the data is clean. In the grid it’s a task named for the unit test, upstream of the model’s real build, so a logic regression fails the PR before the model runs against production data at all.

The third gate lives in the sources, not the models: source freshness. dbt source freshness checks the max timestamp in each raw table against a threshold and fails if the upstream feed went stale. Run it as the first thing in CI (and as an early task in the DAG) and you catch “the ingestion job died and the bookshop’s raw orders are 30 hours old” before you build a full pipeline on top of yesterday’s data and wonder why the numbers are flat.

dbt source freshness   # warn_after / error_after come from the source YAML

Contracts stop bad shapes, unit tests stop bad logic, freshness stops building on stale inputs, and data tests catch bad values after the fact. Four gates, four failure modes, all rendered as tasks in the same graph.

run_results.json is your observability, if you read it

The grid is a fine dashboard for is it green. It’s a poor one for how long did each model take, and is that getting worse. dbt already answers that — it writes run_results.json to target/ on every invocation, one record per node with its status, execution time, rows affected, and the compiled adapter response. The whole run’s performance history is sitting in a file after every dbt call. The work isn’t generating it; it’s not throwing it away.

The concrete pattern is a post-run task that parses run_results.json and loads the per-node timings into a metadata table you can trend:

Before the code, the trap — because the obvious version of this task does not work, and it fails in a way that looks like your fault.

Cosmos does not leave target/ on disk. Every Cosmos task copies the project into a tempfile.TemporaryDirectory(), runs dbt there, and deletes it. So a downstream task that does open(f"{project_dir}/target/run_results.json") gets a FileNotFoundError — not because the artifact wasn’t written, but because the directory it was written into no longer exists. This is true in every execution mode, LOCAL included; it is not a container thing.

Cosmos’s answer is to ship the artifacts out for you. Point it at object storage and it uploads target/ after each invocation:

# .env  —  the [cosmos] section
AIRFLOW__COSMOS__REMOTE_TARGET_PATH=s3://bookshop-dbt-artifacts/target/
AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID=aws_default

Now the artifacts outlive the task, in a location every downstream step can reach, and your observability task reads from there rather than from a directory that was deleted before it ran:

import json
from datetime import datetime
from airflow.sdk import task
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

@task
def load_run_results(key: str):
    body = S3Hook(aws_conn_id="aws_default").read_key(
        key=key, bucket_name="bookshop-dbt-artifacts",
    )
    results = json.loads(body)

    invocation = results["metadata"]["invocation_id"]
    rows = [
        {
            "invocation_id": invocation,
            "run_at": datetime.utcnow(),
            "node": r["unique_id"],
            "status": r["status"],
            "execution_time": r["execution_time"],
            "rows_affected": r.get("adapter_response", {}).get("rows_affected"),
        }
        for r in results["results"]
    ]
    warehouse_hook().insert_rows("meta.dbt_run_results", rows)

Wire that as a task downstream of the dbt task group (it runs whether the run passed or failed — use a trigger rule so a partial failure still records what did run) and you accumulate a meta.dbt_run_results table that answers questions the grid can’t: which model’s runtime doubled this month, which test flakes across invocations, what the p95 build time is per node. That’s the raw material for a real dashboard, and it costs one parse task because dbt did the measurement for you.

When you want that story turned into a product instead of a table, the dbt-native answer is Elementary — a package you add to packages.yml that installs models which sit on top of your run_results and manifest artifacts and materialize a run history, plus a suite of anomaly tests (volume, freshness, and column-value distributions) that fail when today’s numbers deviate from the model’s own recent history rather than a hardcoded threshold:

# packages.yml
packages:
  - package: elementary-data/elementary
    version: 0.16.0

models:
  - name: orders
    tests:
      - elementary.volume_anomalies:
          timestamp_column: created_at
      - elementary.freshness_anomalies:
          timestamp_column: created_at

Those anomaly tests are ordinary dbt tests — Cosmos renders them as task cells like any other — but they encode “this table normally gets ~2,000 orders a day and today it got 4” as a test failure, which is exactly the drift a static not_null can’t see. Elementary also ships a report (edr report generates a self-contained HTML dashboard from the run artifacts) and an alerting integration, so the observability you were hand-rolling off run_results.json becomes a maintained package. Roll your own metadata table when you want a couple of trends in your own stack; reach for Elementary when you want anomaly detection and a report without building either.

Retries that cost one model — and don’t double-load it

Some failures aren’t bugs — a warehouse connection drops, a lock times out, a transient nothing. In the single-task world every transient blip reran the entire project. Per-model tasks make retries proportional:

default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=1),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=10),
}

default_args propagates to every rendered task, so a flaky int_order_payments retries itself while stg_orders, which already succeeded, sits untouched. The retry unit is now a model, because the task unit is a model. retry_exponential_backoff=True spaces the attempts out — one minute, then two, then four — instead of hammering a warehouse that’s already struggling, and max_retry_delay caps how far that backoff can stretch. For a transient lock, backing off is strictly better than retrying instantly into the same contention.

There’s a correctness hazard hiding in that convenience, and it’s the reason to think before setting retries globally. A retry re-runs the model’s dbt run. For a view or a table materialization that’s harmless — the model is fully rebuilt from scratch, so a second attempt produces the identical result. For an incremental model it is not harmless. An incremental run appends or merges new rows since the last watermark. If the model runs, commits its insert, and then the task fails — a network blip after the warehouse already applied the write — the retry runs the incremental logic again and can load the same interval twice. retries on a non-idempotent incremental is a double-load waiting for a bad night.

The precondition for safe retries on incremental models is idempotency: the incremental logic has to produce the same table whether it runs once or three times for a given interval. That’s a delete+insert or merge on a deterministic key scoped to the run’s data interval, not a blind insert. Get the incremental model idempotent and retries: 2 is free insurance; leave it non-idempotent and every retry is a chance to corrupt the table. This is a whole topic — the data-interval-aware incremental patterns, the --full-refresh escape hatch, and how backfills interact with all of it — and it’s the subject of a later chapter on incrementals under backfills. The one-line version here: don’t set blanket retries on incremental models until they’re idempotent. For those specific models, either make them idempotent first or drop retries to 0 and let a human decide.

Alerting in 3.x: SLAs are gone, callbacks and Deadline Alerts replace them

If you learned Airflow on 2.x, your instinct for “tell me when this runs too long” is the sla parameter. It’s gone — SLAs were removed in Airflow 3. The replacement is two mechanisms that are more honest about what they do.

Callbacks handle the failure and success cases. on_failure_callback fires when a task fails, which is the hook for turning a red cell into a page:

def alert_on_failure(context):
    ti = context["task_instance"]
    notify_slack(f":red_circle: {ti.dag_id}.{ti.task_id} failed — {ti.log_url}")

default_args = {"on_failure_callback": alert_on_failure, ...}

Attach it via default_args and every model task inherits it, so the alert names the model that broke and links straight to its log. Note the distinction between task-level and DAG-level callbacks: a task callback fires per failing task, so a run where five models fail sends five pages. A DAG-level on_failure_callback fires once for the run. For a Cosmos DAG that renders dozens of tasks, task-level failure alerts are how you get an alert storm — one bad upstream model fails, and its ten downstream tasks each fire their own page. The usual shape is: DAG-level callback for “the run failed, here’s the run,” and a task-level callback only on the specific tasks whose individual failure you genuinely want paged separately.

For the “took too long” case SLAs used to cover, 3.x introduces Deadline Alerts — a DeadlineAlert you attach to a DAG that defines a deadline relative to a reference time and a callback to fire when it’s breached:

from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference
from datetime import timedelta

DbtDag(
    # ...
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
        interval=timedelta(hours=2),
        callback=SyncCallback(alert_slow_run),   # a bare function is rejected
    ),
)

Unlike the old SLA, which quietly recorded misses in a table most people never looked at, a Deadline Alert fires a real callback when the run blows past its window — “the nightly bookshop build was supposed to finish by 4am and it’s 6,” delivered somewhere you’ll see it. It’s the SLA you actually wanted: a deadline with an action, not a deadline with a log row.

Both of these route through the Notifier abstraction (BaseNotifier), which is 3.x’s structured way to write a reusable notifier — a class with a notify(context) method — instead of hand-rolling a callback function per DAG. Providers ship them (SlackNotifier, SmtpNotifier); you write your own by subclassing BaseNotifier and reuse it across every DAG and deadline. Whichever you use, the last mile is alert-storm dedupe: a single upstream failure cascades, and without a suppression key you page ten times for one root cause. The practical fixes are DAG-level callbacks for run failure, a dedupe key in the notifier (hash of dag+root-cause within a window) so repeats within N minutes collapse into one message, and reserving task-level pages for the handful of models where isolated failure genuinely means something different.

CI depth: where the production manifest comes from, and the scratch-schema problem

Everything so far protects the nightly run. CI protects the thing that breaks nightly runs — the pull request. The cheapest gate is dbt parse, which compiles the project and fails on a broken ref(), a bad column reference, or malformed YAML before any SQL touches the warehouse. Layer the slim CI from the last post on top and a PR builds only the models it changed, deferring the rest to production:

name: dbt-ci
on: pull_request
jobs:
  build-changed:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - uses: astral-sh/setup-uv@v4
      - run: uv pip install --system dbt-core dbt-postgres
      - run: dbt deps
      # Land the production manifest CI defers against:
      - run: aws s3 cp s3://bookshop-dbt-artifacts/prod/manifest.json ./prod-artifacts/manifest.json
      - run: dbt parse
      - run: dbt source freshness
      - run: |
          dbt build --select "state:modified+" --defer --state ./prod-artifacts \
            --target ci
        env:
          DBT_SCHEMA: "ci_pr_${{ github.event.number }}"
          DBT_HOST: ${{ secrets.DBT_HOST }}
          DBT_USER: ${{ secrets.CI_DBT_USER }}
          DBT_PASSWORD: ${{ secrets.CI_DBT_PASSWORD }}

The line that matters most is the one that lands ./prod-artifacts/manifest.json. Slim CI is only as good as that baseline (post five’s whole argument), and in CI it doesn’t appear by magic — it’s downloaded from the published location your production deploy writes to. The nightly run publishes its target/manifest.json to object storage on success; CI pulls that exact file down before dbt build so state:modified+ compares the PR against last-known-good prod. No manifest, and CI rebuilds the world; stale manifest, and it under-builds. The aws s3 cp line is the state discipline made executable.

Then the ephemeral schema story. A PR build must not write into production tables, so it runs against a dedicated CI target with its own credentials (CI_DBT_USER, not the prod user) and a per-PR scratch schema — ci_pr_1234, keyed off the PR number so concurrent PRs don’t collide. --defer means the PR only builds its changed models into that scratch schema and reads everything unchanged from prod, so the scratch schema stays small. The catch is cleanup: every PR leaves a ci_pr_* schema behind, and without a sweep your warehouse fills with dead scratch. The fix is a teardown step on PR close that drops the schema:

  cleanup:
    if: always()
    runs-on: ubuntu-latest
    steps:
      - run: |
          dbt run-operation drop_schema \
            --args "{schema: ci_pr_${{ github.event.number }}}"

The dbt clone alternative sidesteps building into scratch entirely. Instead of rebuilding changed models, dbt clone --state ./prod-artifacts creates zero-copy clones of the production objects into the CI schema (on warehouses that support cheap cloning — Snowflake, BigQuery, Databricks), so CI starts from a full copy of prod for near-free and then builds only the modified models on top. On a warehouse with cloning it’s often cheaper and more faithful than defer, because the PR sees real production data for every upstream, not a manifest’s description of it. On Postgres — the bookshop’s engine here — there’s no cheap clone, so defer is the right tool; the clone path is what you reach for once this setup points at a cloud warehouse.

On merge, the deploy is its own short job: ship the Astro project so the scheduler picks up the new DAG and dbt code, and — critically — that production run is what publishes the next baseline manifest.

  deploy:
    needs: build-changed
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: astro deploy --deployment-id ${{ secrets.ASTRO_DEPLOYMENT_ID }}
        env:
          ASTRO_API_TOKEN: ${{ secrets.ASTRO_API_TOKEN }}

Tests on the PR, deploy on the merge. The manifest production publishes becomes the state baseline the next PR compares against — the post-five artifact discipline and the CI gate are the same system viewed from two ends.

Final thoughts

Almost nothing in this post was a new capability. Test-as-task, per-model retries, the grid-as-dashboard, slim CI — every one of them fell out of the single decision made in post three, to let Airflow see the graph dbt already draws. What this post added is the discipline that keeps that legibility honest: the warn-severity test that’s silently green until on_warning_callback speaks for it, the run_results.json you’d otherwise let dbt delete, the contract that fails a bad shape before it exists, the retry you must not set on a non-idempotent incremental, and the deadline alert that replaced the SLA Airflow 3 took away. That’s the actual thesis of the series: you don’t integrate dbt and Airflow by making one run inside the other, you integrate them by refusing to hide either one’s structure — and then you do the unglamorous work of not hiding it back by accident.

The bookshop ran on Postgres the whole way through, deliberately — warehouse-agnostic, laptop-reproducible, no cloud account required to follow along. The honest next move is to point this exact setup at a warehouse built for it: swap the profile, keep the DAG, and suddenly dbt clone and zero-copy CI schemas are on the table. That’s the Snowflake capstone this track is building toward — the same Cosmos graph, the same slim CI, the same test cells, aimed at a real cloud warehouse where the cost of rebuilding the world is the reason all of this mattered in the first place. The orchestration doesn’t change. Only the thing underneath it gets bigger.

Next: How Cosmos Loads and Runs dbt