A New DAG Silently Reprocessed a Year of History

1. What caused ~540 runs to fire immediately

catchup was never set, and it defaults to True. With start_date=datetime(2025, 1, 1) and a daily schedule, the instant the DAG is turned on in mid-2026, the scheduler creates a DAG run for every daily interval between January 1, 2025 and now — roughly 540 days — and schedules them all essentially at once, bounded only by available worker slots and max_active_runs (which wasn't set either, so it fell back to the default concurrency limit rather than something deliberately chosen). This is documented Airflow behavior, not a bug: nobody asked for a backfill, but leaving catchup at its default is an implicit request for exactly that.

2. Why there are also duplicate rows

This is a second, independent bug: write_snapshot calls _insert_snapshot_row, which is a plain insert with no delete/overwrite of the target date's existing row first. Combined with retries: 2, any write_snapshot task instance that fails partway through (e.g., succeeds at inserting but then hits a transient DB timeout before returning success, or is retried after a scheduler hiccup) will insert a second row for the same date on its retry. This would be a latent bug even without the catchup incident — the incident just multiplied the number of opportunities for it to fire, because 540 runs firing under rate-limiting pressure (which itself increases the rate of task failures and therefore retries) is exactly the condition that exposes a non-idempotent write. The two problems are independent: fixing catchup alone would still leave the insert able to duplicate rows on any future retry; fixing the insert alone would not have prevented the rate-limiter incident.

3. The fix

@dag(
    dag_id="daily_revenue_snapshot",
    schedule="0 7 * * *",
    start_date=datetime(2025, 1, 1),
    catchup=False,                     # no automatic historical runs
    max_active_runs=1,                 # bound concurrency for any future backfill
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def daily_revenue_snapshot():

    @task
    def pull_revenue(data_interval_start=None) -> str:
        return _call_billing_api(data_interval_start)

    @task
    def write_snapshot(payload: str, data_interval_start=None) -> None:
        date_str = data_interval_start.strftime("%Y-%m-%d")
        _delete_snapshot_row(date_str)      # make the write idempotent
        _insert_snapshot_row(payload, date_str)

    write_snapshot(pull_revenue())

daily_revenue_snapshot()

With catchup=False, turning the DAG on only schedules runs going forward from now, and write_snapshot deleting the target date's existing row before inserting means any retry, manual re-run, or future backfill of a given date produces the same final state instead of a duplicate. To still populate the desired year of history, run an explicit, bounded backfill rather than relying on the deploy-time default: airflow dags backfill -s 2025-01-01 -e 2026-06-30 daily_revenue_snapshot, ideally with --rate-limit-aware pacing or a lowered max_active_runs for the backfill window specifically, so the billing API's 500-requests/day limit isn't hit again — an explicit backfill is auditable and controllable in a way the accidental catchup-triggered flood never was, and because the write is now idempotent, even a backfill that gets interrupted and re-run is safe.