Task retries after approval expiry: reconciling a committed export without authorising a new one

[AffixIO]

I'm testing a task where permission to perform a data export expires, but a retry may still need to confirm an export that already completed. I'd appreciate a sanity check on the Airflow lifecycle assumptions.

The local case is deliberately small: a TaskFlow task validates a signed approval for fixed fictional inputs, writes the export and its operation record in one SQLite transaction, then raises an exception after commit. On retry, it checks for that operation first. An identical completed operation returns RECONCILED_EXISTING; if nothing was committed, the approval must still be valid before a new write. The operation ID stays fixed across attempts.

On Airflow 3.1.0, dag.test() ran the first attempt, marked it up for retry, then completed on the second attempt with one export row. Separate tests cover expiry before a new write and reconciliation after expiry. This only demonstrates a local transactional sink, not exactly-once delivery to an external system or a distributed executor test.

Is keeping that distinction entirely inside the task callable the appropriate pattern, particularly when a failed task is manually cleared? My intended behaviour is that clearing the task does not itself grant permission for a new export: the retained operation record still governs reconciliation, and a new logical export needs a newly approved operation ID. I'm not using try_number as the idempotency key; it is only used to inject the first-attempt failure in the demo.

The runnable example and test results are at https://github.com/AffixIO/airflow-approved-export. I maintain AffixIO, whose local SDK supplies the signature check in this example. There's no request to add it to Airflow, and no real data or external export endpoint is involved. If there's an existing example addressing expiring authorisation versus completed-side-effect reconciliation, I'd rather build on that than duplicate it.

[kinqsradio]

Yes, keeping that decision inside the task is the right place, because Airflow has no notion of "this side effect already happened" or "this attempt is authorised". A retry, a manual clear and a re-trigger are all just "run the callable again" to the scheduler. Clearing resets the task instance state so it's scheduled again, and nothing more. So your rule that clearing doesn't grant permission for a new export holds automatically, as long as the callable checks the operation record first. The best practices say the same thing more generally: tasks should be idempotent, and Airflow won't do it for you.

A few points worth adding to the design:

• Make the check-and-write atomic, not just ordered. On a distributed executor, a task whose heartbeat is lost can be marked failed and retried while the first process is still running. "Check for the operation, then write" in two steps can then produce two exports. A unique constraint on the operation ID in the same transaction as the export makes the second writer fail instead of duplicating. You get this for free in SQLite. Make sure the real sink has an equivalent (a unique key, a conditional put, or an idempotency key on the external API).
• Keep the operation ID out of Airflow's run identity. You already avoid try_number. The same goes for run_id and the logical date if a new approved export should get a new ID. Passing the ID in via params or dag_run.conf (or reading it from the approval) means a re-triggered run with the same ID reconciles, and a new export needs a new approval, which is what you described.
• Decide what "reconciled" returns. If downstream tasks pull an XCom from this task, make RECONCILED_EXISTING return the same payload as the original success, so a cleared-and-reconciled run looks the same downstream.
• External sinks: for a sink you can't write to transactionally, the usual pattern is write-ahead: record "intent" with the operation ID, perform the export using that ID as the idempotency key, then mark it committed. A retry that finds "intent" without "committed" asks the sink whether that ID exists, instead of re-checking the approval.

I'm not aware of an Airflow example that covers the expiring-approval part specifically. Most examples stop at idempotent writes.