Skip to content

Add on_skipped_intervals_callback for catchup=False DAGs - #68359

Open
Tegh25 wants to merge 23 commits into
apache:mainfrom
Tegh25:feature/add_dag_catchup_skip_callback
Open

Add on_skipped_intervals_callback for catchup=False DAGs#68359
Tegh25 wants to merge 23 commits into
apache:mainfrom
Tegh25:feature/add_dag_catchup_skip_callback

Conversation

@Tegh25

@Tegh25 Tegh25 commented Jun 10, 2026

Copy link
Copy Markdown
Contributor

  • Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
  • For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
  • When adding dependency, check compliance with the ASF 3rd Party License Policy.
  • For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

closes: #66791

When a Dag is configured with catchup=False and the scheduler advances past missed intervals (for example after a restart or re-enable), the skip happens silently: no DagRun rows are created and no hook fires. Operators who need observability for skipped partitions (SLA accounting, compliance audit trails, dashboard counters) had to compare last_automated_data_interval against wall clock externally.

This PR adds two symmetric extension points:

  1. on_skipped_intervals_callback — a Dag-level callback on the SDK
    DAG constructor, using the same context-based signature as
    on_success_callback / on_failure_callback. The context includes
    dag, reason ("skipped_intervals"), and skipped_range (a
    DataInterval from the previous automated run's data_interval_end to
    the new run's data_interval_start).

  2. on_intervals_skipped — an AIP-61 listener hookspec for plugins
    that need in-process observability without reloading the Dag file.
    The listener receives a SkippedIntervalsSummary with the same
    skipped_range.

How it works

Skipped intervals are detected at scheduled DagRun creation time (timetable-agnostic):

  • When the scheduler creates a new SCHEDULED run for a catchup=False
    Dag, it compares the new run's data_interval_start against the
    data_interval_end of the most recent prior scheduled run.
  • If there is a gap, it builds a SkippedIntervalsSummary with
    skipped_range spanning those two boundaries. The scheduler does not
    enumerate every skipped interval on the hot path.
  • Detection is gated: this runs only when not dag.catchup AND
    (has_on_skipped_intervals_callback is set OR an on_intervals_skipped
    listener implementation is registered).

Listener fires synchronously in the scheduler (same pattern as on_dag_run_running for Dag start events).

Callback follows the existing Dag callback pipeline:
DagSkippedIntervalsCallbackRequestDatabaseCallbackSink → Dag File Processor reloads the Dag file and invokes the callable. The skipped gap is serialized as a (start, end) datetime pair (skipped_range) in the request because no DagRun row exists for skipped intervals and the processor cannot reliably reconstruct the gap later.

To enumerate individual skipped intervals (for example to count them or log each one), call context["dag"].iter_dagrun_infos_between(skipped_range.start, skipped_range.end) inside the callback, or the equivalent on the Dag in a listener implementation.

Files changed

Area Files
SDK Dag definition task-sdk/src/airflow/sdk/definitions/dag.py, task-sdk/src/airflow/sdk/definitions/context.py
Timetable types airflow-core/src/airflow/timetables/base.py
Serialization serialization/definitions/dag.py, serialized_objects.py
Callback transport callbacks/callback_requests.py, dag_processing/processor.py
Supervisor schema task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_06_23.py, schema.json
Listener listeners/spec/dag.py (new), listeners/listener.py
Scheduler jobs/scheduler_job_runner.py

Usage (preview)

from airflow.sdk import DAG


def notify_skipped(context):
    skipped_range = context["skipped_range"]
    print(f"Gap: {skipped_range.start} -> {skipped_range.end}")
    for info in context["dag"].iter_dagrun_infos_between(
        skipped_range.start, skipped_range.end
    ):
        if info.data_interval is None:
            continue
        interval = info.data_interval
        print(f"Skipped: {interval.start} -> {interval.end}")


with DAG(
    "my_dag",
    schedule="@daily",
    catchup=False,
    on_skipped_intervals_callback=notify_skipped,
) as dag:
    ...

Follow-up (Will be added to this PR)

  • Unit and integration tests (scheduler skip detection, callback execution, listener hook, serialization roundtrip)

  • User documentation (on_skipped_intervals_callback in Dag reference, on_intervals_skipped in listeners docs)

  • Newsfragment in airflow-core/newsfragments/ (user-facing feature)

Test plan

To be completed before marking Ready for review:

  • Scheduler creates skip callback request when catchup=False Dag jumps past intervals and callback is configured

  • Listener on_intervals_skipped fires with correct intervals

  • Dag File Processor executes callback with skipped_intervals in context

  • No overhead when callback and listener are both unset

  • Serialization roundtrip preserves has_on_skipped_intervals_callback

  • prek run --from-ref main --stage pre-commit passes


Was generative AI tooling used to co-author this PR?
  • Yes - Cursor (Auto)
    Generated-by: Cursor (Auto) following the guidelines

@boring-cyborg

boring-cyborg Bot commented Jun 10, 2026

Copy link
Copy Markdown

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

  • Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
  • In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
  • Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
  • Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
  • Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
  • Be sure to read the Airflow Coding style.
  • Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
    Apache Airflow is a community-driven project and together we are making it better 🚀.
    In case of doubts contact the developers at:
    Mailing List: dev@airflow.apache.org
    Slack: https://s.apache.org/airflow-slack

@Tegh25

Tegh25 commented Jun 15, 2026

Copy link
Copy Markdown
Contributor Author

Hi @ferruzzi, please review when you get the chance. Thanks

@ferruzzi

Copy link
Copy Markdown
Contributor

Sorry for the delay, I'll look tomorrow. The failing CI is going to have to be addressed. Run your static-checks locally for a start, then rebase main and do a force-push. We'll see what's left failing and can look at them together if you need a hand.

@Tegh25
Tegh25 force-pushed the feature/add_dag_catchup_skip_callback branch from d5c46ca to 97fb4d3 Compare June 17, 2026 19:51

@ferruzzi ferruzzi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good start, left some comments and I can make another pass when you get those addresseed

Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py Outdated
Comment thread airflow-core/src/airflow/dag_processing/processor.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py Outdated
Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py Outdated
Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py Outdated
Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py Outdated
Comment thread airflow-core/src/airflow/callbacks/callback_requests.py Outdated
@Tegh25
Tegh25 requested review from Lee-W and uranusjr as code owners June 19, 2026 04:32
@Tegh25
Tegh25 force-pushed the feature/add_dag_catchup_skip_callback branch from 0ff46f9 to 4285a9c Compare June 19, 2026 06:48
@Tegh25
Tegh25 requested a review from ferruzzi June 19, 2026 06:48
@Tegh25
Tegh25 force-pushed the feature/add_dag_catchup_skip_callback branch 2 times, most recently from 3e21bbe to 6dad54d Compare June 22, 2026 04:42

@ferruzzi ferruzzi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Left a question for discussion but otherwise LGTM.

Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py
HeadVersion(),
Version("2026-06-23", AddDagSkippedIntervalsCallbackRequest),
Version("2026-06-16"),
Version("2026-05-23"),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please leave this comment open as a reminder to verify this file before we merge.

@uranusjr

Copy link
Copy Markdown
Member

I feel we lack a discussion on this feature. Can someone raise a thread on the dev list so maintainers are aware of this?

@Tegh25

Tegh25 commented Aug 13, 2026

Copy link
Copy Markdown
Contributor Author

Thanks for working on this @Tegh25, could you resolve conflicts and move forward with the PR?

Hi @1fanwang, I will try to get the merge conflicts resolved today.

@Tegh25
Tegh25 force-pushed the feature/add_dag_catchup_skip_callback branch from 3f5851a to 8493d5a Compare August 17, 2026 16:03
@uranusjr

Copy link
Copy Markdown
Member

We should have some test cases on skipped_range since it’s a half-open-ish envelope built from the previous run's data_interval_end and the new run's data_interval_start, both boundaries are off-by-one candidates. Otherwise I think this is generally fine.

Comment thread airflow-core/src/airflow/dag_processing/processor.py Outdated
Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py Outdated


@hookspec
def on_intervals_skipped(dag_id: str, summary: SkippedIntervalsSummary):

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Based on existing naming patterns between callbacks and listeners, this probably should be on_dag_skipped_intervals?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

From what I've observed, listener names are expressed in terms of the event being emitted. Here the event is specifically “intervals skipped” when catchup=False, and the fact that it’s Dag-scoped is already implied by it living in listeners/spec/dag.py and by the scheduler code calling hook.on_intervals_skipped.

I'm happy to change the name if you feel strongly about it. However, given the maturity of this PR with the implementation, tests, docs, newsfragment, etc., it may be more effort than its worth to change the hook name for a small improvement. The original issue also mentioned adding a listener hook with the specific name of on_intervals_skipped.

Tegh25 and others added 23 commits August 19, 2026 12:33
When catchup=False and the scheduler jumps past missed intervals, there
was no way to observe which data intervals were skipped. Add a Dag-level
on_skipped_intervals_callback (context-based, like on_success_callback)
and an AIP-61 listener hook on_intervals_skipped. The scheduler detects
gaps at scheduled run creation and fires the listener in-process; user
callbacks are dispatched to the Dag File Processor after commit.
Scheduler (test_scheduler_job.py)
_collect_skipped_intervals: empty when catchup=True, no callback/listener, no prior run, no gap, returns 3 intervals when a gap exists
_create_dag_runs: returns DagSkippedIntervalsCallbackRequest with correct payload
_do_scheduling: dispatches request via executor.send_callback

Dag processor (test_processor.py)
TestExecuteCallbacks::test_execute_callbacks_routes_skipped_intervals_request: routing in executecallbacks
TestExecuteDagSkippedIntervalsCallback: callback execution, context (dag, reason, skipped_intervals), multiple callbacks, missing callback warning, missing DAG error

Listener (test_skipped_intervals_listener.py + skipped_intervals_listener.py)
on_intervals_skipped fired when the scheduler creates a Dag run across a gap (listener-only, no Dag callback)

Serialization and callback requests
test_dag_serialization.py: has_on_skipped_intervals_callback round-trip (mirrors success/failure callback tests)
test_callback_requests.py: DagSkippedIntervalsCallbackRequest JSON round-trip and CallbackRequest union validation

All 15 new tests pass locally.
Both these fixes were flagged by CI in the previous commit.

Issue 1:
The addition of the `on_skipped_intervals_callback` in the callback types table was not correct syntactically with rst.

Issue 2:
The serialization tests were failing since the `has_on_skipped_intervals_callback` key was missing but not expected.
Since this callback was implemented similarly to `has_on_success_callback`, it has been added to the ignored set.
When a Dag with catchup=False skips many scheduled intervals (long pause,
scheduler downtime, etc.), the original design enumerated every missed
interval in the scheduler via iter_dagrun_infos_between(), materialized
the full list in DagSkippedIntervalsCallbackRequest, and passed it through
the listener hook and dag-processor callback. That is O(n) in time and
memory and can stall the scheduler or produce huge callback payloads for
high-frequency schedules.

Replace the full interval list with a compact SkippedIntervalsSummary
(count plus skipped_range envelope) and compute the count in O(1) time
from the schedule and gap boundaries.

Scheduler and Dag integration:
- Add SkippedIntervalsSummary (count, skipped_range) in timetables.base.
- Add Timetable.count_skipped_intervals_between(); default raises
  NotImplementedError.
- Implement O(1) counting on DeltaMixin (timedelta/relativedelta schedules)
  and CronMixin (uniform cron periods, plus calendar shortcuts for
  monthly and weekly expressions).
- Add SerializedDAG.summarize_skipped_intervals_between() delegating to
  the Dag timetable; returns None when there is no gap or counting is
  unsupported.
- Change SchedulerJobRunner._collect_skipped_intervals() to return
  SkippedIntervalsSummary | None instead of a list of interval tuples.

Callback and listener API:
- on_intervals_skipped listener: (dag_id, summary: SkippedIntervalsSummary)
  instead of (dag_id, skipped_intervals: list[DataInterval]).
- on_skipped_intervals_callback context: skipped_interval_count and
  skipped_range instead of skipped_intervals.
- DagSkippedIntervalsCallbackRequest carries skipped_interval_count and
  skipped_range; add from_summary() / to_summary() helpers.
- Dag processor builds the callback context from the summary.

Documentation and examples:
- Update callbacks.rst and listeners.rst for the new context and listener
  signature; document iter_dagrun_infos_between() for full enumeration in
  the dag processor.
- Update SDK Dag docstring and example event_listener plugin.

Schema:
- Update task-sdk supervisor schema and v2026_06_16 version migration for
  the revised DagSkippedIntervalsCallbackRequest fields.

Tests:
- Add test_skipped_intervals_summary.py with parametrized O(1) vs reference
  counts (delta/cron daily, hourly, monthly) and summarize helpers.
- Update scheduler, callback request, dag processor, and listener tests
  for the summary shape.

Custom timetables must implement count_skipped_intervals_between() to
participate; unsupported built-in timetables silently skip notification.
Cron expressions outside the supported fast paths raise
AirflowTimetableInvalid from CronMixin (not swallowed by summarize).
…xt, update schema versioning

The skipped-intervals callback receives a dedicated payload, not task template
Context; a proper TypedDict removes the mypy ignore and matches runtime behavior.

Updated the api version in the schema from 2026-06-16 -> 2026-06-23.
Co-authored-by: D. Ferruzzi <ferruzzi@amazon.com>
Calculating the skipped_interval_count would either take O(n) time in the scheduler or would require
mixin methods that would be unreliable for custom timetables. Instead, this field has been removed so
that the skipped intervals callback can work reliably and efficiently. Users can access the specific
skipped interval data by calling other methods in their callback/listener implementations.
All source, tests, and docs have been updated to reflect this change.
When the schema was bumped to 2026-06-23, version 2026-06-16 was removed
from the bundle, breaking coordinator tests and any bundles compiled
against that schema version. Restore the intermediate version to the
bundle chain to support in-flight deployments.

Also document SkippedIntervalsCallbackContext in the public API docs
since it was added to __all__ but missing from the Sphinx inventory.
Originally, this listener was called within _create_dag_runs(). This meant that the
hook is called repeatedly if transaction retries occur.
This change moves the listener call into _do_scheduling(), after the DB commit.
…kipped interval fields

Replaced field-level schema(...).didnt_exist instructions with a
convert_response_to_previous_version_for(DagFileParseRequest) hook that filters
callback_requests, dropping entries with type == "DagSkippedIntervalsCallbackRequest" when
downgrading for older lang-SDK bundles. The original instructions did not accurately revert
the schema version.
…ble runs

Default Airflow 3 schedules use zero-width CronTriggerTimetable /
DeltaTriggerTimetable intervals, so a bare prev_end < new_start check
treated every normal schedule advance as a catchup gap. Detect skips by
comparing the timetable's expected next logical date with the run being
created, and compute the listener-hook presence once per scheduling loop.
Skipped intervals fires when no Dag run is made, so it will have no
run_id. The former implementation expected a run_id in the request.
Keep Dag callback exception tags consistent across callback paths and centralize
skipped-interval precondition handling in one place to reduce duplicate checks.
@Tegh25
Tegh25 force-pushed the feature/add_dag_catchup_skip_callback branch from 508ee0f to 20546cd Compare August 19, 2026 17:35
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add on_catchup_skip_callback for DAGs with catchup=False

5 participants