hkc-8010 commented on PR #66854: URL: https://github.com/apache/airflow/pull/66854#issuecomment-4957753267
@uranusjr @Andrushika you're right, and I've stopped defending the current shape. Returning 204 while a failed `register_asset_changes_in_db` is only logged and counted means a dropped asset event silently never fires its downstream DAG, and "follow up later" isn't good enough. As it stands the PR isn't an improvement on its own. Before I push more code I want to agree on the shape. Proposal: a durable transactional outbox plus a scheduler reconciliation sweep, so a dropped registration is guaranteed to be delivered eventually rather than lost. 1. **Durable marker committed atomically with the TI state.** On the success path, alongside the existing early `session.commit()`, I insert one row into a new `asset_registration_outbox` table capturing `ti_id` and the (already-validated) `task_outlets` and `outlet_events`. Because it commits in the same transaction as the state update, the 204 is only returned once the pending-registration record is durable, so even a crash between the state commit and registration can't lose the event. That is the "not silently broken" guarantee. 2. **Inline registration stays post-commit, as an optimization.** The expensive work (scheduled-dag scan, `AssetDagRunQueue` inserts, alias association) still runs after the lock is released, so the #66853 contention fix is preserved. On success it deletes the outbox row in the same transaction as the asset writes, which keeps it exactly-once against that row; on failure it rolls the asset writes back and leaves the outbox row for the scheduler. The outbox insert itself touches no hot rows and runs no queries, so it doesn't reintroduce the `task_instance` lock contention. 3. **Scheduler reconciles leftovers.** A recurring sweep (claiming rows with `FOR UPDATE SKIP LOCKED`, HA-safe, modeled on the existing heartbeat/zombie purge) replays any leftover outbox rows through `register_asset_changes_in_db` and deletes them, so the events land as `AssetEvent` and `AssetDagRunQueue` rows and `_create_dag_runs_asset_triggered` picks them up. A `stats.gauge` exposes the pending backlog so operators can see if reconciliation falls behind. @Andrushika thanks, your reorder (register before taking the TI lock, then commit state and assets together) is a clean way to shrink the lock window and I considered it seriously. I leaned toward the outbox for two reasons. The reorder still needs a full audit of every `task_instance` writer for lock-ordering and deadlock risk before I'd trust it, and it still loses the event if the process dies between the asset work and the commit. The outbox gives an explicit, testable at-least-once guarantee instead. I'm happy to go the reorder route if you'd both prefer it; I don't have a strong attachment beyond wanting the guarantee to be verifiable. Two questions before I implement: - Does this shape address the consistency concern for you? - Should the outbox and reconciliation land in this PR (it adds a table, a migration, and a scheduler timer), or as a stacked PR that merges together with this one? I'd rather not split it in a way that leaves a window where the drop is unhandled on `main`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
