Airflow daily DAG loaded the same day twice: logical date vs run time

The finance team noticed it first. Their weekly revenue dashboard showed Sunday 6 April with roughly double the usual order value and Saturday 5 April with nothing at all. Our orders_daily_load DAG, which copies the previous day's orders from the operational Postgres replica into the warehouse, had reported every run as green.

The warehouse told the story in one query:

SELECT order_date, COUNT(*) AS rows, COUNT(DISTINCT order_id) AS distinct_orders
FROM analytics.fct_orders
WHERE order_date BETWEEN '2025-04-04' AND '2025-04-07'
GROUP BY order_date ORDER BY order_date;

 order_date | rows  | distinct_orders
------------+-------+-----------------
 2025-04-04 | 18412 |           18412
 2025-04-06 | 39866 |           19933
 2025-04-07 | 18170 |           18170

5 April was missing and 6 April was loaded twice. No task had failed, no retry had fired, and the DAG code had not changed in three weeks.

Reproducing it

The Airflow UI showed something unusual for that weekend. The scheduler pod had been stuck in a crash loop from Saturday evening until Monday 7 April at 08:40 UTC because of an expired database credential. Once it came back, it created two runs back to back:

scheduled__2025-04-05T00:00:00+00:00   start 2025-04-07 08:41:12   success
scheduled__2025-04-06T00:00:00+00:00   start 2025-04-07 08:41:15   success

We reproduced it on a local Airflow 2.10.5 with airflow dags test, passing two different logical dates and then printing what the task actually queried. Both runs executed on the same wall-clock day, so both computed the same "yesterday".

What we ruled out

• A duplicate trigger. We checked the dag_run table for manual runs on that DAG; there were none, only the two scheduled__ runs.
• Retries inserting twice. Each task instance had try_number = 1, so no partial load was repeated.
• Timezone drift in the source. created_at in the source table is timestamptz and our query converts to UTC explicitly, so the boundaries were correct for any single day.

Root cause

The extract task did not use anything Airflow passed to it. It computed the target date like the cron script it was ported from:

target_day = (datetime.now(timezone.utc) - timedelta(days=1)).date()

That only works if every run executes on the day after the data it should load. Airflow's catch-up broke that assumption: two missed intervals were executed on 7 April, both computed 6 April, and the load task appended rows with a plain INSERT. The 5 April interval was "successful" but loaded the wrong day. A colleague had once tried to fix the date logic using {{ ds }} minus one day, which would have shifted everything by another day, so the pull request had been reverted and the wall-clock version stayed.

What is happening under the hood

Airflow schedules data intervals, not moments. A DAG with schedule="@daily" and the default cron data-interval timetable creates one run per interval:

The logical_date (formerly execution_date, and the source of the ds template value) is the start of the interval. The run itself can only start once the interval has ended, so a run labelled 5 April normally starts just after midnight on 6 April. People who think of it as "the date the job runs" end up subtracting a day that Airflow already subtracted.

When the scheduler is down, the intervals do not disappear. With catchup=True (the Airflow 2 default through catchup_by_default), it creates every missed run once it is healthy again, and they all execute right away. Even with catchup=False, the most recent missed interval still runs late. Anything derived from now() is therefore wrong for those runs, and the same applies to airflow tasks clear re-runs and to airflow dags backfill.

Manual triggers add one more twist in Airflow 2: the logical date of a manually triggered run is the trigger time, while the data interval is inferred from the timetable. Code that reads data_interval_start and data_interval_end behaves the same for scheduled, manual, cleared and backfilled runs. Code that reads ds or now() does not.

Environment

Minimal reproduction

from datetime import datetime, timedelta, timezone

import pendulum
from airflow.decorators import dag, task


@dag(schedule="@daily", start_date=pendulum.datetime(2025, 4, 1, tz="UTC"), catchup=True)
def orders_daily_load():
    @task
    def extract():
        # Broken: depends on when the task runs, not on which interval it represents.
        target_day = (datetime.now(timezone.utc) - timedelta(days=1)).date()
        print(f"loading {target_day}")

    extract()


orders_daily_load()

Run two logical dates on the same afternoon and both print the same day:

airflow dags test orders_daily_load 2025-04-05
airflow dags test orders_daily_load 2025-04-06

Who is affected

Any DAG that derives its window from the wall clock, or that treats ds as "today", is fragile. It works on a healthy scheduler and fails on catch-up, backfill, clears and late runs. Airflow 3.0, released on 22 April 2025, changes several defaults (catchup is now off by default and logical dates can be empty for some trigger types), which makes reading the data interval even more important if you plan to upgrade.

The fix

We changed the task to take its window from the run context and made the load idempotent per partition, so a re-run of the same interval replaces data instead of appending.

 @task
-def extract():
-    target_day = (datetime.now(timezone.utc) - timedelta(days=1)).date()
-    rows = fetch_orders(target_day)
-    insert_rows(rows)
+def extract(data_interval_start=None, data_interval_end=None):
+    # Airflow injects these context values by parameter name.
+    rows = fetch_orders(start=data_interval_start, end=data_interval_end)
+    replace_partition(day=data_interval_start.date(), rows=rows)

The source query uses a half-open range, which is exactly what the interval represents:

def build_window(start, end):
    # Pure function: easy to unit test, no clock involved.
    return start.isoformat(), end.isoformat()


def fetch_orders(start, end):
    lo, hi = build_window(start, end)
    sql = """
        SELECT order_id, customer_id, total_cents, created_at
        FROM orders
        WHERE created_at >= %(start)s AND created_at < %(end)s
    """
    return replica.fetch_all(sql, {"start": lo, "end": hi})

And the load deletes the target partition before inserting, inside one transaction:

def replace_partition(day, rows):
    with warehouse.transaction() as tx:
        tx.execute("DELETE FROM analytics.fct_orders WHERE order_date = %(d)s", {"d": day})
        tx.copy_rows("analytics.fct_orders", rows)

For operators that take templated SQL, the equivalent is the interval macros:

WHERE created_at >= '{{ data_interval_start }}'
  AND created_at <  '{{ data_interval_end }}'

Finally we repaired the damaged days by clearing the two runs, which now load their own intervals:

airflow tasks clear orders_daily_load -s 2025-04-05 -e 2025-04-06 -y

Alternatives and trade-offs

1. catchup=False alone. It reduces how many runs fire after an outage but does not fix the bug: the latest missed run still executes late and a manual clear still loads the wrong day. We kept catch-up on because we want every day loaded.
2. ds with no subtraction. For a daily cron interval ds equals the start date of the interval, so it works for scheduled runs. It is less explicit than the interval and differs for manual runs in Airflow 2, so we standardised on the interval values.
3. CronTriggerTimetable. If the job truly is "run at this time and look at whatever is there", a trigger timetable makes the logical date the run time. That is the right model for snapshots, not for loading a bounded day of events.
4. MERGE on order_id instead of delete plus insert. Better when late-arriving updates must be applied, more expensive on large partitions.

How to test the fix

A unit test calls the window helper with explicit interval values while the clock is moved to a "late" moment (the time_machine fixture comes from the time-machine pytest plugin), so it proves the result does not depend on when the run executes:

import pendulum

from dags.orders_daily_load import build_window


def test_window_comes_from_interval_not_clock(time_machine):
    time_machine.move_to("2025-04-07 08:41:00+00:00")  # late run, as after the outage
    start = pendulum.datetime(2025, 4, 5, tz="UTC")
    end = pendulum.datetime(2025, 4, 6, tz="UTC")
    assert build_window(start, end) == ("2025-04-05T00:00:00+00:00", "2025-04-06T00:00:00+00:00")

Then an end-to-end check of two intervals on the same afternoon:

airflow dags test orders_daily_load 2025-04-05
airflow dags test orders_daily_load 2025-04-06

The logs now print two different windows, and running the first command again leaves the row count for 5 April unchanged.

Confirming in production

After clearing the two runs, the same warehouse query returned one row per order for every day:

 order_date | rows  | distinct_orders
------------+-------+-----------------
 2025-04-04 | 18412 |           18412
 2025-04-05 | 17305 |           17305
 2025-04-06 | 19933 |           19933
 2025-04-07 | 18170 |           18170

We also cleared a random run from February to confirm a re-run is a no-op: the row count before and after was identical (16,884), and the task log showed the expected February window.

Regression guard

• A data quality task at the end of the DAG fails the run if COUNT(*) <> COUNT(DISTINCT order_id) for the interval's partition, or if the partition is empty on a weekday.
• A DAG-level test in CI imports every DAG file and greps the task source for datetime.now(, date.today( and pendulum.now(; any hit in a task that loads data needs an explicit reviewer approval.
• We alert when the scheduler heartbeat is older than five minutes, so catch-up storms are expected rather than discovered by finance.

Security implications

None beyond the bug itself. The fix does not change credentials or data access. The incident did start from an expired database credential, so we moved that secret into the secrets backend with rotation alerts. Duplicate financial rows are also an integrity issue: anyone using this table for payouts or reconciliation should treat a doubled day as an incident, not a dashboard glitch.

Prevention checklist

• Derive every query window from data_interval_start and data_interval_end, never from the clock.
• Use half-open ranges (>= start AND < end) so adjacent intervals never overlap.
• Make loads idempotent per interval (overwrite the partition or MERGE on a key).
• Test late runs explicitly by running two logical dates on the same day.
• Read the Airflow 3 upgrade notes before upgrading; several scheduling defaults changed.

References

• Airflow 2.10 documentation, DAG Runs (data interval, catchup, backfill) — https://airflow.apache.org/docs/apache-airflow/2.10.5/core-concepts/dag-run.html
• Airflow 2.10 documentation, Templates reference — https://airflow.apache.org/docs/apache-airflow/2.10.5/templates-ref.html
• Airflow 2.10 documentation, Timetables — https://airflow.apache.org/docs/apache-airflow/2.10.5/authoring-and-scheduling/timetable.html
• Airflow 3.0.0 release notes — https://airflow.apache.org/docs/apache-airflow/3.0.0/release_notes.html