Retries Are Duplicating Rows in a Metrics Table

A daily DAG computes a marketing-spend rollup and appends it to a reporting table:

@task(retries=3, retry_delay=timedelta(minutes=10))
def compute_and_append_spend(data_interval_start=None) -> None:
    rows = _aggregate_spend_from_ad_platforms(data_interval_start)
    _append_rows(rows, table="reporting.daily_spend")

_aggregate_spend_from_ad_platforms calls three different ad platforms' APIs and sometimes one of the three times out. On days when that happens, the task fails, Airflow retries it (up to 3 times), and it eventually succeeds. Finance notices that reporting.daily_spend has roughly 1.4x the expected row count for dates where a retry occurred, and totals are inflated to match.

1. Explain exactly how a task retrying can produce 1.4x the expected rows rather than either 1x (correct) or 2x/3x/4x (a full duplicate set per attempt).
2. Redesign compute_and_append_spend so it's safe to retry any number of times.
3. A teammate suggests "just set retries=0 so this can never happen." Evaluate that proposal.