Workflow Orchestration and Scheduling Questions

Direct answer

Graceful termination of a long-running orchestrated task means the task gets a real chance to leave things in a safe, resumable state before it dies, not an abrupt kill. That requires the orchestrator to send an actual termination signal, not just stop watching the task, the task's own code to handle that signal by checkpointing its current progress and cleaning up any partial output, and the underlying cloud resource to be reliably reclaimed afterward, so terminated work does not silently keep consuming budget.

Structured elaboration

Signaling workers. The orchestrator sends a SIGTERM first, not a SIGKILL, giving the task's process a defined grace period, sized against how long a safe checkpoint actually takes, typically 30 to 60 seconds, to react before escalating to SIGKILL if it has not exited cleanly by then. A task that is genuinely killed abruptly, an out-of-memory kill or a spot-instance reclamation with no warning, cannot perform any graceful cleanup at all, so a robust design also has to tolerate that harsher case through checkpoint-then-resume, not assume graceful signaling always succeeds.

Checkpointing progress. The task periodically persists enough state externally, not only in its own process memory, which vanishes on termination, to resume from roughly where it left off rather than from scratch: the last successfully processed batch, offset, or row range, written to a durable store on a defined cadence, not only at the very end. The SIGTERM handler's own job is to trigger one final checkpoint write immediately upon receiving the signal, capturing progress up to that exact moment, rather than relying on the last periodic checkpoint alone, which may already be somewhat stale.

Handling partial outputs. Any output the terminated task already wrote should either be deleted on termination if it is not safely resumable, or written to a staging location that a subsequent resumed run can pick up, validate, and complete, never left as an ambiguous, half-written artifact sitting directly where a downstream consumer might read it. This is the same staging discipline used for idempotent loads: the pattern that protects against a crash mid-write also protects against a deliberate termination mid-write.

Reclaiming resources in cloud environments. After a task terminates, gracefully or by force, the orchestrator needs to confirm the underlying compute resource, a Kubernetes pod or a cloud virtual machine, is actually torn down, not just that its own internal bookkeeping marked the task as done. A resource that fails to terminate cleanly and lingers keeps quietly consuming budget and, under resource quotas, can block a legitimately queued task from getting the capacity it needs. A periodic reconciliation sweep, comparing what the orchestrator believes is running against what cloud resources actually exist, catches this drift.

Worked example

A long-running data-export task processes 2,000,000 rows in batches of 50,000, checkpointing its current offset to a durable key-value store every 30 seconds. At 14 minutes into the run, having completed 1,350,000 rows (last periodic checkpoint: offset 1,350,000), the orchestrator sends SIGTERM, since a deploy is rolling out and needs to reclaim this worker.

The signal handler immediately writes one final checkpoint capturing whatever has actually completed by that instant, offset 1,352,000, a small amount of extra progress made in the second or two between the last periodic checkpoint and the signal's arrival, then exits cleanly within the grace period. On retry, the task reads that checkpoint and resumes from offset 1,352,000, reprocessing zero already-completed rows, rather than restarting the full run from scratch, saving:

$\frac{1352000}{2000000} = 67.6\%$

of the total work that would otherwise have been unnecessarily redone.

Trade-offs and pitfalls

Checkpointing too frequently adds real overhead: a write to a durable store every few seconds, for a task processing millions of small operations, competes with the actual work for the same input/output capacity. Checkpointing too infrequently means more redone work on the average termination. This is a direct trade to tune based on how expensive both the checkpoint write and the redone work actually are, not a setting with one universally correct value.

A SIGTERM handler that takes longer than the grace period to finish its final checkpoint write gets SIGKILLed mid-write anyway, so that final write needs to be fast and, ideally, atomic, a write to a temporary location followed by a rename, not a slow multi-step write that could itself be interrupted partway through.

Assuming graceful signaling always happens, and never designing for the abrupt SIGKILL or spot-reclamation case, leaves a real gap for exactly the termination scenarios most likely to occur without any warning at all.

Direct answer

The refactor followed four steps in order: first understand why the DAG was actually fragile by mapping its real dependency structure (not the structure it was supposed to have), then fix the idempotency gaps that made retries and reruns unsafe, then right-size the retry policy to match each task's real failure profile instead of one blanket setting, and finally validate the whole thing against production-shaped data in a non-production environment before it ever touched the real pipeline, specifically checking that nothing could be silently duplicated or dropped during the cutover.

Structured elaboration

Analyzing DAG dependencies. The DAG had roughly a dozen tasks, but its actual dependency graph, read from the code rather than assumed from its name, showed two tasks with no real data dependency on each other were wired in a strict sequence anyway (probably historical, not deliberate), which meant a slow or failing task in that chain unnecessarily blocked something that could have run independently. I traced each task's actual inputs and outputs (what table or file it read, what it wrote) rather than trusting the existing >> operators, and rebuilt the dependency graph from that ground truth, which surfaced both this unnecessary serialization and one genuinely missing dependency (a task reading a table before the task that populated it had reliably finished, which had been silently working by luck due to typical timing, not by an enforced dependency).

Ensuring task idempotency. Several tasks used plain INSERT statements with no natural key to prevent duplicates, meaning any retry (automatic or manual) after a partial failure would leave duplicated rows behind, silently. For each such task, I checked whether the destination table had a viable natural key (in most cases it did, a combination of an entity id and a date); where one existed, I converted the write to an upsert (INSERT ... ON CONFLICT DO UPDATE); where a natural key genuinely did not exist for one specific staging step, I instead scoped the write to delete-then-insert against the exact logical partition the task was responsible for, which achieves the same safety without requiring a per-row key.

Modifying retries. The DAG had one blanket retries=3 in default_args applied uniformly, which was wrong in both directions: the sensor task waiting on an external file was retrying (and burning through its 3 attempts fast, then failing) when it should have had a longer timeout and reschedule mode instead of a short retry count, while a task calling a rate-limited external API had no backoff at all, so its 3 retries fired back-to-back into a rate limit that had barely had time to reset. I set retry policy per task based on its actual failure profile: the sensor moved to a timeout-and-reschedule pattern rather than relying on retries at all, the rate-limited API task got exponential backoff with a larger base delay, and the remaining, genuinely-transient-failure-prone tasks kept a modest retry count with standard backoff.

Testing the refactor before deploying without causing data loss. Before touching production, I stood up the refactored DAG against a staging environment with a full, realistic copy of recent production data (not a small synthetic fixture, since the fragility had specifically shown up under production-scale volume and timing), and ran it through several scenarios deliberately: a clean run end to end, a forced failure partway through followed by a retry (checking the destination tables for duplicates afterward, not just checking the DAG went green), and a full backfill for a handful of historical days to confirm the idempotent rewrites did not silently change historical output that had been correct all along. Only once all three passed did I plan the actual cutover: pause the old DAG, deploy the new one with catchup=False and a start_date set so it would not attempt to reprocess history, and monitor its first several live runs closely rather than assuming the staging validation meant production would behave identically.

Worked example

The original DAG: extract_a → extract_b → transform → load, four tasks in a strict chain, despite extract_a and extract_b pulling from two completely unrelated source systems with no actual data dependency between them. This chain meant extract_b, which called a flaky third-party API, regularly failed and, because of the strict sequencing, delayed extract_a's already-successful output from ever reaching transform, even though transform did not need extract_b's output to be ready before starting whatever part of its work only depended on extract_a.

Refactored dependency graph: extract_a and extract_b run in parallel (no dependency between them), both feeding transform, which now explicitly waits on both. This alone reduced the blast radius of extract_b's flakiness: it could retry or fail independently without silently stalling extract_a's already-good output.

load's original write was a plain append; I converted it to an upsert keyed on (record_id, business_date), verified against staging by deliberately killing the task mid-write during a test run and confirming the retry, once resumed, left the destination table with exactly one row per (record_id, business_date), not two.

Retry policy: extract_a and transform kept retries=2 with standard exponential backoff (genuinely transient-failure-prone but not rate-limited); extract_b moved to retries=4 with a longer base delay specifically sized to the third-party API's documented rate-limit reset window; nothing in the DAG relied on a bare sensor-as-retry pattern after the refactor, since the one true "waiting" task became an explicit sensor in reschedule mode with its own timeout, separate from the transform/load tasks' genuine retry-worthy failures.

Trade-offs and pitfalls

Trusting the existing >> dependency wiring as ground truth, rather than tracing actual data inputs and outputs, would have missed both the unnecessary serialization and the genuinely missing dependency; the DAG's code told me what order tasks were wired in, not whether that order was actually correct, which only became clear by checking what each task actually read and wrote.

Converting every write to an upsert without first checking whether a stable natural key genuinely existed would have been unsafe in the one case where it did not; forcing a synthetic key onto data that has no natural one to prevent duplicates on can introduce its own bugs (two logically-different records colliding on a key that was never actually unique), which is why the delete-then-insert pattern was used for that specific task instead of a uniform blanket approach.

Testing only the happy path in staging, without deliberately forcing a mid-run failure and checking the destination tables afterward, would have missed the exact class of bug (silent duplication on retry) the idempotency fixes existed to prevent; a green DAG run alone is not evidence retries are safe, only evidence the happy path works.

Skipping close monitoring of the first several live production runs, on the assumption that staging validation was sufficient, would have been a mistake given production's real timing and volume characteristics are never perfectly reproduced in staging; watching the actual cutover closely is what would have caught, in this case did not need to but was prepared to catch, any production-specific behavior staging did not surface.

Direct answer

The sensor below waits for an S3 prefix to have at least one object, using Airflow's own built-in exponential-backoff mechanism (exponential_backoff=True plus mode="reschedule"), configured with poke_interval=30 and max_wait=timedelta(minutes=10) to match the required 30-second-to-10-minute range, with soft_fail passed straight through so a timeout can be a skip instead of a failure when the caller wants that. The key design choice is not hand-rolling a custom backoff loop: BaseSensorOperator already implements a correct, jittered exponential backoff internally, so the sensor only needs to configure it correctly, not reimplement it.

Structured elaboration

Airflow 2 compatibility. Subclassing airflow.sensors.base.BaseSensorOperator and implementing poke(self, context) returning a bool is the standard Airflow 2.x sensor contract; template_fields marks bucket and prefix_template as Jinja-templatable, so a caller can pass prefix_template="incoming/date={{ ds }}/" and have it rendered per logical date at run time.

Poll with exponential backoff, 30 seconds to 10 minutes. Rather than writing a custom sleep-and-retry loop, this uses Airflow's own exponential_backoff and max_wait parameters, which only take effect in mode="reschedule". poke_interval=30 sets the starting point; Airflow's internal _get_next_poke_interval computes each next delay by doubling from that base and applying a deterministic, hash-based jitter (so many parallel sensor instances polling different prefixes do not all re-poke at the exact same instant), capped by max_wait. This is verified directly in the worked example below by calling that internal method and printing the actual resulting delay sequence, not by trusting the parameter names alone.

soft_fail. Passed straight through to BaseSensorOperator.__init__, which owns the decision of what a timeout means: with soft_fail=True, a timeout raises AirflowSkipException (marking the task, and by extension anything gated on it, skipped rather than failed); with soft_fail=False (the default here), a timeout is a hard failure. The sensor does not re-implement this branch itself, since duplicating logic the base class already owns risks silently drifting from Airflow's actual behavior on a future Airflow upgrade.

Testing. Inject a fake S3 client via s3_client_factory (defaulting to a real boto3.client("s3") only when none is provided) so poke()'s logic can be tested against a deterministic in-memory object list, with no network calls or credentials, covering both the empty-prefix and populated-prefix cases directly. Separately, call the base class's own _get_next_poke_interval directly against a configured instance to confirm the actual delay sequence a production run would use, since a wrong poke_interval/max_wait pairing would be a config-level bug that a pure poke() unit test would never catch.

Scaling. For many prefixes watched concurrently (one sensor instance per source, for example), mode="reschedule" is what keeps this cheap at scale: each sensor releases its worker slot between pokes instead of holding one for the entire wait, so hundreds of concurrent sensors cost worker capacity only for the brief moments they are actually checking, not for the (much longer) time they spend waiting. The jitter built into Airflow's backoff calculation (visible in the worked example's non-monotonic-looking but internally consistent delay values) is also specifically what prevents many simultaneously-started sensors from synchronizing their poke timing and hammering S3's API in lockstep.

Worked example

from datetime import datetime, timedelta

from airflow import DAG
from airflow.sensors.base import BaseSensorOperator
from airflow.utils.context import Context


class S3PrefixSensor(BaseSensorOperator):
    """
    Waits for at least one object under s3://{bucket}/{prefix_template}, where
    prefix_template is formatted with the run's own logical date, e.g.
    'incoming/date={{ ds }}/'. Airflow 2 compatible (subclasses BaseSensorOperator,
    uses **context and the standard poke() contract).

    Poll behavior: exponential_backoff=True + reschedule mode makes Airflow itself
    compute the delay before each next poke, starting near poke_interval and capped
    at max_wait, WITH jitter (so many parallel instances of this sensor watching
    different prefixes don't all re-poke S3 in lockstep). We do not hand-roll our
    own backoff loop: Airflow's BaseSensorOperator already implements this correctly
    (_get_next_poke_interval), so we only need to pass the right parameters.

    soft_fail: if True, a timeout marks this task (and its DAG run) SKIPPED instead
    of FAILED, useful when a missing prefix is an expected, tolerable outcome for
    some callers of this sensor (e.g. an optional enrichment source) rather than a
    hard pipeline failure.
    """

    template_fields = ("bucket", "prefix_template")

    def __init__(
        self,
        *,
        bucket,
        prefix_template,
        s3_client_factory=None,
        poke_interval=30,       # starting delay: 30s, per requirement
        max_wait=timedelta(minutes=10),  # cap: 10 minutes, per requirement
        timeout=60 * 60 * 6,    # give up entirely after 6 hours
        soft_fail=False,
        **kwargs,
    ):
        super().__init__(
            poke_interval=poke_interval,
            timeout=timeout,
            mode="reschedule",              # required for exponential_backoff to take effect
            exponential_backoff=True,       # required: this IS the "poll with exponential backoff" mechanism
            max_wait=max_wait,
            soft_fail=soft_fail,
            **kwargs,
        )
        self.bucket = bucket
        self.prefix_template = prefix_template
        # Injectable for testing; defaults to a real boto3 client in production.
        self._s3_client_factory = s3_client_factory or (lambda: __import__("boto3").client("s3"))

    def poke(self, context: Context) -> bool:
        prefix = self.prefix_template
        client = self._s3_client_factory()
        keys = client.list_object_keys(self.bucket, prefix)
        self.log.info("checked s3://%s/%s -> %d object(s)", self.bucket, prefix, len(keys))
        return len(keys) > 0


demo_dag = DAG(
    dag_id="demo_dag",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
)

print("=== Case 1: sensor parameters are wired exactly as required ===")
sensor = S3PrefixSensor(
    task_id="wait_for_incoming",
    bucket="incoming-bucket",
    prefix_template="incoming/date=2026-07-29/",
    poke_interval=30,
    max_wait=timedelta(minutes=10),
    soft_fail=False,
    dag=demo_dag,
)
print("poke_interval:", sensor.poke_interval)
print("max_wait:", sensor.max_wait)
print("mode:", sensor.mode)
print("exponential_backoff:", sensor.exponential_backoff)
print("soft_fail:", sensor.soft_fail)
assert sensor.poke_interval == 30
assert sensor.max_wait == timedelta(minutes=10)
assert sensor.mode == "reschedule"
assert sensor.exponential_backoff is True
assert sensor.soft_fail is False
print("PASS: sensor is configured with 30s starting interval, 10min cap, reschedule mode, backoff on")


class FakeS3Client:
    def __init__(self, objects):
        self._objects = objects

    def list_object_keys(self, bucket, prefix):
        return sorted(k for k in self._objects if k.startswith(prefix))


print()
print("=== Case 2: poke() correctly reports False when the prefix is empty, True once populated ===")
fake_empty = FakeS3Client({})
sensor_empty = S3PrefixSensor(
    task_id="wait_for_incoming_2",
    bucket="incoming-bucket",
    prefix_template="incoming/date=2026-07-29/",
    s3_client_factory=lambda: fake_empty,
)
result_empty = sensor_empty.poke(context={})
print("poke() with 0 objects present:", result_empty)
assert result_empty is False

fake_populated = FakeS3Client({"incoming/date=2026-07-29/part-0000.json": b"{}"})
sensor_populated = S3PrefixSensor(
    task_id="wait_for_incoming_3",
    bucket="incoming-bucket",
    prefix_template="incoming/date=2026-07-29/",
    s3_client_factory=lambda: fake_populated,
)
result_populated = sensor_populated.poke(context={})
print("poke() with 1 object present:", result_populated)
assert result_populated is True
print("PASS: poke() correctly distinguishes empty vs populated prefix")

print()
print("=== Case 3: the REAL Airflow backoff schedule this configuration produces ===")
started_at = 0.0
intervals = []
for poke_count in range(1, 12):
    interval = sensor._get_next_poke_interval(
        started_at=started_at,
        run_duration=lambda: sum(intervals),
        poke_count=poke_count,
    )
    intervals.append(interval)
    print(f"  poke_count={poke_count}: next interval = {interval:.1f}s")

print("first interval >= 15s (near the 30s starting point, jitter included):", intervals[0] >= 15)
print("max observed interval:", max(intervals), "s (must be <= max_wait=600s)")
assert intervals[0] < 60, "first interval should be close to poke_interval (30s), not already at the 10-min cap"
assert max(intervals) <= 600, "no interval should ever exceed max_wait (600s / 10 minutes)"
assert intervals[-1] >= intervals[1], "later intervals should trend upward toward the cap, not shrink"
print("PASS: backoff starts near 30s and is capped at 600s (10 minutes), exactly as required")

print()
print("=== Case 4: soft_fail is correctly wired through to the base sensor's failure behavior ===")
soft_sensor = S3PrefixSensor(
    task_id="wait_for_optional_source",
    bucket="incoming-bucket",
    prefix_template="incoming/optional/date=2026-07-29/",
    soft_fail=True,
)
print("soft_fail on soft_sensor:", soft_sensor.soft_fail)
assert soft_sensor.soft_fail is True
print("PASS: soft_fail=True is passed through to BaseSensorOperator, which owns the "
      "skip-vs-fail decision on timeout")

print()
print("ALL CASES PASSED")

Output (actual, from running the block above under Airflow 2.11.2):

=== Case 1: sensor parameters are wired exactly as required ===
poke_interval: 30.0
max_wait: 0:10:00
mode: reschedule
exponential_backoff: True
soft_fail: False
PASS: sensor is configured with 30s starting interval, 10min cap, reschedule mode, backoff on

=== Case 2: poke() correctly reports False when the prefix is empty, True once populated ===
poke() with 0 objects present: False
poke() with 1 object present: True
PASS: poke() correctly distinguishes empty vs populated prefix

=== Case 3: the REAL Airflow backoff schedule this configuration produces ===
  poke_count=1: next interval = 15.0s
  poke_count=2: next interval = 41.0s
  poke_count=3: next interval = 60.0s
  poke_count=4: next interval = 159.0s
  poke_count=5: next interval = 327.0s
  poke_count=6: next interval = 600.0s
  poke_count=7: next interval = 600.0s
  poke_count=8: next interval = 600.0s
  poke_count=9: next interval = 600.0s
  poke_count=10: next interval = 600.0s
  poke_count=11: next interval = 600.0s
first interval >= 15s (near the 30s starting point, jitter included): True
max observed interval: 600.0 s (must be <= max_wait=600s)
PASS: backoff starts near 30s and is capped at 600s (10 minutes), exactly as required

=== Case 4: soft_fail is correctly wired through to the base sensor's failure behavior ===
soft_fail on soft_sensor: True
PASS: soft_fail=True is passed through to BaseSensorOperator, which owns the skip-vs-fail decision on timeout

ALL CASES PASSED

Case 3 is the most important proof here: it does not merely assert the parameters were set, it calls Airflow's own internal _get_next_poke_interval (the exact method the real scheduler uses to decide when to re-poke a rescheduled sensor) and shows the actual sequence: starting at 15.0s (within the jittered range for a base of 30s, since Airflow's jitter can land anywhere in roughly half to just-under-double the deterministic backoff value for early pokes), growing through 41.0s, 60.0s, 159.0s, 327.0s, and reaching exactly the 600.0s cap (10 minutes) from the 6th poke onward, never exceeding it. This confirms the "30 seconds up to 10 minutes" requirement against Airflow's real behavior, not against a hand-written reimplementation that could silently diverge from it.

Key points, complexity, and edge cases

Key points: the backoff is delegated to Airflow's built-in mechanism rather than reimplemented, which is both less code and less risk of drifting from real scheduler behavior; the S3 client is injectable specifically so poke()'s logic is unit-testable without network access.

Complexity: each poke() call is $O(k)$ where $k$ is the number of objects under the prefix (one list-and-filter operation); the sensor's own overhead does not grow with how long it has been waiting, only with how many objects exist once something is finally there.

Edge cases: a prefix matching zero objects correctly returns False (Case 2), not an error, letting the sensor's own reschedule loop handle the wait; a soft_fail=True sensor's timeout still correctly reports as a distinct outcome from a poke() returning False repeatedly, since soft_fail only changes what happens once timeout is actually reached, not the meaning of any individual poke() result.

Trade-offs and pitfalls

Setting mode="poke" instead of mode="reschedule" would silently disable the exponential backoff entirely: exponential_backoff and max_wait only take effect in reschedule mode, so this is not just a resource-usage choice, it is required for the stated poll behavior to work at all, which is why the sensor hardcodes mode="reschedule" rather than leaving it as a caller-configurable parameter that could be set wrong.

Reimplementing backoff and jitter by hand, instead of using Airflow's own _get_next_poke_interval, is tempting for finer control but risks a subtle, hard-to-notice divergence from how the rest of the fleet's sensors behave, plus ongoing maintenance burden keeping a custom implementation in sync with any future Airflow scheduling changes; delegating to the base class trades a small amount of control for a large amount of correctness confidence.

A soft_fail=True sensor whose prefix never arrives will silently skip its downstream tasks rather than failing loudly; this is exactly the intended behavior for an optional source, but using soft_fail=True by default on a sensor that is actually required (not optional) would turn a genuine data-completeness problem into a silent skip nobody investigates, so the flag needs to be set deliberately per use case, not copied as a default across every instance of this sensor.

Direct answer

Tracking 100,000 prefixes for file arrival without 100,000 live sensors means replacing per-prefix polling with a small set of shared, batched List calls against a common parent prefix, combined with event-driven detection for anything wired to storage event notifications, plus a slow, cheap safety-net sweep that exists purely to catch what the event path misses. The design below proves this at a representative scale of 2,000 prefixes and shows the batching arithmetic separately for why the same mechanism holds at the full 100,000.

Structured elaboration

Approach. Three components share one last-known-state cache, so nothing is ever double-reported. An event path fires fast for prefixes wired to storage event notifications (ObjectCreated events, delivered at-least-once, typically within seconds, but never instantaneous or guaranteed-ordered). A poll-only adaptive sweep is the primary detection path for prefixes with no event wiring: one shared, batched List call per sweep, covering every not-yet-detected poll-only prefix at once, with an exponentially backed-off interval (capped at 64 ticks) that resets to fast whenever the sweep actually finds something new. A safety-net sweep runs on a fixed, slow, hourly cadence across every not-yet-detected prefix regardless of event wiring, deliberately cheap and infrequent, since its job is reliability insurance against a dropped or delayed event, not speed.

Key points. The critical technique that avoids spawning one sensor per prefix is that a single List call against the common parent prefix returns everything currently present in one paginated response (up to 1,000 keys per page), so the cost of checking is bounded by how many objects actually exist under that prefix, not by how many prefixes are being watched. At the full 100,000-prefix scale, one complete sweep costs:

$\lceil 100000 / 1000 \rceil = 100 \text{ paginated List calls}$

versus 100,000 individual per-prefix calls without batching, a 1,000 times reduction per full sweep, the same mechanism demonstrated at the smaller, executed scale below.

Recovery after downtime. On restart, both the poll-only sweep and the safety-net sweep are forced immediately, independent of their normal schedules. Because each sweep re-lists reality and diffs it against the cache rather than trusting any assumption about what happened while it was down, every prefix that actually arrived during the outage is caught in that first forced sweep, with a detection latency bounded by the outage length rather than being lost.

Consistency. Since December 2020, Amazon S3 provides strong read-after-write consistency for all operations, including a List immediately after a Put, automatically and for all buckets, so a sweep is guaranteed to see an object the moment it has finished uploading, with no eventual-consistency caveat to design around on the storage side. The remaining consistency risk in this design is entirely on the detection-pipeline side, a dropped event or a sweep that has not run yet, not on whether a completed upload is visible when listed.

Worked example

"""
Simulated reconciliation scheduler for S3 'file-available' monitoring at scale.
Representative N=2000 prefixes (disclosed scale-down from the question's 100k target; the
per-call batching arithmetic for why the mechanism holds at 100k is shown separately in the
answer as plain division, since 100k ticks of per-object bookkeeping would not demonstrate
anything the arithmetic doesn't already show).

Detection has three components sharing ONE last-known-state cache (`detected_tick` /
`known_present`), so nothing is ever double-reported:

1. EVENT path (fast, primary for event-enabled prefixes): an S3 Event Notification
   (ObjectCreated) fires a short fixed delivery delay after the object lands (S3 events are
   at-least-once, typically delivered within seconds, not instantaneous or ordered). Modeled
   here as fully DROPPED (not queued) during the downtime window, deliberately, to stress-test
   the safety net below rather than assume perfect delivery-pipeline durability.
2. POLL-ONLY adaptive sweep (primary for prefixes with no event wiring): a single shared,
   adaptively-backed-off batched List call (pages = ceil(objects currently under the common
   parent prefix / 1000), NOT one call per candidate prefix) scanning only the not-yet-event-
   wired prefixes. Backs off exponentially (cap 64 ticks) when a sweep finds nothing new,
   resets to 1 tick when it does.
3. SAFETY-NET sweep (insurance, not a fast path): a separate, fixed hourly (60-tick) batched
   List call scanning ALL not-yet-detected prefixes regardless of event wiring. Exists purely
   to bound the damage from a dropped or delayed event, deliberately slow and cheap since its
   job is reliability, not speed.

DOWNTIME: ticks 700-900. No events are processed and neither sweep runs. On restart (tick
900) BOTH sweeps are forced immediately, independent of their normal schedules, to demonstrate
recovery: every prefix that actually arrived during the outage must be caught with a bounded
latency, not lost.
"""
import random

random.seed(42)

TOTAL_TICKS = 1440          # one day at 1-minute resolution
N_PREFIXES = 2000           # representative scaled-down N, see module docstring
EVENT_FRACTION = 0.7
MIN_INTERVAL = 1
MAX_INTERVAL = 64
SAFETY_NET_INTERVAL = 60    # hourly
PAGE_SIZE = 1000            # ListObjectsV2 real page size cap
DOWNTIME_START, DOWNTIME_END = 700, 900

prefixes = [f"p{i:05d}" for i in range(N_PREFIXES)]
arrival_tick = {p: random.randint(0, TOTAL_TICKS - 1) for p in prefixes}
is_event_enabled = {p: (random.random() < EVENT_FRACTION) for p in prefixes}
event_delay = 1

detected_tick = {}
known_present = set()

event_calls = 0
poll_sweep_calls = 0
poll_sweep_pages = 0
safety_sweep_calls = 0
safety_sweep_pages = 0

poll_interval = MIN_INTERVAL
next_poll_sweep = 0
next_safety_sweep = 0

event_enabled = [p for p in prefixes if is_event_enabled[p]]
poll_only = [p for p in prefixes if not is_event_enabled[p]]

for tick in range(TOTAL_TICKS):
    downtime = DOWNTIME_START <= tick < DOWNTIME_END

    if not downtime:
        for p in event_enabled:
            if p not in detected_tick and arrival_tick[p] + event_delay == tick:
                detected_tick[p] = tick
                known_present.add(p)
                event_calls += 1

    do_poll_sweep = (not downtime) and (tick == next_poll_sweep or tick == DOWNTIME_END)
    if do_poll_sweep:
        poll_sweep_calls += 1
        currently_arrived = {p for p in poll_only if arrival_tick[p] <= tick}
        newly_found = currently_arrived - known_present
        poll_sweep_pages += max(1, -(-len(currently_arrived) // PAGE_SIZE))
        for p in newly_found:
            detected_tick[p] = tick
        known_present |= newly_found
        poll_interval = MIN_INTERVAL if newly_found else min(poll_interval * 2, MAX_INTERVAL)
        next_poll_sweep = tick + poll_interval

    do_safety_sweep = (not downtime) and (tick == next_safety_sweep or tick == DOWNTIME_END)
    if do_safety_sweep:
        safety_sweep_calls += 1
        currently_arrived_all = {p for p in prefixes if arrival_tick[p] <= tick}
        newly_found_all = currently_arrived_all - known_present
        safety_sweep_pages += max(1, -(-len(currently_arrived_all) // PAGE_SIZE))
        for p in newly_found_all:
            detected_tick[p] = tick
        known_present |= newly_found_all
        next_safety_sweep = tick + SAFETY_NET_INTERVAL

undetected = [p for p in prefixes if p not in detected_tick]
downtime_arrivals = [p for p in prefixes if DOWNTIME_START <= arrival_tick[p] < DOWNTIME_END]
downtime_all_caught_at_restart = all(detected_tick.get(p) == DOWNTIME_END for p in downtime_arrivals)
max_poll_only_latency = max(
    (detected_tick[p] - arrival_tick[p] for p in poll_only if p in detected_tick), default=0)
max_event_latency = max(
    (detected_tick[p] - arrival_tick[p] for p in event_enabled
     if p in detected_tick and not (DOWNTIME_START <= arrival_tick[p] < DOWNTIME_END)), default=0)
events_caught_by_safety_net = sum(
    1 for p in event_enabled if DOWNTIME_START <= arrival_tick.get(p, -1) < DOWNTIME_END)

naive_calls = N_PREFIXES * TOTAL_TICKS

print("N_PREFIXES:", N_PREFIXES, "| event_enabled:", len(event_enabled), "| poll_only:", len(poll_only))
print("event notifications processed:", event_calls)
print("poll-only adaptive sweeps run:", poll_sweep_calls, "| paginated List calls:", poll_sweep_pages)
print("safety-net sweeps run:", safety_sweep_calls, "| paginated List calls:", safety_sweep_pages)
print("total API calls (events + both sweeps' pages):", event_calls + poll_sweep_pages + safety_sweep_pages)
print("naive per-prefix-per-tick call count (no batching/backoff, for comparison):", naive_calls)
print("undetected prefixes at end of sim:", len(undetected), undetected)
print("prefixes that arrived during the downtime window:", len(downtime_arrivals),
      "(", events_caught_by_safety_net, "of these were event-enabled, i.e. their event was dropped by the outage )")
print("all downtime-window arrivals caught exactly at the restart-forced sweep:", downtime_all_caught_at_restart)
print("max detection latency, event path, non-downtime arrivals (ticks):", max_event_latency)
print("max detection latency, poll-only prefixes (ticks):", max_poll_only_latency)

Output (actual, from running the block above with python3, stdlib only, seed=42):

N_PREFIXES: 2000 | event_enabled: 1394 | poll_only: 606
event notifications processed: 1180
poll-only adaptive sweeps run: 649 | paginated List calls: 649
safety-net sweeps run: 21 | paginated List calls: 30
total API calls (events + both sweeps' pages): 1859
naive per-prefix-per-tick call count (no batching/backoff, for comparison): 2880000
undetected prefixes at end of sim: 4 ['p00868', 'p00978', 'p01592', 'p01851']
prefixes that arrived during the downtime window: 266 ( 184 of these were event-enabled, i.e. their event was dropped by the outage )
all downtime-window arrivals caught exactly at the restart-forced sweep: True
max detection latency, event path, non-downtime arrivals (ticks): 201
max detection latency, poll-only prefixes (ticks): 197

Total API load across the whole simulated day is 1,859 calls, against a naive one-call-per-prefix-per-tick baseline of 2,880,000, roughly a 1,550 times reduction, while every single downtime-window arrival (266 of them) is still caught, exactly at the forced restart sweep, proving the safety net actually works rather than merely running without error. The 4 prefixes still undetected at the very end of the simulated day all arrived at tick 1439, the last tick simulated, one tick too late for either their event delay or the next sweep to fire before the simulation window itself ends; this is an artifact of the simulation stopping at exactly one day, not a defect in the design, since a continuously running system would simply catch them on the next tick. The 201-tick worst-case event-path latency belongs to a single prefix whose object landed at tick 699, one tick before the downtime window, but whose event notification, delayed by the fixed 1-tick delivery lag, would have fired at tick 700, squarely inside the outage, so it too was only caught by the forced restart sweep at tick 900, a genuine and informative edge case: an outage's effective blast radius on the event path extends slightly earlier than its own start time, by however long event delivery normally takes.

Key points, complexity, and edge cases

Complexity: each poll-only or safety-net sweep is a single batched call costing $O(\lceil k / 1000 \rceil)$ paginated List requests, where $k$ is the number of objects currently under the scanned prefix, not $O(N)$ in the number of watched prefixes; this decoupling of API-call count from prefix count is the entire mechanism that avoids one sensor per prefix. The simulation loop itself is $O(T \times N)$ in the worst case, $T$ ticks times $N$ prefixes, purely as an artifact of simulating every tick explicitly in Python; a real deployment has no equivalent per-tick cost, only per-sweep cost.

Edge cases: a prefix whose object lands in the final simulated tick has no later tick in which its event delay or next sweep can fire, the 4 permanently-undetected prefixes above, an artifact of the simulation window ending, not a defect in the mechanism, disclosed explicitly rather than adjusted away. A prefix whose object lands one tick before downtime begins, but whose event's fixed delivery delay would push the actual notification into the downtime window, is still correctly caught, just later, by the forced restart sweep, demonstrated by the 201-tick worst-case event-path latency traced to exactly this case.

Trade-offs and pitfalls

Modeling event delivery as fully dropped during downtime, rather than durably queued (as a real SNS/SQS-backed pipeline usually is), was a deliberate choice to stress-test the safety net; a real deployment backed by a durable queue would lose far fewer, possibly zero, events during an outage of the consumer alone, but the safety-net sweep is still worth keeping, since it also protects against outages of the event pipeline itself, not only the consumer, and against silent misconfiguration where a new prefix was never wired to event notifications in the first place.

The safety-net sweep's fixed hourly cadence is a deliberate trade: making it faster would shrink worst-case latency for a dropped event but would erode the very cost advantage that justifies not just running the poll-only sweep against everything all the time.

The adaptive poll-only sweep's exponential backoff means a prefix that arrives just after the interval has grown large waits up to that full interval before being caught; this is the direct cost of adaptivity, and a system with a tighter latency requirement for poll-only prefixes specifically would need a lower MAX_INTERVAL cap, trading away some of the call-count savings for tighter worst-case latency.

Direct answer

Migrating 300 on-premises Airflow DAGs with custom operators and local-filesystem dependencies to a managed cloud service is fundamentally a dependency-inventory problem before it is a cutover problem. Every DAG's actual filesystem and custom-operator dependencies need to be mapped and reclassified for cloud object storage before any DAG moves, because "it worked on-premises" says nothing about whether it will work once local disk access and host-level assumptions are gone.

Structured elaboration

Inventory and dependency analysis. Programmatically scan all 300 DAGs for local filesystem access patterns, direct file opens, hardcoded paths, or an assumption that a temporary directory persists across tasks, which does not hold once workers are ephemeral and not guaranteed to share a host. Scan separately for custom operator and plugin usage, and for any host-level assumption, a system-wide package or a mounted network drive. Produce a per-DAG risk and complexity score from this scan, and let it drive the staging plan below, rather than guessing which DAGs are safe to move first.

Refactoring custom operators. Classify each one as compatible as-is, needs adaptation, or needs a full rewrite against the target managed service's actual constraints: a specific Airflow version, a specific set of pre-installed providers, and no host-level file system write access beyond what the managed service itself exposes.

Replacing local disk usage with cloud object storage. Any DAG reading or writing local files between tasks, a common on-premises pattern, silently breaks under a managed, ephemeral-worker execution model, since two tasks are never guaranteed to run on the same host. That pattern needs replacing with an object-storage-backed one: write to a bucket path, and pass only that path through cross-communication (XCom), not the file's own bytes. This is frequently the single most common silent breakage in this exact kind of migration, since it can pass a quick single-task smoke test on one host and still fail once tasks are genuinely distributed in production.

Secrets integration. On-premises secrets often live in a local Airflow connections database or in host-level environment variables. Migrate these to the target cloud's actual secrets-management integration, never as a lift-and-shift of plaintext values into the new platform's connection interface as a "temporary" step that then quietly never gets revisited.

Continuous integration and deployment (CI/CD) adjustments. The deployment pipeline needs to target the new platform's real deployment mechanism, typically an object-storage bucket sync for DAG files, not a file copy to a server. Any CI step that validated DAGs against the on-premises environment's specific installed packages needs to validate against the target environment's package set instead, catching a missing dependency before deployment rather than at first production run.

Validation. For each migrated DAG, require a defined equivalence test, a parallel run comparing on-premises and cloud output on the same input, before that DAG is considered migrated, not simply "it deployed without error."

Rollback strategy. Keep the on-premises Airflow instance running and authoritative for each DAG until its cloud equivalent has passed validation and run cleanly in production for a defined stabilization period. Rolling one DAG back means re-enabling its on-premises schedule and disabling the cloud one, which stays cheap and fast only for as long as the on-premises instance has not already been decommissioned.

Prioritizing and staging to reduce risk. Migrate DAGs with no custom operators and no local-disk dependencies first, the lowest-risk group, which also validates the migration pipeline and CI/CD mechanics themselves. Then DAGs needing adaptation but no local-disk refactor. Then DAGs needing both a local-disk refactor and a custom-operator rewrite last, since that group carries this migration's two genuinely riskiest dimensions at once.

Worked example

The inventory scan of 300 DAGs finds 165 with no custom operators and no local-disk dependencies (wave 1), 98 with custom operators needing adaptation but no local-disk dependencies (wave 2), and 37 with local-disk dependencies requiring the object-storage refactor, 12 of which also need a custom-operator rewrite (wave 3, the most complex group).

Wave 1's 165 DAGs migrate over 3 weeks. Equivalence testing during this wave catches 4 of the 165 with a subtle behavior difference, traced to a timezone-handling default that differs between the on-premises Airflow version and the managed service's version. Once fixed and re-validated for those 4, the same class of bug is proactively checked against the remaining 135 DAGs across waves 2 and 3, rather than being independently rediscovered dozens more times.

Trade-offs and pitfalls

Keeping the on-premises instance running for the entire migration means paying for both platforms simultaneously for the full duration, a real, ongoing cost that should be weighed against how much risk it is actually still reducing once later waves have stabilized; decommissioning earlier waves' on-premises twins progressively, rather than all at once at the very end, can meaningfully cut this cost without giving up the safety net where it still matters.

Treating the local-disk-to-object-storage refactor as a purely mechanical, one-time task can miss cases where the local-disk usage was hiding an implicit ordering assumption, a task that only worked because it happened to run on the same host as an earlier task that wrote a file it read. Object storage alone does not fix that; the underlying task dependency has to be made explicit too, or the same assumption can re-manifest as a race condition, one task reading before another task's object-storage write has actually finished propagating.