Release TI lock before asset listener callbacks

[uranusjr]

Asset registration on the task-success path (ti_update_state) ran the listener hooks synchronously inside the transaction holding a row lock on the task_instance table. A slow listener, multiplied across a large fan-out of asset events, could hold that lock for minutes, causing statement timeouts.

The listener hooks are now deferred until the end of the endpoint instead of executed inline during asset event creation. Registration writes to the database still happen under the caller's transaction, so durability is unchanged; this only moves the best-effort listener hooks off the lock.

Close #66853.

[jason810496]

No comments from my end, LGTM with this direction, thanks.

[jason810496]

Could we also address the loading issue that #66854 (comment) comment pointed out in this one? Since this PR is small enough. Thanks.

[kaxil]

2 minor comments lgtm otherwise

[Andrushika]

Nit: Previously, listeners ran after each event was created. Now their callbacks are collected and run only after all outlets have been processed.

If the second outlet fails with a non-DB error, an event already created for the first outlet can still be committed, but its callbacks are lost because the function never returns them.

(This may be fine since listeners are best effort, just to confirm that this is intentional)

[uranusjr]

Registration callback failures cannot bubble here because the hook caller unconditionally wraps them in a try-except (see #70951 (comment)). So all callbacks will always be run unless you kill the server process entirely.

[Andrushika]

Sorry, my wording was not clear... I meant a failure during registration itself, before the callbacks are handed back.

For example, if 9 out of 10 outlets are processed and the last one raises a non-DB exception, asset_callbacks here stays empty:

asset_callbacks = ()

try:
    # raises inside, while processing the 10th outlet
    query, updated_state, asset_callbacks = _create_ti_state_update_query_and_update_state(...)
    # so this assignment never happens and asset_callbacks is still ()
except Exception:
    ...  # caught, TI marked FAILED, execution continues

session.commit()  # the 9 events are already in the transaction, so they are committed here

for callback in asset_callbacks:  # empty, the loop never runs
    callback()

So the 9 events are committed, but their listeners are never called. Probably still fine as best effort, just wanted to make sure this case was considered!