Publish flow-run state events after their transaction commits

[devin-ai-integration]

closes #22654

InstrumentFlowRunStateTransitions.after_transition published the flow-run state change event inline, while orchestration rules were unwinding inside set_flow_run_state — i.e. before the API route's db.session_context(begin_transaction=True) committed. A subscriber could receive prefect.flow-run.Suspended and still read the run as Running, and a transition that was rolled back emitted an event anyway.

This PR builds the event exactly as before (it needs the live session to resolve related resources) but publishes it from a docket task queued as the transaction commits:

event = await flow_run_state_change_event(session=context.session, ...)
publish_after_commit(context.session, event)

call_after_commit (server/utilities/_post_commit.py) queues hooks on the session and drains them from SQLAlchemy's after_commit event. That event fires inside the greenlet the async session used to commit, so sqlalchemy.util.await_only runs each hook as a coroutine on the original event loop — no cross-loop hand-off and no fire-and-forget task. Hooks belong to the transaction that was active when they were registered, so rolling back a transaction (or a savepoint) discards its hooks, and hook errors are logged rather than failing the commit.

publish_after_commit (server/events/_publishing.py) then hands the event to docket rather than emitting it itself, so publishing gets a task's retries and observability:

await docket.add(publish_event, key=f"publish-event:{event.id}")(event, event.id)

Details

• Single (POST /flow_runs/{id}/set_state) and bulk (POST /flow_runs/bulk_set_state) transitions both funnel through models.flow_runs.set_flow_run_state, so both are fixed centrally, along with internal callers (services, event actions) that wrap the call in a session context.
• Savepoints (session.begin_nested()) are handled explicitly: releasing one dispatches after_commit but isn't durable, so hooks stay queued until the enclosing transaction commits; rolling one back discards only the hooks registered inside it.
• publish_event is registered in server/api/background_workers.py with Retry(attempts=5, delay=0.5s). Its key is the event id, so a duplicate enqueue for the same event dedupes.
• server/utilities/_docket.py gives code with neither a request nor a task context access to the docket: background_worker() serves the docket it runs, and get_docket() prefers docket's own current_docket when we're already inside a task. Ephemeral servers get this for free — their lifespan starts a worker on the memory:// docket.
• Failure modes: the docket.add itself happens after the commit, so a failure there is logged and dropped rather than reverting the committed transition (durable enqueue would need an outbox row written inside the transaction, which this PR doesn't add). With no docket at all — orchestration driven directly against a session, outside a running server — the event is published in-line, still after the commit.
• Publishing from a worker makes per-run event order best-effort. Trigger evaluation and the task-run recorder already tolerate that via follows/causal ordering, but websocket subscribers can now see two events for the same run out of order.
• New regression tests, all of which fail without the fix:
  • tests/server/orchestration/test_flow_run_instrumentation_policies.py::TestPublishingEventsAfterCommit — nothing is published until the commit; a rolled-back transition publishes nothing; for both the single and bulk API routes, reading the run from a separate session at publish time sees the committed state (before the fix the bulk case even deadlocked on SQLite's writer lock, since the emit happened while the write transaction was open).
  • tests/events/server/test_publishing.py — the commit only enqueues the task, a rollback enqueues nothing, a failed enqueue doesn't fail the commit, and the no-docket path publishes in-line.
  • tests/server/utilities/test_post_commit.py — ordering, single delivery, rollback and savepoint-rollback discarding, and error isolation for the hook utility.
• Existing instrumentation tests already commit before asserting on AssertingEventsClient, and pass unchanged.

Checklist

• This pull request references any related issue by including "closes Flow-run state events can be delivered before their states are committed #22654"
• If this is a complex change, a maintainer has confirmed the proposed approach on the linked issue.
  • The issue asks for either publishing after commit or a transactional outbox; this takes the former, with the docket hand-off @desertaxle asked for in review.
• If this pull request adds or changes functionality, it includes tests or explains why tests are not needed.
• If this pull request changes user-facing behavior, it updates documentation or explains why documentation is not needed.
  • No docs change: event payloads and timing semantics are not documented beyond "state change events", and this only tightens delivery ordering.
• If this pull request removes docs files, it includes redirect settings in mint.json.
• If this pull request adds functions or classes, it includes helpful docstrings.

Link to Devin session: https://app.devin.ai/sessions/b48782b353cd420f8338ce44c7c66b44

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 0d9e7eee6a

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[codspeed-hq]

Merging this PR will not alter performance

✅ 8 untouched benchmarks
⏩ 1 skipped benchmark¹

---

_{Comparing devin1/oss-8109-publish-flow-run-state-events-after-commit (bb45f4d) with main (edc4fdd)}

Footnotes

1. 1 benchmark was skipped, so the baseline result was used instead. If it was deleted from the codebase, click here and archive it to remove it from the performance reports. ↩

[desertaxle]

Could Docket run these post-commit hooks instead of awaiting them inside SQLAlchemy's after_commit callback? A registered task could provide retries and observability, but we'd need to preserve ephemeral operation and define enqueue failure after commit.

[desertaxle]

Returning here keeps hooks registered inside the rolled-back savepoint. The outer commit then drains them, so a reverted transition still emits an event. Can you track hook ownership by transaction scope and test nested rollback?

[devin-ai-integration]

You're right, fixed in de92bd5. Each hook is now stored with the transaction that was active when it was registered (get_nested_transaction() or get_transaction()), and after_soft_rollback drops every hook whose owning transaction is the one that rolled back or a descendant of it. So a savepoint rollback discards only the hooks registered inside it, and an outer rollback still discards everything.

Two tests: test_hooks_are_discarded_when_their_savepoint_rolls_back (fails on the previous commit) and test_hooks_survive_the_rollback_of_a_later_savepoint.

[devin-ai-integration]

Could Docket run these post-commit hooks instead of awaiting them inside SQLAlchemy's after_commit callback?

It could — the hook body would become a registered task taking the already-built Event as a serializable argument, and docket.add(..., key=f"flow-run-state-event:{state_id}") after commit would give us retries, idempotency keys, and task-level observability. Some things I'd want to settle first, which is why I kept this PR to the narrow ordering fix:

• The enqueue is still the unretried step. It happens after the commit, so a Redis hiccup there is the same log-and-drop we have today, just moved. Nothing is gained unless the enqueue itself is durable (i.e. an outbox row in the same transaction, which is the other option the issue mentions).
• Ephemeral runs on memory:// docket (server.docket.url default), so durability doesn't improve there, and execution_ttl=timedelta(0) means tasks are dropped rather than retried under load.
• Ordering. Awaiting in the commit unwind keeps state events for a run in transition order, which trigger evaluation and websocket subscribers depend on. Handing them to a worker pool makes ordering best-effort unless we key tasks per flow run and serialize them.
• Events already pass through the messaging layer (in-memory or Redis), so docket would be a second queue in front of that one.

Happy to do it as a follow-up if you want the retry/observability story — a transactional outbox drained by a docket task is probably the version that actually buys durability. Want me to open an issue for it, or fold it into this PR?

[desertaxle]

Want me to open an issue for it, or fold it into this PR?

Fold it into this PR.

[devin-ai-integration]

Done in bb45f4d. The post-commit hook now enqueues a docket task instead of emitting the event itself:

• publish_event(event, event_id) in server/events/_publishing.py is a registered task (Retry(attempts=5, delay=0.5s), event_id is Logged, task key is publish-event:{event.id}, so a re-enqueue of the same event dedupes).
• server/utilities/_docket.py exposes the docket to code that has neither a request nor a task context: background_worker() serves the docket it's running, and get_docket() prefers docket's own current_docket when we're already inside a task. Ephemeral gets this for free since its lifespan starts a worker on the memory:// docket.
• Enqueue failure after commit: the docket.add happens in the commit unwind, and a failure there is logged and dropped — the commit still stands and the transition is not reverted. Same failure mode as before, just moved from the emit to the enqueue; making it durable needs the outbox row inside the transaction, which I left out of this PR.
• No docket at all (orchestration driven directly against a session, e.g. models.flow_runs.set_flow_run_state in tests or embedded use) falls back to publishing in-line, still after the commit.

Consequence worth flagging: state events are now published by a worker, so per-run delivery order is best-effort. Trigger evaluation and the task-run recorder already handle that through follows/causal ordering, but websocket subscribers can now see two events for the same run out of order.

Tests in tests/events/server/test_publishing.py: enqueue-only-after-commit (nothing published until the worker runs), rollback enqueues nothing, enqueue failure doesn't fail the commit, and the no-docket in-line path.