Airflow Missed Run Backfill Guard with Kestra

id: airflow-paused-dag-missed-run-backfill-guard
namespace: company.team
description: |
  Find daily logical dates in an Airflow DAG window that have no DAG run,
  then optionally backfill those dates one at a time after approval. The plan
  is bounded by max_backfill_count and every requested date is verified again
  from Airflow after the backfill.

triggers:
  - id: audit_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled schedule for a safe audit. Trigger inputs always force
      read-only mode.
    cron: "0 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

inputs:
  - id: airflow_base_url
    type: STRING
    displayName: Airflow Base URL
    defaults: "http://localhost:8080"
    description: Airflow webserver base URL reachable by the Kestra worker.
  - id: dag_id
    type: STRING
    displayName: DAG ID
    defaults: "target_dag"
    description: Daily DAG whose logical dates are audited.
  - id: window_start
    type: STRING
    displayName: Window Start
    defaults: "2026-10-01T00:00:00Z"
    description: Inclusive UTC logical date at the start of the audit window.
  - id: window_end
    type: STRING
    displayName: Window End
    defaults: "2026-10-05T00:00:00Z"
    description: Inclusive UTC logical date at the end of the audit window.
  - id: max_backfill_count
    type: INT
    displayName: Maximum Backfill Count
    defaults: 3
    description: Hard ceiling on missing daily dates that may be requested in one run.
  - id: auto_remediate
    type: BOOL
    displayName: Auto Remediate
    defaults: false
    description: Enables the approval and backfill branch when true.
  - id: dry_run
    type: BOOL
    displayName: Dry Run
    defaults: true
    description: Keeps the flow read-only when true. Must be false with
      auto_remediate true to write.

tasks:
  - id: read_runs_before
    type: io.kestra.plugin.core.http.Request
    description: Read all existing DAG runs from Airflow before planning. Missing
      intervals have no run object to alert on.
    uri: "{{ inputs.airflow_base_url }}/api/v1/dags/{{ inputs.dag_id
      }}/dagRuns?limit=1000"
    method: GET
    headers:
      Authorization: "Basic {{ (secret('POSTGRES_USERNAME') ~ ':' ~
        secret('POSTGRES_PASSWORD')) | base64encode }}"
      Accept: application/json

  - id: plan_missing
    type: io.kestra.plugin.core.output.OutputValues
    description: Derive the expected daily UTC dates, subtract Airflow's logical
      dates and cap the requested list before any approval.
    values:
      plan: >-
        {{ {'runs': outputs.read_runs_before.body, 'start': inputs.window_start,
        'end': inputs.window_end, 'max': inputs.max_backfill_count} | toJson |
        jq('. as $r | ($r.runs | fromjson) as $j |
        ([$j.dag_runs[]?.logical_date] // []) as $present | [range(($r.start |
        fromdateiso8601); (($r.end | fromdateiso8601) + 86400); 86400) |
        todateiso8601] as $expected | [$expected[] | . as $d |
        select(([$present[] | .[0:10] | select(. == ($d[0:10]))] | length) ==
        0)] as $missing | {expected: $expected, missing: $missing, requested:
        ($missing[0:$r.max]), missing_count: ($missing | length), over_limit:
        (($missing | length) > $r.max)} | tojson') }}

  - id: log_plan
    type: io.kestra.plugin.core.log.Log
    description: Record the measured gap and the bounded request before any write gate.
    message: >-
      Airflow gap audit for {{ inputs.dag_id }}: {{
      outputs.plan_missing.values.plan | jq('first | fromjson | .missing_count')
      | first }} missing daily dates, requested {{
      outputs.plan_missing.values.plan | jq('first | fromjson | .requested |
      length') | first }} of max {{ inputs.max_backfill_count }}. Dates={{
      outputs.plan_missing.values.plan | jq('first | fromjson | .requested |
      join(",")') | first }}

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    description: Only a non-dry run with auto_remediate enabled can reach approval
      or TriggerDagRun.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      (outputs.plan_missing.values.plan | jq('first | fromjson | .requested |
      length') | first) > 0 and not (outputs.plan_missing.values.plan |
      jq('first | fromjson | .over_limit') | first) }}"
    then:
      - id: approve_backfill
        type: io.kestra.plugin.core.flow.Pause
        description: Approve the exact chronological logical-date list. Declining leaves
          Airflow unchanged.
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: Set true only after reviewing the requested logical dates.

      - id: approval_gate
        type: io.kestra.plugin.core.flow.If
        description: Continue only when the Pause explicitly returned approved true.
        condition: "{{ outputs.approve_backfill.onResume.approved ?? false }}"
        then:
          - id: read_runs_after_pause
            type: io.kestra.plugin.core.http.Request
            description: Re-read Airflow immediately after approval so a newly-created run
              makes the old plan stale instead of being duplicated.
            uri: "{{ inputs.airflow_base_url }}/api/v1/dags/{{ inputs.dag_id
              }}/dagRuns?limit=1000"
            method: GET
            headers:
              Authorization: "Basic {{ (secret('POSTGRES_USERNAME') ~ ':' ~
                secret('POSTGRES_PASSWORD')) | base64encode }}"
              Accept: application/json

          - id: plan_after_pause
            type: io.kestra.plugin.core.output.OutputValues
            description: Recompute the bounded missing-date plan from the post-approval
              Airflow read.
            values:
              plan: >-
                {{ {'runs': outputs.read_runs_after_pause.body, 'start':
                inputs.window_start, 'end': inputs.window_end, 'max':
                inputs.max_backfill_count} | toJson | jq('. as $r | ($r.runs |
                fromjson) as $j | ([$j.dag_runs[]?.logical_date] // []) as
                $present | [range(($r.start | fromdateiso8601); (($r.end |
                fromdateiso8601) + 86400); 86400) | todateiso8601] as $expected
                | [$expected[] | . as $d | select(([$present[] | .[0:10] |
                select(. == ($d[0:10]))] | length) == 0)] as $missing |
                {requested: ($missing[0:$r.max]), missing_count: ($missing |
                length), over_limit: (($missing | length) > $r.max)} | tojson')
                }}

          - id: stale_plan_check
            type: io.kestra.plugin.core.output.OutputValues
            description: Compare the approved request with the fresh Airflow plan and refuse
              a changed target set.
            values:
              result: >-
                {{ {'before': outputs.plan_missing.values.plan, 'after':
                outputs.plan_after_pause.values.plan} | toJson | jq('. as $r |
                (($r.before | fromjson | first | fromjson).requested) as $b |
                (($r.after | fromjson | first | fromjson).requested) as $a |
                {same: ($b == $a), before: $b, after: $a} | tojson') }}

          - id: reject_stale_plan
            type: io.kestra.plugin.core.flow.If
            description: A changed run list fails closed before any TriggerDagRun write.
            condition: "{{ not (outputs.stale_plan_check.values.result | jq('first |
              fromjson | .same') | first) }}"
            then:
              - id: fail_stale_plan
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Airflow run list changed during approval. Approved={{
                  outputs.stale_plan_check.values.result | jq('first | fromjson
                  | .before | join(\",\")') | first }} current={{
                  outputs.stale_plan_check.values.result | jq('first | fromjson
                  | .after | join(\",\")') | first }}. No backfill was started."

          - id: backfill_missing_dates
            type: io.kestra.plugin.core.flow.Loop
            description: Trigger each missing logical date in chronological order with one
              active Airflow run at a time.
            values: "{{ outputs.plan_after_pause.values.plan | jq('first | fromjson |
              .requested') | first }}"
            concurrencyLimit: 1
            tasks:
              - id: trigger_missing_run
                type: io.kestra.plugin.airflow.dags.TriggerDagRun
                description: Create and wait for exactly one missing logical date. Airflow
                  receives the date as logical_date and returns its terminal
                  state.
                baseUrl: "{{ inputs.airflow_base_url }}"
                dagId: "{{ inputs.dag_id }}"
                wait: true
                pollFrequency: PT2S
                maxDuration: PT10M
                body:
                  logical_date: "{{ item.value }}"
                  conf:
                    kestra_execution_id: "{{ execution.id }}"
                    backfill_guard: true
                options:
                  auth:
                    type: BASIC
                    username: "{{ secret('POSTGRES_USERNAME') }}"
                    password: "{{ secret('POSTGRES_PASSWORD') }}"

          - id: read_runs_after_backfill
            type: io.kestra.plugin.core.http.Request
            description: Read Airflow after all sequential triggers. The run list, not
              TriggerDagRun acceptance, proves recovery.
            uri: "{{ inputs.airflow_base_url }}/api/v1/dags/{{ inputs.dag_id
              }}/dagRuns?limit=1000"
            method: GET
            headers:
              Authorization: "Basic {{ (secret('POSTGRES_USERNAME') ~ ':' ~
                secret('POSTGRES_PASSWORD')) | base64encode }}"
              Accept: application/json

          - id: verify_backfill
            type: io.kestra.plugin.core.output.OutputValues
            description: Fail the execution when any requested date is still absent or has a
              failed Airflow run.
            values:
              result: >-
                {{ {'runs': outputs.read_runs_after_backfill.body, 'requested':
                outputs.plan_after_pause.values.plan} | toJson | jq('. as $r |
                ($r.runs | fromjson) as $j | (($r.requested | fromjson | first |
                fromjson).requested) as $req | [$req[] | . as $d |
                ([$j.dag_runs[]? | select((.logical_date[0:10]) == ($d[0:10]))])
                as $hits | if ($hits | length) == 0 then {date: $d, state:
                "missing"} elif (($hits[0].state | ascii_downcase) == "failed")
                then {date: $d, state: "failed"} else empty end] as $bad | {bad:
                $bad, complete: (($bad | length) == 0)} | tojson') }}

          - id: fail_incomplete_backfill
            type: io.kestra.plugin.core.flow.If
            description: Make missing or failed requested dates an explicit failed
              execution.
            condition: "{{ not (outputs.verify_backfill.values.result | jq('first | fromjson
              | .complete') | first) }}"
            then:
              - id: fail_verification
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Airflow backfill verification failed: {{
                  outputs.verify_backfill.values.result | jq('first | fromjson |
                  .bad | map((.date + \"=\" + .state)) | join(\",\")') | first
                  }}"

          - id: log_backfill_complete
            type: io.kestra.plugin.core.log.Log
            message: "Airflow backfill verified for {{ inputs.dag_id }}. requested={{
              outputs.plan_after_pause.values.plan | jq('first | fromjson |
              .requested | length') | first }} bad={{
              outputs.verify_backfill.values.result | jq('first | fromjson |
              .bad | length') | first }}"
        else:
          - id: log_declined
            type: io.kestra.plugin.core.log.Log
            message: "Backfill approval declined. Airflow was not changed."
    else:
      - id: log_read_only
        type: io.kestra.plugin.core.log.Log
        message: "Read-only result for {{ inputs.dag_id }}. auto_remediate={{
          inputs.auto_remediate }} dry_run={{ inputs.dry_run }} missing={{
          outputs.plan_missing.values.plan | jq('first | fromjson |
          .missing_count') | first }} requested={{
          outputs.plan_missing.values.plan | jq('first | fromjson | .requested |
          length') | first }} over_limit={{ outputs.plan_missing.values.plan |
          jq('first | fromjson | .over_limit') | first }}. No Airflow runs were
          triggered."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when Airflow cannot be read, a backfill run fails or
      verification fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Airflow missing-run backfill guard FAILED for {{ inputs.dag_id }} in execution {{ execution.id }}. Check Airflow reachability, run state and the verification task before trusting the partition window."
      }

Safety

A paused DAG can skip scheduled intervals without producing a failed run. The downstream table is short while Airflow reports no error. This flow only triggers dates that have no run in the audited window, caps the request and verifies the resulting run list.

Guard	What it stops
Inclusive window and daily logical-date subtraction	Backfilling dates outside the audited DAG window
`max_backfill_count`	An unbounded catch-up after a long maintenance pause
`auto_remediate` plus `dry_run`	Accidental writes from the scheduled safe path
`Pause` with explicit approval	Starting runs without a human review of the exact dates
Fresh run-list read after approval	Applying a stale plan after another actor created a run
`Loop` `concurrencyLimit: 1`	Flooding the Airflow pool or changing chronological order
Post-run run-list and state verification	Treating an accepted trigger as proof that the partition recovered

How it works

read_runs_before (io.kestra.plugin.core.http.Request) reads Airflow's DAG run list and plan_missing (io.kestra.plugin.core.output.OutputValues) subtracts present logical dates from the inclusive daily UTC window.
log_plan records the exact missing count and the bounded request. The scheduled trigger is disabled and forces auto_remediate: false and dry_run: true.
Only auto_remediate: true, dry_run: false and a non-empty plan reach approve_backfill (io.kestra.plugin.core.flow.Pause). A declined pause leaves Airflow unchanged.
read_runs_after_pause and plan_after_pause repeat the detector. stale_plan_check fails before any write when the approved list changed.
backfill_missing_dates (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) calls io.kestra.plugin.airflow.dags.TriggerDagRun with each logical date and waits for its terminal state before moving to the next date.
read_runs_after_backfill reads Airflow again and verify_backfill fails when a requested date is missing or failed. The errors block sends a Slack alert for execution failures.

What you get

A measurable list of absent daily DAG runs instead of an alert that only sees failed runs.
A bounded chronological backfill with a human approval point and stale-plan protection.
Airflow-native verification of every requested logical date after the write.
A disabled safe-mode schedule and an auditable Kestra execution for every audit.

Who it's for

Data platform teams whose Airflow DAGs feed partitioned tables.
On-call engineers investigating a maintenance pause or scheduler outage.
Operators who need a controlled catch-up path that cannot flood an Airflow pool.

Why orchestrate this with Kestra

Airflow records no failed run for an interval that never created a run, so its scheduler and task alerts cannot identify the missing partition by themselves. Kestra adds the bounded date arithmetic, approval pause, sequential trigger loop, stale-plan check and independent read-back in one auditable execution.

Tested end to end

Tested against Kestra 2.0.5 and Apache Airflow 2.9.3 with a pinned Airflow test fixture. The fixture uses a daily DAG with intentionally absent logical dates. The evidence packet records healthy, dry-run, declined approval, remediation, repeat, ineffective and stale-plan cases with Airflow CLI counts before and after.

Case	Execution	Result
Healthy control	`3rzXh61AWXDTNZB2xMG2yQ`	SUCCESS, 0 missing dates, no triggers
Degraded dry run	`7etsV9rfuyqbjdqGCQ4W3u`	SUCCESS, 3 missing dates listed, no triggers
Declined approval	`rVggGIMXdCmdy0K5upL3j`	SUCCESS, approval false, no triggers
Remediation	`543QsR1QtZllZ44ubXitSX`	SUCCESS, 3 requested dates, verification bad=0
Repeat	`6tHSiQN9n5Icg8Ej8xKUDC`	SUCCESS, 0 missing dates, no duplicate triggers
Ineffective write	`7lMUcw4of6d6dwaaODIBuy`	FAILED, Airflow run ended failed and verification reported the date
Approval race	`5mEMTgs22yhe7W3CzJRyou`	FAILED before write, stale plan named the externally created date

Prerequisites

Airflow 2.x with the stable REST API enabled and a DAG scheduled daily in UTC.
A Kestra worker that can reach the Airflow webserver on the same network.
Airflow permission to list DAG runs and create DAG runs for the target DAG.

Secrets

POSTGRES_USERNAME and POSTGRES_PASSWORD: Airflow REST API basic-auth credentials for this flow. Use your deployment's secret backend to store the Airflow user and password under these keys.
SLACK_WEBHOOK_URL: Slack incoming webhook used by the errors block.

Inputs

Input	Type	Default	Purpose
`airflow_base_url`	STRING	`http://localhost:8080`	Airflow URL from the Kestra worker
`dag_id`	STRING	`target_dag`	DAG to audit and backfill
`window_start`	STRING	`2026-10-01T00:00:00Z`	Inclusive UTC start
`window_end`	STRING	`2026-10-05T00:00:00Z`	Inclusive UTC end
`max_backfill_count`	INT	`3`	Maximum requested dates
`auto_remediate`	BOOL	`false`	Enables approval and writes
`dry_run`	BOOL	`true`	Keeps execution read-only

Quick start

Start a pinned Airflow test fixture using the reproduction steps and create the POSTGRES_USERNAME, POSTGRES_PASSWORD and SLACK_WEBHOOK_URL secrets.
Import this flow into namespace company.team and run it with the default window. Confirm the plan reports the fixture's absent dates and starts no runs.
Run with auto_remediate: true and dry_run: false, resume the Pause with approved: true and confirm airflow dags list-runs shows one successful run for each requested logical date.
Run the same inputs again. The detector should report zero missing dates and trigger no new runs.

How to extend

Change the schedule interval only when the expected logical-date generator and the DAG's timetable use the same UTC cadence.
Add a second independent partition check when a DAG run can succeed without writing its downstream table.
Route the errors message to paging when a failed backfill needs immediate operator action.

Limits: this blueprint assumes one daily UTC interval and an Airflow API response containing all runs in the requested limit. It proves run existence and terminal state, not downstream table contents. Increase limit and add pagination before using it for a DAG with more than 1000 runs in the response.

Airflow Paused DAG Missed Run Backfill Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Airflow Paused DAG Missed Run Backfill Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

NocoDB Database Platform Health, Storage Engine, and Project Sentinel

Stop a Bad Monthly Load with SLA Assertions on NYC Taxi Data in DuckDB

Object Storage Integrity and Streaming Replica Parity Audit

Search OpenAlex for Related Scholarly Works

OpenMetadata Catalog Health and Change-Event Sentinel

More Infrastructure Blueprints

Neon Postgres Compute Scale And Idle Sentinel

Netlify Deploy Failure Sentinel

Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

Netdata Node Health and System Alarm Sentinel

NetBox Drift Reconciliation with GitHub Audit Trail

NetBird Mesh Peer Connectivity and Security Sentinel

New to Kestra?