hkc-8010 commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3700077358
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -497,12 +502,38 @@ def ti_update_state(
extra=json.dumps({"host_name": hostname}) if hostname else
None,
)
)
+ # Durably record the successful task's asset events in the SAME
transaction that commits the
+ # state, so the marker can never be lost in a crash between this
commit and their
+ # registration. The scheduler drain is the single writer that
registers them; the request
+ # never runs asset registration itself, which is what keeps the
task_instance row lock short
+ # under high fan-out. Only a genuine RUNNING->SUCCESS transition
reaches here (a duplicate
+ # SUCCESS->SUCCESS short-circuits earlier), so a completion is never
enqueued twice.
+ if (
+ updated_state == TaskInstanceState.SUCCESS
+ and isinstance(ti_patch_payload, TISuccessStatePayload)
+ and (ti_patch_payload.task_outlets or
ti_patch_payload.outlet_events)
+ ):
+ _enqueue_asset_events(
+ task_instance_id=task_instance_id,
+ dag_id=dag_id,
+ run_id=run_id,
+ task_id=task_id,
+ map_index=map_index,
+ task_outlets=ti_patch_payload.task_outlets,
+ outlet_events=ti_patch_payload.outlet_events,
+ session=session,
+ )
+ # Commit the state, log entry and durable queue marker together so
they land atomically and
+ # the row lock is released promptly.
+ session.commit()
Review Comment:
I kept the explicit commit in the route on purpose. That commit is the
outbox boundary for this path: the TI state update, audit log row, and
`asset_event_queue` marker need to land together before the request moves on.
If we defer that to request teardown, the optional `clear_on_success` block
gets folded into the same transaction. I do not think that is a safe trade
here. `clear_on_success` is best-effort cleanup, and a failure there should not
roll back an already-successful task or its durable queue marker. That would
turn a cleanup problem into a task-state correctness problem. So for this PR I
took the narrow naming, logging, and comment cleanup, but I left the
route-level commit boundary intact.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -521,7 +552,9 @@ def ti_update_state(
task_id=task_id,
map_index=map_index,
)
+ session.commit()
except Exception:
+ session.rollback()
Review Comment:
Same reasoning here. I left the route-level commit boundary in place because
it is the outbox boundary for the TI state update, audit log row, and durable
queue marker. If this is deferred to session teardown, the optional
`clear_on_success` cleanup gets folded into the same transaction, and a cleanup
failure can affect an already-successful task. I do not think that is the right
trade in this PR, so I kept the narrow review cleanups and left the commit
boundary unchanged.
--
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]
