[saitejabandaru-in]

Closes #71923.

Motivation

When a sensor fails, the scheduler calls TaskInstance.fetch_handle_failure_context. This method clears older task_reschedule records and mutates ti.id to a new UUID by calling ti.prepare_db_for_next_try, and then it calls on_task_instance_failed listener hooks before flushing the updated ti.id to the database.

If a listener hook blocks for seconds (e.g. OpenLineage timing out), the transaction window is extended. Concurrently, an API server might process a reschedule request from the supervisor and insert a new task_reschedule row using the old ti.id. When the listener finishes, the scheduler flushes UPDATE task_instance SET id = <new_id> WHERE id = <old_id>. But since the newly inserted task_reschedule points to old_id, PostgreSQL rejects the update with a ForeignKeyViolation, causing the scheduler to crash.

Changes

• Moved ti.prepare_db_for_next_try(session) in fetch_handle_failure_context to run after the listener hooks.
• This reduces the race window between the task_reschedule delete and the ti.id update flush from seconds down to microseconds.
• Bonus correctness fix: Listener hooks now receive the TaskInstance with the old (actual) UUID that experienced the failure, rather than the prematurely rolled new UUID that hasn't run yet.

[kaxil]

Closing this. Main already notifies listeners before allocating the next try: fetch_handle_failure_context calls ti.notify_failure(error) and only then ti.prepare_db_for_next_try(session) (#73554). The block this PR edits no longer exists, and it adds no test.