Drain sequential event tasks during shutdown

[harsh21234i]

Fixes #6755.

Summary

• update EventProcessor._stop_tasks() to keep draining tasks dispatched during graceful shutdown
• include same-token sequential tasks created by _finish_task() within the remaining shutdown budget
• add regression coverage for same-token queued events draining before shutdown completes

Tests

• uv run pytest tests/units/reflex_base/event/processor/test_event_processor.py -q
• uv run ruff check packages/reflex-base/src/reflex_base/event/processor/event_processor.py tests/units/reflex_base/ event/processor/test_event_processor.py
• uv run ruff format --check packages/reflex-base/src/reflex_base/event/processor/event_processor.py tests/units/ reflex_base/event/processor/test_event_processor.py
• uv run pyright packages/reflex-base/src/reflex_base/event/processor/event_processor.py tests/units/reflex_base/ event/processor/test_event_processor.py

[greptile-apps]

Greptile Summary

This PR fixes a bug where same-token sequential events queued during graceful shutdown were cancelled instead of drained. The old as_completed loop captured _tasks at shutdown time and missed tasks dispatched later by _finish_task, whereas the new while self._tasks loop re-evaluates _tasks after each asyncio.wait(FIRST_COMPLETED) call so every newly-dispatched sequential task is picked up within the remaining time budget.

• _stop_tasks rewrite: replaced the as_completed-based loop with a deadline-anchored while self._tasks: loop using asyncio.wait(FIRST_COMPLETED), which continuously re-evaluates self._tasks and therefore sees tasks added by _finish_task callbacks on the fly.
• Simpler outstanding-tasks list: because _finish_task already removes completed tasks from self._tasks, the post-loop outstanding_tasks computation no longer needs a set-difference; list(self._tasks.values()) is sufficient and correct.
• Regression test: test_stop_drains_same_token_sequential_backlog enqueues 10 slow sequential events under a 2-second shutdown budget and asserts all complete in order without cancellation.

Confidence Score: 5/5

Safe to merge — the change is tightly scoped to the shutdown drain loop and is backed by a direct regression test.

The logic change is correct: asyncio done callbacks run synchronously when a task finishes, so self._tasks is already updated by the time each asyncio.wait call returns. The while loop therefore always works on a fresh snapshot, the deadline guard prevents infinite looping, and the if-not-finished-tasks break handles timeouts cleanly.

No files require special attention.

Important Files Changed

Filename	Overview
packages/reflex-base/src/reflex_base/event/processor/event_processor.py	Replaces as_completed-based drain with a while loop using asyncio.wait(FIRST_COMPLETED), allowing newly-dispatched sequential tasks added by _finish_task to be picked up within the same shutdown budget.
tests/units/reflex_base/event/processor/test_event_processor.py	Adds test_stop_drains_same_token_sequential_backlog, which enqueues 10 sequential slow events and verifies all run to completion (in order, not cancelled) within a 2-second graceful timeout.
packages/reflex-base/news/6791.bugfix.md	New changelog entry describing the bug fix.

_{Reviews (2): Last reviewed commit: "Add changelog for event processor shutdo..." | Re-trigger Greptile}

[codspeed-hq]

Merging this PR will not alter performance

✅ 26 untouched benchmarks
⏩ 8 skipped benchmarks¹

---

_{Comparing harsh21234i:fix/event-processor-stop-drains-token-queues-6755 (ba1a27f) with main (2edaf12)}

Footnotes

1. 8 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩

[harsh21234i]

hey @FarhanAliRaza can you go through this