Waiting on an externally, asynchronously generated file before continuing is a common scenario in Airflow.
But “I’ll wait as long as it takes” is a heavy promise to make on your infrastructure’s behalf.

The Existing Problem

I’m currently maintaining a system integration pipeline scheduled through Airflow.

The main steps are:

1. Receive parameters
2. Call an external API to generate a file for exchange
3. Once generation completes, continue with the downstream process

The “wait for the file to finish generating” step has gone through two different approaches across projects.

Version 1

Call the external API directly and wait for it to respond only once the file is ready.

Generating the file takes a while, and the connection often got cut by an API timeout, failing the whole task outright.
The fix at the time was simply extending the timeout: treating the symptom, not the cause.

Version 2

The external service was changed to write the file’s generation-complete status into Redis as soon as it was ready.
The engineer at the time hand-rolled a sensor that waited for the Redis key to appear before moving on, sidestepping the timeout problem.

It looked like the timeout issue was solved, but running it in production surfaced a few new problems:

• When the ECS instance hosting a worker sits idle, it gets reclaimed automatically, even while a task is still waiting on it.
• The Redis key can expire; if the check doesn’t catch it in time, the status becomes unreadable, even though the file was actually generated successfully.

Worse, these two problems can become entangled:

Once the ECS instance is reclaimed, nobody’s left to keep checking, and the key just quietly expires.

That leaves the task in a Schrödinger’s-cat state: a failure that isn’t necessarily a real failure.

The current workaround is to extend the scale-in cooldown period and bump up the default worker count.
Still treating the symptom, not the cause.

The Solution

S3KeySensor + deferrable to the Rescue!

Waiting on an externally, asynchronously generated file before continuing is actually a very common scenario in Airflow, and the built-in S3KeySensor exists specifically to handle it.

It can directly detect whether a given key exists on S3, using the S3 object itself as the thing being watched, so there’s no Redis-key-expiration problem to worry about.

S3KeySensor is also a standard implementation maintained by an official Airflow provider. Airflow’s own documentation confirms that with deferrable=True, it doesn’t occupy a worker slot, so it’s also immune to workers being interrupted when an ECS instance gets reclaimed.

It supports three modes:

• mode="poke" (the default): the task occupies a worker slot until the condition is met.
• mode="reschedule": each check reschedules a worker to run it, releasing the worker slot between checks.
• deferrable=True (defaults to False): hands off “waiting” entirely to an independent triggerer process, without occupying a worker.

These three take precedence over one another, which you can see in the source of S3KeySensor.execute():

apache/airflow GitHub - providers/amazon/…/sensors/s3.py

def execute(self, context: Context) -> None:
    if not self.deferrable:
        super().execute(context)   # mode="poke"/"reschedule" only takes effect here
    else:
        if not self.poke(context=context):
            self._defer()          # hand off to the triggerer in the background, no worker occupied

When deferrable=True, it only pokes once: if the condition isn’t met, it’s handed straight to the triggerer, and whatever mode is set to no longer has any effect.
And mode only matters when deferrable=False.

Since this example uses deferrable=True, there’s no need (and no reason) to set mode at all.

Defaults for the other relevant parameters:

• poke_interval: defaults to 60 seconds, the interval between checks.
• timeout: defaults to 604800 seconds (7 days), after which the task is marked as failed if the condition still isn’t met.

Proof of Concept

I wrote a simple DAG to verify the whole mechanism:

import logging
from datetime import UTC, datetime, timedelta

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.sdk import DAG, Param, Variable, get_current_context, task
from airflow.sdk.definitions.param import ParamsDict
from services.aws.s3_client import S3Client

logger = logging.getLogger(__name__)

default_args = {
    "owner": "Developer",
    "depends_on_past": False,
    "start_date": datetime(2026, 9, 22, tzinfo=UTC),
    "retry_delay": timedelta(minutes=1),
}

with DAG(
    dag_id="s3_key_sensor_test",
    default_args=default_args,
    schedule=None,
    catchup=False,
    max_active_runs=1,
    dag_display_name="S3KeySensor Proof of Concept",
    tags=["AWS"],
    doc_md="""\
        ## Testing how `S3KeySensor` works

        1. `get_s3_path`: generates the S3 path to watch
        2. `wait_for_key`: waits for the S3 object at the path from `get_s3_path` to appear before continuing
        3. `fetch_value`: once detected, reads the content with the existing `S3Client` and returns its length (bytes)
        4. `print_value`: receives the return value from `fetch_value` and prints it
    """,
    params=ParamsDict(
        {
            "bucket_name": Param(
                default="",
                type="string",
                title="S3 Bucket",
                description="S3 bucket name",
            ),
            "bucket_key": Param(
                default="",
                type="string",
                title="Bucket Key",
                description="S3 object key",
            ),
        }
    ),
) as dag:

    @task(task_id="get_s3_path", multiple_outputs=True)
    def get_s3_path() -> dict[str, str]:
        """In production the path is computed by application logic; here it's taken directly from the params passed in when triggering the DAG, for testing convenience

        multiple_outputs=True stores each dict key as its own XCom entry, so downstream tasks can use subscript syntax like s3_path["bucket_name"] to read the value; otherwise it would come back as None

        Returns:
            dict[str, str]: each key maps to its own XCom entry
        """
        params = get_current_context().get("params", {})
        return {
            "bucket_name": params["bucket_name"],
            "bucket_key": params["bucket_key"],
        }

    s3_path = get_s3_path()

    wait_for_object = S3KeySensor(
        task_id="wait_for_object",
        bucket_name=s3_path["bucket_name"],  # type: ignore[index]
        bucket_key=s3_path["bucket_key"],  # type: ignore[index]
        deferrable=True,  # whether to hand off to the triggerer, without occupying a worker slot
        poke_interval=30,  # seconds between checks, >= 60 recommended in production
        timeout=3600,  # timeout in seconds
    )

    @task(task_id="fetch_object", retries=0)
    def fetch_object(bucket_name: str, bucket_key: str) -> int:
        content = S3Client(bucket_name=bucket_name).get_object(bucket_key)

        logger.info("[S3KeySensor] Detected file at S3 %s/%s", bucket_name, bucket_key)
        return len(content)

    @task(task_id="print_object_length", retries=0)
    def print_value(content_length: int) -> None:
        print(f"[S3KeySensor] Successfully retrieved {content_length} bytes")

    fetch_object_task = fetch_object(
        s3_path["bucket_name"],  # type: ignore[index]
        s3_path["bucket_key"],  # type: ignore[index]
    )

    wait_for_object.set_downstream(fetch_object_task)
    print_value(fetch_object_task)  # type: ignore[arg-type]

if __name__ == "__main__":
    dag.cli()

1. Get the S3 path

In production, the file’s upload path is generated by application logic according to a set of rules.
For testing convenience, this is replaced with manually entering bucket_name/bucket_key when triggering the DAG.

You can see bucket_name and bucket_key recorded as separate entries on the XCom tab, while return_value is the dict packaged together and passed out as a whole.

This is why multiple_outputs=True matters: it lets each key in the dict be stored as its own XCom entry, so downstream tasks can read values directly with subscript syntax like s3_path["bucket_name"].

Without it, XCom would only store a single return_value, and s3_path["bucket_name"] would actually come back as None instead of the expected string.

2. Wait for the S3 object with S3KeySensor

The deferrable sensor waits for a file to actually show up at the path produced by get_s3_path.

The audit log clearly shows the state transitions: running -> deferred -> running -> success.

The task switches into deferred almost immediately after starting, and spends roughly 7 minutes waiting, during which it occupies no worker at all.

3. Read the file content once detected

A file is uploaded manually to simulate the external API generating it.

Once S3KeySensor detects the file, it resumes from deferred and continues execution.

The log matches the actual file’s content length.

Caveats

Using deferrable mode requires the triggerer component to be enabled in the environment; without it, the task gets stuck in deferred and is never woken up.

This PoC doesn’t set aws_conn_id explicitly; it simply relies on the default boto3 credential chain from the MWAA execution role.

The “task fails when its worker gets reclaimed” problem isn’t 100% eliminated; it’s traded for a much lighter dependency.

During the deferred phase of execution, since work has been offloaded to the triggerer, the task no longer occupies a worker slot, and you have more free workload capacity.

The waiting logic now runs on an independent triggerer process, so it becomes dependent on that triggerer staying alive.
The good news is that the deferred state is persisted in the metadata DB, so even if the triggerer process restarts, the task isn’t immediately marked as failed, which is what gives it high availability.

Airflow automatically re-schedules triggers that were on that host to run elsewhere.

Once the condition is met and the trigger fires, a worker still has to be handed back to actually finish executing the task:

• The trigger runs until it fires, at which point its source task is re-scheduled by the scheduler.
• The scheduler queues the task to resume on a worker node.

In other words, using deferrable eliminates

• the risk of the worker process it depends on during the wait being reclaimed by ECS at any moment

and trades it for the much lower-risk

• dependency on the triggerer staying alive

  • the deferred task’s state lives in the metadata DB
  • even if one triggerer dies, the scheduler automatically reassigns its triggers to another still-living triggerer

• need to be handed a worker again once execution resumes

  • that worker is only occupied for the actual, short execution time

Reference: Deferrable Operators & Triggers - Apache Airflow official docs