Workflow Orchestration and Scheduling Questions

Direct answer

When a report has to publish now but some hours are missing, the right response is never to silently substitute a guess and publish as if the data were complete; it is to choose an explicit, clearly-labeled stand-in value for the missing hours (the last known-good comparable value, or an interpolation, whichever is more defensible for this specific metric), publish with a visible caveat naming exactly which hours are incomplete, and separately schedule the real fix (the actual backfill) so the report is silently corrected, and visibly re-confirmed as final, once it lands.

Structured elaboration

Short-term workaround options, and how to choose between them.

• Carry-forward (use the last known-good value for the missing hour). Appropriate when the metric is relatively stable hour to hour and a slightly stale value is a reasonable approximation, for example a slowly-changing inventory count.
• Interpolation between the surrounding known hours. Appropriate when the metric has a smooth intra-day trend (traffic ramping through the day) and a straight-line or comparable estimate between the hour before and after the gap is more accurate than simply repeating the last value.
• Explicit "pending" placeholder, with the report showing a partial total or a clearly marked gap rather than any estimated number. Appropriate when the metric is high-stakes enough (a financial figure, a compliance number) that ANY estimated value risks being mistaken for a real one, and an honest gap is safer than a plausible-looking guess.
• Exclude the affected hours from the reporting window entirely, showing a smaller but fully real window (for example, a 21-hour total instead of 24, clearly labeled), when the report's consumer would rather have a smaller, fully accurate number than a full-window number partly built on estimates.

The right choice depends on what the number is actually used for: a directional trend dashboard can tolerate carry-forward or interpolation with a caveat; a number feeding a financial or compliance decision should generally use the placeholder-or-exclude approach, since being wrong with confidence is worse than being visibly incomplete.

Making sure consumers understand the caveats. A workaround value with no visible flag defeats the entire point of choosing a careful, defensible estimate; the report itself needs an explicit, visible marker (a footnote, an asterisk on the affected hours, a banner stating "hours 14-16 are provisional, pending pipeline reprocessing, expected corrected by [time]") so anyone reading it knows exactly which numbers to treat with caution and roughly when a corrected version will be available. This should be as specific as possible: naming the affected hours and giving a real expected-resolution time is far more useful to a stakeholder than a generic "data may be incomplete" disclaimer that does not say which part.

Longer-term fix. Once the underlying DAG failure is diagnosed and resolved, backfill exactly the affected hours (not the whole day, to avoid unnecessary reprocessing of hours that were already correct), overwrite the report's provisional numbers with the real, recomputed values, and send a follow-up notice confirming the report is now final and correct, referencing the earlier caveat so anyone who saw the provisional version knows it has since been resolved. This closing step is often skipped once the immediate pressure is off, but it is what actually closes the loop for stakeholders who specifically remember seeing a caveat and are waiting to know whether it was ever resolved.

Worked example

An hourly sales-aggregate DAG failed 3 hours ago; hours 0-13 completed normally, hours 14-16 are missing, and hour 17 (the current hour) has not run yet. A daily summary report is due to publish now, needing a full 0-17 window.

Given this is a sales figure feeding an internal operations dashboard (directional, not a financial-close number), the team chooses interpolation for the 3 missing hours, using the average of hour 13's and hour 17's... except hour 17 has not run yet either, so interpolation between two known points is not actually available. Recognizing this, the workaround shifts: hours 14-16 use carry-forward from hour 13 (the last fully known value), and the report explicitly shows hour 17 as excluded from the total rather than estimated, since there is no surrounding data to interpolate from yet. The published report shows a clear banner: "hours 14-16 use a provisional carry-forward estimate pending pipeline reprocessing; hour 17 is not yet available and is excluded from this total; expected fully corrected by [end of day]."

Two hours later, the underlying DAG failure (a transient database connection issue) is fixed, and a backfill runs for hours 14, 15, 16, and the now-completed 17, producing the real, correct values. The report is republished with all hours final, and a follow-up notice states: "the earlier provisional figures for hours 14-16 have now been corrected with actual data; today's report is final."

Trade-offs and pitfalls

Choosing to interpolate or carry-forward without checking whether the surrounding hours are actually available (as almost happened in the worked example) is a real, easy mistake under time pressure: an interpolation computed from one real value and one still-missing value is not actually an interpolation, it is a guess dressed up as one, and shipping it without noticing the second input was also missing would have produced a specific, confidently-wrong number rather than an honestly-labeled gap.

Publishing a workaround value without a visible caveat, on the reasoning that "we'll fix it soon anyway so it doesn't matter," is the single most damaging mistake in this scenario: a stakeholder who acts on an unflagged, temporarily-wrong number before the fix lands has already made a decision based on it, and there is no way to retroactively un-make that decision the way there is a way to retroactively correct the report's numbers.

Forgetting the closing "this has now been corrected" notice leaves stakeholders who saw the caveat with no way to know whether the number they are now looking at is the corrected final version or still the provisional one, which quietly erodes trust in every future caveat the team publishes, since "we'll let you know when it's fixed" starts to feel like an empty promise if it is not consistently followed through.

Direct answer

A simple nightly batch ETL DAG needs five concerns beyond just the extract/transform/load tasks themselves: confirmation the upstream data actually exists before extracting, a retry policy and an SLA so a slow or failing run is caught, a guarantee that two nightly runs never overlap, a lightweight record of what ran and produced what, and an alert path when something fails. Below is a concrete design covering all five with specific operators and parameters, not just the task shape.

Structured elaboration

DAG structure and task types.

graph LR
    A[wait_for_upstream_file] --> B[extract]
    B --> C[transform]
    C --> D[load]
    D --> E[record_lineage]

• wait_for_upstream_file: an S3KeySensor (or the equivalent for the actual source system) in reschedule mode, checking for the arrival marker of the day's source data before anything downstream starts.
• extract: a PythonOperator using S3Hook.download_file() (Airflow's amazon provider ships no built-in "S3-to-local" transfer operator; the prebuilt transfer operators either run the opposite direction or target a specific system, e.g. LocalFilesystemToS3Operator, S3ToRedshiftOperator, or Snowflake's CopyFromExternalStageToSnowflakeOperator reading a preconfigured S3-backed external stage) that pulls the day's raw data into a staging location.
• transform: a PythonOperator running the actual business logic (or a BashOperator/KubernetesPodOperator invoking a dbt or Spark job if the transform is heavier than fits comfortably in the scheduler's own worker process).
• load: a PythonOperator or a SQL-execution operator (e.g. SQLExecuteQueryOperator) that writes the transformed output into its final partitioned table, using an upsert or overwrite-by-partition pattern so the load step is itself safely rerunnable.
• record_lineage: a small PythonOperator that writes a lineage/metadata record after a successful load (detailed below).

Sensors and dependencies for upstream data availability. The wait_for_upstream_file sensor is the first task, not an assumption baked into extract; this makes "the source wasn't ready yet" a distinct, visible failure state (a sensor timeout) rather than extract failing with a generic file-not-found error that gives an on-call engineer less signal. poke_interval=300 (check every 5 minutes) and timeout=10800 (give up after 3 hours) in reschedule mode, so the sensor does not hold a worker slot for the whole wait.

Retry and SLA settings. default_args={"retries": 2, "retry_delay": timedelta(minutes=5), "retry_exponential_backoff": True} on the transform and load tasks, since a database hiccup or a transient compute failure is worth a couple of automatic retries; the sensor's own timeout, not a retry count, governs how long it waits for genuinely missing upstream data. An sla=timedelta(hours=2) set on the DAG (or on the final task) fires Airflow's SLA-miss callback if the whole pipeline has not completed within 2 hours of its scheduled start, independent of whether any individual task actually failed, catching the "technically still running, just very slow" case that a pure failure alert would miss.

Preventing concurrent runs. max_active_runs=1 on the DAG definition, plus catchup=False so a redeploy or an unpause does not immediately queue a backlog of historical runs that would otherwise also compete for the same "only one at a time" slot. Without this, a slow night's run still executing at the next scheduled trigger time would start a second, overlapping run against the same target tables, risking a race on the load step.

Capturing lineage and metadata. The record_lineage task writes a row to a lineage table (or emits an event to a metadata system) after a successful load, containing: the DAG run's logical date, the source file(s) consumed (captured from the sensor/extract step via XCom), row counts in and out, and a completion timestamp. This does not need a heavyweight lineage system to be useful; even a simple append-only table (dag_id, logical_date, source_files, rows_extracted, rows_loaded, completed_at) answers "what ran, on what input, producing what" for any later investigation, without requiring someone to dig through task logs.

Alerting on failures. on_failure_callback set on the DAG (or per task, if different tasks should notify different owners) posts to a Slack channel with the DAG id, the failing task, the logical date, and a link to the log; a separate sla_miss_callback handles the SLA-breach case distinctly from a hard failure, since "still running, just late" and "failed outright" call for different responses. Both route to the owning team's on-call channel rather than a shared generic alert firehose, so the right people see it without having to filter.

Worked example

Concretely, for a DAG scheduled @daily with start_date fixed at a past date:

• 00:00: scheduled trigger fires for the logical date that just completed. wait_for_upstream_file starts polling.
• 00:00-00:35: the sensor polls every 5 minutes starting at trigger time (00:00, 00:05, ..., 00:30, 00:35). The upstream marker file lands at 00:32, after the 00:30 poke (the 7th check) but before the 00:35 poke (the 8th check), so the sensor detects it and succeeds at 00:35.
• 00:35-00:41: extract runs (6 minutes), pulling the day's raw data.
• 00:41-00:52: transform runs (11 minutes); on this run it hits a transient database connection error at 00:44, fails, and the retry policy (2 retries, 5-minute exponential backoff) retries at 00:49, succeeding, finishing at 00:52 total elapsed for the task including the one retry.
• 00:52-00:58: load runs (6 minutes), upserting into the final partitioned table.
• 00:58-00:59: record_lineage writes the metadata row.
• Total elapsed from scheduled trigger to completion: 59 minutes, comfortably inside the 2-hour SLA, so no SLA-miss alert fires. Had the upstream file instead not landed until 03:10 (after the sensor's 3-hour, 10800-second timeout expires at 03:00 from the 00:00 start), the sensor would time out first and fail with a clear timeout reason, on_failure_callback would post to Slack naming wait_for_upstream_file as the failing task, and the SLA-miss callback would separately fire once the 2-hour mark passed, giving on-call two distinct, correctly-labeled signals rather than one ambiguous one.

Trade-offs and pitfalls

Skipping the dedicated upstream-availability sensor and letting extract fail instead is a common shortcut that trades away diagnostic clarity: a sensor timeout says "the source wasn't ready," while a bare extract failure could mean that or a dozen other things, and an on-call engineer has to read logs to tell the difference every time instead of once, in the design.

Omitting max_active_runs=1 is an easy mistake that only shows up under load: the DAG behaves correctly every night it finishes quickly, and only fails the night it runs slow enough to still be executing at the next trigger, which makes the bug intermittent and harder to attribute to the missing setting when it finally happens.

Treating lineage capture as optional "nice to have" and skipping it is a common corner cut for a "simple" DAG, but it is exactly the simple DAGs that tend to accumulate undocumented tribal knowledge about what each run actually processed; a minimal lineage row costs one extra task and pays for itself the first time someone needs to answer "was table X built from the corrected or the buggy version of yesterday's data."

Finally, routing both hard failures and SLA misses to the same alert with the same urgency conflates two different situations: a hard failure needs someone now, while an SLA miss on a pipeline that is still running might resolve itself in ten more minutes; treating them identically either causes alert fatigue on SLA misses or under-reacts to real failures, depending on which way the team calibrates.

Direct answer

The routine below runs dbt, materializes to Snowflake, gates success behind data quality (DQ) checks, and posts to Slack, structured so retries only apply to the dbt step (the step actually prone to transient failure), idempotency comes from keying every run by its logical date and skipping already-succeeded work rather than blindly re-executing, and a per-run metadata record (models run, rows materialized, DQ pass/fail counts, retry attempts, terminal status) is captured for every run regardless of outcome, which is what a dashboard or an audit actually needs.

Structured elaboration

Retry logic. Only the dbt run step is wrapped in retry logic, via a small decorator that retries on a distinguished RetryableError up to a fixed attempt count with exponential backoff, and re-raises after the last attempt so the caller can record a genuine failure rather than retrying forever. Materialization and DQ checks are not retried the same way: a Snowflake write failure or a DQ check failure is a different kind of problem (data-correctness or infrastructure-permission related, not "try again in a moment"), and retrying those blindly risks masking a real, persistent issue behind an automatic retry that happens to eventually stop failing for the wrong reason.

Idempotency considerations. The run log is keyed by logical_date, and the orchestration function's very first check is whether a run for that date already succeeded; if so, it returns the existing record immediately without calling dbt, without writing to Snowflake again, and without re-running DQ checks. This is what makes re-triggering the same logical date, whether from an orchestrator-level retry, a manual rerun, or an accidental double-trigger, safe by construction rather than something that has to be reasoned about separately for every call site. The Snowflake materialization step is itself also idempotent independent of the run-log check (it overwrites the row count keyed by logical_date rather than appending), which is a deliberate second layer of protection: even if the run-log check were somehow bypassed, the underlying write would still converge to the same state rather than duplicating data.

Exposing run metadata for dashboarding and audits. Every run, whether it ends in success, dq_failed, or failed, produces a RunRecord with the same shape: logical date, status, dbt models run, rows materialized, DQ checks passed and failed, and the number of attempts each retried step took. Capturing this uniformly, not just on success, is what makes the resulting log usable for a dashboard (which needs to show failed and DQ-failed runs, not just successful ones) and for an audit (which needs to answer "how many attempts did this specific date take, and did it ever fail DQ before eventually succeeding," not just "did it eventually succeed").

Worked example

import time
from dataclasses import dataclass, field


class RetryableError(Exception):
    pass


class DataQualityError(Exception):
    pass


def with_retries(max_attempts=3, base_delay=0.0, backoff=2):
    """Decorator: retry only on RetryableError, exponential backoff, re-raise after
    exhausting attempts. base_delay=0.0 in this harness so the demo runs instantly;
    in production this would be a real delay (e.g. base_delay=30)."""
    def decorator(fn):
        def wrapper(*args, **kwargs):
            delay = base_delay
            last_exc = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except RetryableError as e:
                    last_exc = e
                    if attempt == max_attempts:
                        raise
                    time.sleep(delay)
                    delay *= backoff
            raise last_exc  # pragma: no cover
        return wrapper
    return decorator


@dataclass
class RunRecord:
    """Metadata exposed for dashboarding and audits: one row per logical run."""
    logical_date: str
    status: str = "pending"
    dbt_models_run: int = 0
    rows_materialized: int = 0
    dq_checks_passed: int = 0
    dq_checks_failed: int = 0
    attempts: dict = field(default_factory=dict)  # step_name -> attempt count
    error: str = None


class FakeDbtRunner:
    """Stand-in for `dbt run`. fail_until controls how many calls fail before succeeding,
    to exercise the retry path deterministically."""
    def __init__(self, fail_until=0):
        self.fail_until = fail_until
        self.calls = 0

    def run(self):
        self.calls += 1
        if self.calls <= self.fail_until:
            raise RetryableError(f"dbt run transient failure (attempt {self.calls})")
        return {"models_run": 12}


class FakeSnowflake:
    """Idempotent materialization: keyed by logical_date, overwrite not append."""
    def __init__(self):
        self.tables = {}  # logical_date -> row_count
        self.write_count = {}  # logical_date -> number of times written

    def materialize(self, logical_date, row_count):
        self.tables[logical_date] = row_count  # overwrite
        self.write_count[logical_date] = self.write_count.get(logical_date, 0) + 1


class FakeDQChecker:
    def __init__(self, force_fail=False):
        self.force_fail = force_fail

    def run_checks(self, row_count):
        checks = [
            ("row_count_positive", row_count > 0),
            ("row_count_within_bounds", 100 <= row_count <= 100000),
            ("no_null_keys", not self.force_fail),
        ]
        passed = sum(1 for _, ok in checks if ok)
        failed = [name for name, ok in checks if not ok]
        return passed, failed


class FakeSlack:
    def __init__(self):
        self.messages = []

    def post(self, channel, text):
        self.messages.append((channel, text))


@with_retries(max_attempts=3, base_delay=0.0)
def run_dbt(dbt_runner):
    return dbt_runner.run()


def orchestrate_nightly_run(logical_date, dbt_runner, snowflake, dq_checker, slack, run_log):
    """
    Idempotency: run_log is keyed by logical_date. If a run for this logical_date already
    SUCCEEDED, this function returns that existing record without redoing any work, so
    re-triggering the same logical date (a retry at the orchestrator level, or a manual
    rerun) is safe and cheap rather than repeating dbt + Snowflake + DQ from scratch.
    """
    existing = run_log.get(logical_date)
    if existing is not None and existing.status == "success":
        return existing

    record = RunRecord(logical_date=logical_date)
    run_log[logical_date] = record

    try:
        # Step 1: dbt run, with retry on transient failure.
        dbt_result = run_dbt(dbt_runner)
        record.attempts["dbt_run"] = dbt_runner.calls
        record.dbt_models_run = dbt_result["models_run"]

        # Step 2: materialize to Snowflake, idempotent (overwrite keyed by logical_date).
        row_count = 5000  # in reality: computed from the dbt run's own output
        snowflake.materialize(logical_date, row_count)
        record.rows_materialized = row_count

        # Step 3: data quality checks BEFORE declaring success.
        passed, failed = dq_checker.run_checks(row_count)
        record.dq_checks_passed = passed
        record.dq_checks_failed = len(failed)

        if failed:
            record.status = "dq_failed"
            record.error = f"failed checks: {failed}"
            slack.post("#data-alerts", f"FAILED: {logical_date} dbt run completed but "
                       f"{len(failed)} data quality check(s) failed: {failed}")
            return record

        record.status = "success"
        slack.post("#data-alerts", f"SUCCESS: {logical_date} nightly run complete, "
                   f"{record.dbt_models_run} models, {record.rows_materialized} rows, "
                   f"{record.dq_checks_passed} DQ checks passed")
        return record

    except RetryableError as e:
        record.status = "failed"
        record.error = str(e)
        record.attempts["dbt_run"] = dbt_runner.calls
        slack.post("#data-alerts", f"FAILED: {logical_date} dbt run failed after "
                   f"{dbt_runner.calls} attempts: {e}")
        return record


print("=== Case 1: transient dbt failure recovers via retry, then run succeeds ===")
run_log = {}
dbt = FakeDbtRunner(fail_until=2)   # fails on attempt 1 and 2, succeeds on attempt 3
sf = FakeSnowflake()
dq = FakeDQChecker(force_fail=False)
slack = FakeSlack()

r1 = orchestrate_nightly_run("2026-07-29", dbt, sf, dq, slack, run_log)
print("status:", r1.status, "| dbt attempts:", r1.attempts, "| rows:", r1.rows_materialized,
      "| dq passed/failed:", r1.dq_checks_passed, r1.dq_checks_failed)
assert r1.status == "success"
assert r1.attempts["dbt_run"] == 3
assert dbt.calls == 3
print("PASS: recovered after 2 failed attempts, 3rd succeeded")

print()
print("=== Case 2: idempotency - re-triggering the SAME logical_date after success is a no-op ===")
dbt_calls_before = dbt.calls
sf_writes_before = sf.write_count["2026-07-29"]
r1_again = orchestrate_nightly_run("2026-07-29", dbt, sf, dq, slack, run_log)
print("status:", r1_again.status, "| dbt.calls unchanged:", dbt.calls == dbt_calls_before,
      "| snowflake write_count unchanged:", sf.write_count["2026-07-29"] == sf_writes_before)
assert r1_again is r1, "re-triggering a successful logical_date must return the SAME record, not redo work"
assert dbt.calls == dbt_calls_before, "dbt must NOT be called again for an already-succeeded logical_date"
print("PASS: no duplicate dbt run, no duplicate Snowflake write, on re-trigger of a succeeded run")

print()
print("=== Case 3: exhausted retries (3 failures, all attempts fail) -> failed status, Slack alert ===")
run_log2 = {}
dbt2 = FakeDbtRunner(fail_until=5)   # fails every time within the 3-attempt budget
r2 = orchestrate_nightly_run("2026-07-30", dbt2, sf, dq, FakeSlack(), run_log2)
print("status:", r2.status, "| error:", r2.error, "| dbt attempts:", r2.attempts)
assert r2.status == "failed"
assert dbt2.calls == 3, "must stop at max_attempts, not retry forever"
print("PASS: gives up after exactly 3 attempts and records failed status")

print()
print("=== Case 4: dbt succeeds but a DQ check fails -> dq_failed status, distinct from a run failure ===")
run_log3 = {}
dbt3 = FakeDbtRunner(fail_until=0)
dq_failing = FakeDQChecker(force_fail=True)
slack3 = FakeSlack()
r3 = orchestrate_nightly_run("2026-07-31", dbt3, sf, dq_failing, slack3, run_log3)
print("status:", r3.status, "| dq passed/failed:", r3.dq_checks_passed, r3.dq_checks_failed,
      "| slack messages:", slack3.messages)
assert r3.status == "dq_failed"
assert r3.dq_checks_failed == 1
assert any("FAILED" in m[1] and "quality" in m[1] for m in slack3.messages)
print("PASS: DQ failure is a distinct terminal state from a dbt run failure, and alerts correctly")

print()
print("=== Metadata exposed for dashboarding/audits (the full run_log) ===")
for date_key, rec in {**run_log, **run_log2, **run_log3}.items():
    print(f"  {date_key}: status={rec.status} models={rec.dbt_models_run} "
          f"rows={rec.rows_materialized} dq_pass={rec.dq_checks_passed} dq_fail={rec.dq_checks_failed} "
          f"attempts={rec.attempts}")

print()
print("ALL CASES PASSED")

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

=== Case 1: transient dbt failure recovers via retry, then run succeeds ===
status: success | dbt attempts: {'dbt_run': 3} | rows: 5000 | dq passed/failed: 3 0
PASS: recovered after 2 failed attempts, 3rd succeeded

=== Case 2: idempotency - re-triggering the SAME logical_date after success is a no-op ===
status: success | dbt.calls unchanged: True | snowflake write_count unchanged: True
PASS: no duplicate dbt run, no duplicate Snowflake write, on re-trigger of a succeeded run

=== Case 3: exhausted retries (3 failures, all attempts fail) -> failed status, Slack alert ===
status: failed | error: dbt run transient failure (attempt 3) | dbt attempts: {'dbt_run': 3}
PASS: gives up after exactly 3 attempts and records failed status

=== Case 4: dbt succeeds but a DQ check fails -> dq_failed status, distinct from a run failure ===
status: dq_failed | dq passed/failed: 2 1 | slack messages: [('#data-alerts', "FAILED: 2026-07-31 dbt run completed but 1 data quality check(s) failed: ['no_null_keys']")]
PASS: DQ failure is a distinct terminal state from a dbt run failure, and alerts correctly

=== Metadata exposed for dashboarding/audits (the full run_log) ===
  2026-07-29: status=success models=12 rows=5000 dq_pass=3 dq_fail=0 attempts={'dbt_run': 3}
  2026-07-30: status=failed models=0 rows=0 dq_pass=0 dq_fail=0 attempts={'dbt_run': 3}
  2026-07-31: status=dq_failed models=12 rows=5000 dq_pass=2 dq_fail=1 attempts={'dbt_run': 1}

ALL CASES PASSED

Fake clients stand in for dbt, Snowflake, and Slack so the orchestration logic itself, not any real external system, is what is being exercised. Case 1 proves the retry path actually recovers from real failures rather than merely being present in the code: FakeDbtRunner(fail_until=2) genuinely raises RetryableError on its first two calls, and the run only succeeds because the decorator retried a third time. Case 2 is the idempotency proof: re-triggering the same logical_date after success leaves dbt.calls and Snowflake's write_count for that date completely unchanged, confirmed by direct before/after comparison, not just by the function returning without error. Case 3 confirms retries are bounded (exactly 3 attempts, not infinite) and produce a distinct failed status. Case 4 confirms a DQ failure after a successful dbt run is its own terminal state (dq_failed), separately alertable from an outright pipeline failure, which matters because the two call for different responses (a DQ failure often means the data itself needs review, not that the pipeline is broken).

Key points, complexity, and edge cases

Key points: retries are scoped to the one step that is actually transient (dbt run), not applied blanket across every step; the idempotency check happens before any work is attempted, not as a cleanup step after; every terminal state produces a metadata record with the same fields, which is what makes the log queryable uniformly.

Complexity: each orchestrated run is $O(1)$ in the number of steps (a fixed, small pipeline: dbt, materialize, DQ check), with the retry decorator adding at most $O(\text{max\_attempts})$ calls to the wrapped step in the worst case; none of this scales with data volume in the orchestration layer itself, since the actual row-level work happens inside dbt and Snowflake, outside what this routine measures or controls.

Edge cases: a logical_date that previously ended in failed or dq_failed (not success) is correctly allowed to be retried from scratch on a fresh trigger, since the idempotency short-circuit only checks for status == "success"; this is deliberate, matching the design principle that only genuinely completed work should be skipped. A DQ check that fails after dbt has already succeeded still leaves the Snowflake write in place (the row_count is materialized before the DQ gate runs in this design) rather than rolling it back, which is a real, stated trade-off discussed below, not an oversight.

Trade-offs and pitfalls

Materializing to Snowflake before running DQ checks (rather than checking a staging copy first and only materializing if checks pass) means a DQ failure leaves incorrect or suspect data live in the destination table, visible to anyone querying it, until a human resolves the alert; the alternative, checking a staging table first, avoids this exposure window but adds a second write step and doubles the storage briefly, a real cost worth incurring specifically for tables with sensitive downstream consumers, and worth skipping for lower-stakes tables where the added complexity is not justified.

Retrying only the dbt step and treating a Snowflake write failure as immediately terminal is a deliberate scope choice, not an oversight: a Snowflake outage is usually either very short (in which case the next scheduled run naturally recovers) or a genuine infrastructure incident that a blind retry would not fix any better than a human noticing the alert and escalating, so the added retry-wrapping complexity was judged not worth it for that step specifically, though a team with a different observed failure profile for their Snowflake connectivity might reasonably choose otherwise.

The in-memory run_log dict in this demo would be a real, persistent, queryable table in production (exactly what "expose run metadata for dashboarding and audits" requires); using a plain dict here keeps the executable example self-contained and fast, but shipping this pattern as-is, without persisting RunRecord somewhere durable, would lose all audit history the moment the orchestrator process restarted, defeating the entire purpose of capturing it.

Direct answer

The decorator below takes max_retries, initial_delay, max_delay, exponential_factor, and jitter exactly as required, plus a retryable_exceptions tuple that is what makes it not swallow unexpected exceptions: only exception types explicitly named there are caught and retried, anything else propagates on the first attempt. Thread safety comes from where the retry state lives: every mutable value (the current delay, the attempt count) is a local variable created fresh inside the wrapper function on each call, never stored on the decorator or the wrapped function itself, so concurrent calls from different threads never share or race on the same state, which the concurrent-callers test below demonstrates directly rather than just asserting.

Structured elaboration

Function signature. retry_with_backoff(max_retries=3, initial_delay=1.0, max_delay=30.0, exponential_factor=2.0, jitter=False, retryable_exceptions=(Exception,), sleep_fn=time.sleep), a decorator factory returning the actual decorator. max_retries is total attempts including the first (so max_retries=3 means the function runs at most 3 times total, not 1 plus 3 retries), chosen to keep "how many times will this actually execute" unambiguous at the call site rather than needing the reader to remember whether the count includes the first attempt. sleep_fn is an injection point defaulting to time.sleep, present specifically so the delay logic can be tested without a test suite actually waiting out real backoff delays.

Not swallowing unexpected exceptions. The except retryable_exceptions: clause only ever catches exception types explicitly listed there; the default (Exception,) is permissive for convenience, but a caller integrating this into orchestrated task code should narrow it to the specific transient failure types worth retrying (a connection timeout, a rate-limit response), so a genuine bug in the task's own logic (a TypeError, a KeyError from malformed input) propagates immediately rather than being silently retried three times and only then surfacing, delaying the real signal.

Thread safety. delay and attempt are declared inside wrapper(), meaning Python creates a fresh, independent set of these variables on every single call, on whatever thread makes that call. There is no class instance, no module-level counter, no mutable default argument holding retry state across calls; two threads invoking the same decorated function at the same moment each get their own private delay/attempt pair on their own stack, with nothing to race on. This is verified directly, not assumed, by Case 5 below.

Integrating with a scheduler's native retry mechanism. An orchestrator's own retry policy (retries, retry_delay in Airflow's default_args) operates at the level of the whole task instance: a failed task is re-queued and re-executed from the top, which means anything the task already did before failing (partial writes, partially-consumed input) needs to be safe to redo: this is the idempotency requirement that governs safe task retries. This decorator operates at a finer grain, inside a single task attempt, retrying one specific operation (an API call, a database write) without re-running the whole task. The two compose rather than compete: use this decorator around the specific operation known to fail transiently and quickly (so the task-level retry budget is not spent on failures this decorator would have absorbed on its own), and keep the orchestrator's own retry as the outer safety net for failures this decorator's max_retries genuinely exhausts, or for failures entirely outside this decorator's scope (the worker process itself crashing, which no in-process decorator can catch). Setting this decorator's own retry budget smaller than the orchestrator's task-level patience (so it fails fast into the orchestrator's retry rather than silently consuming most of the task's allotted runtime on its own) is the practical rule of thumb for how the two should be sized relative to each other.

Worked example

import random
import threading
import time
from functools import wraps


def retry_with_backoff(
    max_retries=3,
    initial_delay=1.0,
    max_delay=30.0,
    exponential_factor=2.0,
    jitter=False,
    retryable_exceptions=(Exception,),
    sleep_fn=time.sleep,
):
    """
    Decorator factory. Parameters:
      max_retries: total attempts allowed (including the first), not "retries after
                   the first", to keep the semantics of "how many times will this
                   function actually run" unambiguous at the call site.
      initial_delay: seconds before the first retry.
      max_delay: hard cap on any single delay, regardless of how many attempts have
                 elapsed.
      exponential_factor: delay multiplier applied after each failed attempt.
      jitter: if True, the actual sleep is a uniform random value in [0, computed_delay]
              rather than the computed delay itself, to avoid many callers retrying in
              lockstep against the same downstream system.
      retryable_exceptions: a tuple of exception types that trigger a retry; anything
              NOT in this tuple propagates immediately, unretried. THIS is the
              "does not swallow unexpected exceptions" requirement: only exceptions the
              caller has explicitly named as retryable are ever caught here.
      sleep_fn: injectable in place of time.sleep, purely so this can be unit-tested
              without a real test suite run taking as long as the backoff schedule itself.

    Thread safety: all mutable retry state (current delay, attempt count) lives in
    local variables inside wrapper(), created fresh on every call. Two threads calling
    the SAME decorated function concurrently each get their own independent local
    state; nothing here is stored on the decorator, the wrapped function, or any
    module-level variable, so there is no shared mutable state for two threads to race
    on. This is the actual mechanism, not a claim: it is what the concurrent-callers
    test below exercises directly.
    """
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            delay = initial_delay
            attempt = 0
            while True:
                attempt += 1
                try:
                    return fn(*args, **kwargs)
                except retryable_exceptions:
                    if attempt >= max_retries:
                        raise
                    sleep_time = min(delay, max_delay)
                    if jitter:
                        sleep_time = random.uniform(0, sleep_time)
                    sleep_fn(sleep_time)
                    delay = min(delay * exponential_factor, max_delay)
        return wrapper
    return decorator


class RetryableError(Exception):
    pass


class FatalError(Exception):
    pass


print("=== Case 1: retries on a retryable failure, then succeeds; delay sequence recorded ===")
recorded_delays = []


def fake_sleep(seconds):
    recorded_delays.append(round(seconds, 4))


call_count = {"n": 0}


@retry_with_backoff(max_retries=5, initial_delay=1.0, max_delay=10.0, exponential_factor=2.0,
                     jitter=False, retryable_exceptions=(RetryableError,), sleep_fn=fake_sleep)
def flaky(fail_times):
    call_count["n"] += 1
    if call_count["n"] <= fail_times:
        raise RetryableError(f"transient failure #{call_count['n']}")
    return "ok"


result = flaky(3)
print("result:", result)
print("total calls:", call_count["n"])
print("recorded delays:", recorded_delays)
assert result == "ok"
assert call_count["n"] == 4  # 3 failures + 1 success
assert recorded_delays == [1.0, 2.0, 4.0], f"expected exponential 1,2,4 (capped at 10), got {recorded_delays}"
print("PASS: recovered after 3 failures, delays doubled correctly (1.0 -> 2.0 -> 4.0)")

print()
print("=== Case 2: delay is capped at max_delay, never exceeds it ===")
recorded_delays2 = []
call_count2 = {"n": 0}


@retry_with_backoff(max_retries=6, initial_delay=1.0, max_delay=3.0, exponential_factor=2.0,
                     jitter=False, retryable_exceptions=(RetryableError,), sleep_fn=lambda s: recorded_delays2.append(round(s, 4)))
def flaky2(fail_times):
    call_count2["n"] += 1
    if call_count2["n"] <= fail_times:
        raise RetryableError("transient")
    return "ok"


result2 = flaky2(5)
print("recorded delays:", recorded_delays2)
assert result2 == "ok"
assert max(recorded_delays2) <= 3.0, "no delay should ever exceed max_delay"
assert recorded_delays2 == [1.0, 2.0, 3.0, 3.0, 3.0], f"expected 1,2,then capped at 3.0 repeatedly, got {recorded_delays2}"
print("PASS: delay sequence is 1.0, 2.0, then capped at max_delay=3.0 for all subsequent retries")

print()
print("=== Case 3: exhausts retries and re-raises the ORIGINAL exception type, not a generic one ===")
call_count3 = {"n": 0}


@retry_with_backoff(max_retries=3, initial_delay=0.01, max_delay=0.02, exponential_factor=2.0,
                     retryable_exceptions=(RetryableError,), sleep_fn=lambda s: None)
def always_fails():
    call_count3["n"] += 1
    raise RetryableError(f"attempt {call_count3['n']} failed")


try:
    always_fails()
    print("FAIL: no exception raised")
except RetryableError as e:
    print("re-raised:", e, "| total calls:", call_count3["n"])
    assert call_count3["n"] == 3, "must stop at exactly max_retries calls, not retry forever"
    print("PASS: exhausted retries at exactly max_retries, re-raised the original RetryableError")

print()
print("=== Case 4: does NOT swallow an exception outside retryable_exceptions ===")
call_count4 = {"n": 0}


@retry_with_backoff(max_retries=5, initial_delay=0.01, retryable_exceptions=(RetryableError,), sleep_fn=lambda s: None)
def fails_with_unexpected_error():
    call_count4["n"] += 1
    raise FatalError("this is NOT in retryable_exceptions and must propagate immediately")


try:
    fails_with_unexpected_error()
    print("FAIL: FatalError was swallowed")
except FatalError as e:
    print("FatalError propagated immediately:", e, "| total calls:", call_count4["n"])
    assert call_count4["n"] == 1, "an unexpected (non-retryable) exception must NOT be retried at all"
    print("PASS: FatalError was not caught, not retried, and not swallowed; it propagated on the first attempt")

print()
print("=== Case 5: thread safety - many concurrent callers each get independent retry state ===")
results = {}
lock_for_results = threading.Lock()


def worker(thread_id, fail_times):
    local_calls = {"n": 0}

    @retry_with_backoff(max_retries=10, initial_delay=0.001, max_delay=0.002, exponential_factor=2.0,
                         retryable_exceptions=(RetryableError,), sleep_fn=lambda s: None)
    def per_thread_flaky():
        local_calls["n"] += 1
        if local_calls["n"] <= fail_times:
            raise RetryableError("transient")
        return local_calls["n"]

    outcome = per_thread_flaky()
    with lock_for_results:
        results[thread_id] = outcome


threads = [threading.Thread(target=worker, args=(i, i % 4)) for i in range(20)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print("results from 20 concurrent threads (thread_id -> total attempts before success):")
print({k: results[k] for k in sorted(results)})
expected = {i: (i % 4) + 1 for i in range(20)}
assert results == expected, "each thread's own fail_times must independently determine its own attempt count"
print("PASS: all 20 concurrent threads completed with exactly the attempt count their own fail_times implies, "
      "confirming no shared/racing state between concurrent calls")

print()
print("ALL CASES PASSED")

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

=== Case 1: retries on a retryable failure, then succeeds; delay sequence recorded ===
result: ok
total calls: 4
recorded delays: [1.0, 2.0, 4.0]
PASS: recovered after 3 failures, delays doubled correctly (1.0 -> 2.0 -> 4.0)

=== Case 2: delay is capped at max_delay, never exceeds it ===
recorded delays: [1.0, 2.0, 3.0, 3.0, 3.0]
PASS: delay sequence is 1.0, 2.0, then capped at max_delay=3.0 for all subsequent retries

=== Case 3: exhausts retries and re-raises the ORIGINAL exception type, not a generic one ===
re-raised: attempt 3 failed | total calls: 3
PASS: exhausted retries at exactly max_retries, re-raised the original RetryableError

=== Case 4: does NOT swallow an exception outside retryable_exceptions ===
FatalError propagated immediately: this is NOT in retryable_exceptions and must propagate immediately | total calls: 1
PASS: FatalError was not caught, not retried, and not swallowed; it propagated on the first attempt

=== Case 5: thread safety - many concurrent callers each get independent retry state ===
results from 20 concurrent threads (thread_id -> total attempts before success):
{0: 1, 1: 2, 2: 3, 3: 4, 4: 1, 5: 2, 6: 3, 7: 4, 8: 1, 9: 2, 10: 3, 11: 4, 12: 1, 13: 2, 14: 3, 15: 4, 16: 1, 17: 2, 18: 3, 19: 4}
PASS: all 20 concurrent threads completed with exactly the attempt count their own fail_times implies, confirming no shared/racing state between concurrent calls

ALL CASES PASSED

Case 1 confirms exponential growth without jitter: 1.0 -> 2.0 -> 4.0, exactly initial_delay * exponential_factor^n. Case 2 confirms the cap: the sequence 1.0, 2.0, 3.0, 3.0, 3.0 shows growth up to max_delay=3.0 and then correctly staying capped rather than continuing to grow. Case 3 confirms the exhausted-retries path re-raises the specific original exception (RetryableError, with its original message), not a wrapped or generic one, after exactly max_retries calls. Case 4 is the "does not swallow unexpected exceptions" proof: FatalError, deliberately not included in retryable_exceptions, propagates on the very first call, with call_count4["n"] == 1 confirming zero retry attempts were made on it. Case 5 is the thread-safety proof: 20 threads, each configured to fail a different number of times (i % 4, cycling 0 through 3), all run concurrently, and every single thread's final attempt count exactly matches what its own, independent fail_times implies, with no cross-contamination between threads despite all 20 sharing the same decorator function and running at the same time.

Key points, complexity, and edge cases

Key points: retryable_exceptions is the mechanism, not a comment, behind "must not swallow unexpected exceptions"; local (per-call) state, not shared state, is the mechanism behind thread safety; sleep_fn injection is what makes the delay logic testable in milliseconds instead of requiring a test suite to actually wait out real backoff delays.

Complexity: $O(\text{max\_retries})$ calls to the wrapped function in the worst case; the decorator itself adds $O(1)$ overhead per attempt beyond the wrapped function's own cost.

Edge cases: max_retries=1 means the function is called exactly once with no retry at all, correctly falling through to raise on the first failure since attempt >= max_retries is true immediately. A jitter=True configuration is not separately asserted on exact values in these cases (since it is intentionally random), but the non-jitter cases directly validate the underlying deterministic delay math jitter is layered on top of.

Trade-offs and pitfalls

Defaulting retryable_exceptions to (Exception,) is convenient for a quick decorator application but defeats the entire purpose of the "does not swallow unexpected exceptions" requirement if a caller never narrows it; the decorator's design puts the responsibility on the call site to pass a specific tuple, which is correct API design, but a team adopting this pattern should treat the default as a placeholder to override, not a safe default to leave in production task code.

Storing delay as a mutable closure variable that gets reassigned each iteration works correctly here specifically because it lives inside wrapper()'s own call frame; a naive refactor that hoisted delay up to be a decorator-level or class-level variable (to "avoid recomputing it") would silently reintroduce a shared-state race between concurrent callers, which is exactly the mistake this design's structure avoids by construction, not by discipline.

Finally, layering this decorator inside an orchestrator-managed task without sizing its total worst-case delay (max_retries attempts each up to max_delay) against the orchestrator's own task timeout can produce a task that looks hung from the orchestrator's perspective while this decorator is still legitimately working through its own backoff schedule; the two retry layers need to be sized with awareness of each other, not configured independently.

Direct answer

In a modern ELT (extract, load, transform) stack, these three tools are not competitors, they occupy three different layers: a managed extract service (Fivetran or Stitch) handles getting raw data out of SaaS sources and into the warehouse, dbt handles transforming that raw data into business-ready models once it is there, and Airflow (or another orchestrator) is the layer that sequences and schedules both of those, plus anything else in the pipeline that is not itself extract or transform. The comparison is not "which one to use," it is "which layer does each one own."

Structured elaboration

Managed extract services (Fivetran, Stitch): the extract-and-load layer. These tools own connecting to SaaS sources (Salesforce, Stripe, HubSpot, and similar), handling each source's specific API quirks, authentication, pagination, and rate limits, and landing raw data into the warehouse on their own managed schedule, with schema drift handling built in. The value proposition is specifically not having to build and maintain dozens of source-specific connectors in-house; a small data team can stand up a dozen SaaS integrations in days rather than months. What they do not do: any real business transformation logic; their job ends at "raw data, faithfully landed."

dbt: the transformation layer. dbt owns turning raw, landed tables into clean, tested, documented business models: staging models that clean and standardize the raw data, intermediate models that join and reshape it, and mart models that represent the business-ready tables analysts and dashboards actually query. dbt runs entirely inside the warehouse (it compiles to SQL and executes it there), has a built-in testing framework (uniqueness, not-null, referential integrity checks), and generates documentation and a lineage graph as a byproduct of how models are defined. What it does not do: extract data from anywhere, or decide when things run; dbt has no native concept of a schedule.

Airflow (or an orchestrator): the sequencing and scheduling layer. Airflow owns triggering the Fivetran/Stitch sync (or waiting for its completion signal), then triggering the dbt run once the extract is confirmed complete, then handling anything else the pipeline needs that is not itself extract or transform: custom Python steps a managed connector does not support, cross-system coordination, retries and alerting across the whole chain, and backfill/reprocessing when something upstream needs to be corrected. Without Airflow (or something playing its role), the extract and transform layers have no coordinated way to know about each other's state; each runs on its own internal schedule with no guarantee the transform actually starts only after a genuinely complete extract.

Worked example

A modern ELT pipeline pulling Salesforce, Stripe, and HubSpot data into Snowflake, transforming it into a unified customer-360 mart:

graph LR
    A[Fivetran: Salesforce sync] --> D[Airflow: wait for sync completion]
    B[Fivetran: Stripe sync] --> D
    C[Fivetran: HubSpot sync] --> D
    D --> E[Airflow triggers: dbt run]
    E --> F[dbt: staging models]
    F --> G[dbt: customer_360 mart]
    G --> H[Airflow: write pipeline_status, alert on failure]

• Fivetran runs three managed syncs (Salesforce, Stripe, HubSpot), each on its own connector-specific schedule, landing raw tables into a raw schema in Snowflake, handling API auth, pagination, and schema drift for each source without any custom code from the data team.
• Airflow does not do any of the extraction itself; it has a task (a sensor, or a call to Fivetran's API checking sync-completion status) that waits until all three syncs report complete for the day, since a dbt run kicked off against a partially-synced raw table would silently build the customer-360 mart from incomplete data.
• dbt then runs, building staging models (one per raw source, standardizing column names and types), then the customer_360 mart, joining across all three cleaned staging models, with dbt's own tests (unique, not_null on the customer key) running as part of the same dbt run/dbt test invocation.
• Airflow wraps the whole sequence: if any step fails (a Fivetran sync error, or a failed dbt test), it retries where appropriate and alerts the data team; on success, it writes a status record so downstream consumers of customer_360 know the day's data is ready.

Trade-offs and pitfalls

Trying to use Airflow to also do the extraction (writing custom Python operators to hit the Salesforce, Stripe, and HubSpot APIs directly, instead of using Fivetran) is a common mistake for teams that already have Airflow and reach for it as a hammer; it means building and maintaining source-specific connector logic, pagination, and schema-drift handling in-house, which is exactly the ongoing maintenance burden managed extract services exist to remove, and it is rarely worth it unless a source has no supported connector or the team has genuinely unusual extraction requirements a managed tool cannot meet.

Running dbt without Airflow (or any orchestrator) coordinating it, relying instead on dbt Cloud's own built-in scheduler or a simple cron job, works fine for a standalone transformation layer but loses the ability to sequence dbt correctly against the extract layer's actual completion state; a dbt run kicked off on a fixed schedule with no awareness of whether Fivetran's sync actually finished on time is the same "fixed schedule guessing at readiness" trap that shows up whenever two independently-scheduled systems are chained without an explicit completion signal between them.

Finally, treating Fivetran/Stitch's schema-drift handling as a substitute for dbt's own testing is a subtle but real gap: a managed extract tool successfully landing a column with a changed data type is not the same as that change being safe for downstream models, and skipping dbt-level tests on the assumption that "the extract layer would have caught it" leaves exactly the kind of quiet, business-logic-breaking change that a source schema change often causes undetected until a dashboard number looks wrong.