From 8f5523bd2e0b4d27578641095f18d62d7eec721d Mon Sep 17 00:00:00 2001 From: lda Date: Thu, 10 Sep 2026 03:57:34 +0700 Subject: [PATCH] fix: release admission transaction before dispatch --- src/wf_scheduling/poll.py | 48 +++++++++++++++++++++++++++------------ 1 file changed, 33 insertions(+), 15 deletions(-) diff --git a/src/wf_scheduling/poll.py b/src/wf_scheduling/poll.py index 109dff27..f24b9ccd 100644 --- a/src/wf_scheduling/poll.py +++ b/src/wf_scheduling/poll.py @@ -16,6 +16,7 @@ contains no fixture input, fixture environments, or canned results. from __future__ import annotations +from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, cast @@ -49,6 +50,14 @@ class BlockedSchedule(ValueError): """A schedule is blocked by a corrupt record and fails closed.""" +@dataclass(frozen=True, slots=True) +class _AdmissionOutcome: + """Result of one atomic admission and any dispatch required afterward.""" + + result: str | None + dispatch_run_id: str | None = None + + def source_for_trigger(trigger: Any) -> OccurrenceSource: """Build a calendar source from a schedule trigger model.""" kind = trigger.kind if hasattr(trigger, "kind") else trigger.get("kind") @@ -316,27 +325,37 @@ class Scheduler: return None def _admit(self, sched: Any, intended: datetime, now: datetime) -> str | None: - """Admit one occurrence under the schedule store's local transaction.""" + """Admit atomically, then dispatch a new run after the transaction. + + The durable admission and all schedule-side markers must commit as one + compound transition. Registration can wait on the event loop, though, + so it deliberately happens after that transaction releases; otherwise + an event-loop admin mutation can wait on a worker that is waiting for + the event loop to register its task. + """ with schedule_store_transaction(self.schedule_store): - return self._admit_locked(sched, intended, now) + outcome = self._admit_locked(sched, intended, now) + if outcome.dispatch_run_id is not None: + self._execute_guarded(outcome.dispatch_run_id, now) + return outcome.result def _admit_locked( self, sched: Any, intended: datetime, now: datetime - ) -> str | None: + ) -> _AdmissionOutcome: # The poller's listing is only a fairness snapshot. Re-read under the # admission transaction so an edit committed before this point wins; # an edit after this point waits and affects a later occurrence. try: current = self.schedule_store.get_schedule(sched.id) except KeyError: - return None + return _AdmissionOutcome(None) if current.revision != sched.revision: - return "schedule-changed" + return _AdmissionOutcome("schedule-changed") sched = current if getattr(sched, "blocked_reason", None): raise BlockedSchedule(getattr(sched, "blocked_reason")) if not sched.enabled or sched.deleted or sched.paused: - return None + return _AdmissionOutcome(None) # Fail closed before any new admission when a corrupt active view # exists for this schedule. for run in self._active(sched.id): @@ -370,7 +389,7 @@ class Scheduler: now=now, admitted_at=now, ) - return existing + return _AdmissionOutcome(existing) prepared = self.preparer.prepare(sched=sched, intended=intended, now=now) if isinstance(prepared, PreparationRejected): self._record( @@ -385,7 +404,7 @@ class Scheduler: if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) _save_consumed_max(self.schedule_store, sched.id, intended) - return None + return _AdmissionOutcome(None) active = self._active(sched.id) if sched.overlap == "skip" and active: self._record( @@ -399,7 +418,7 @@ class Scheduler: if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) _save_consumed_max(self.schedule_store, sched.id, intended) - return None + return _AdmissionOutcome(None) if sched.overlap == "parallel" and len(active) >= sched.max_active_runs: self._record( kind="skipped-overlap", @@ -412,7 +431,7 @@ class Scheduler: if cand is not None and cand.intended_at == intended: self.schedule_store.save_candidate(None, schedule_id=sched.id) _save_consumed_max(self.schedule_store, sched.id, intended) - return None + return _AdmissionOutcome(None) if self._task_load() >= self.capacity: if sched.misfire == "latest": self.schedule_store.save_candidate( @@ -424,7 +443,7 @@ class Scheduler: schedule_id=sched.id, ) _save_consumed_max(self.schedule_store, sched.id, intended) - return "held" + return _AdmissionOutcome("held") if (now - intended).total_seconds() > sched.lateness_allowance_s: self._record( kind="skipped-misfire", @@ -434,8 +453,8 @@ class Scheduler: revision=sched.revision, ) _save_consumed_max(self.schedule_store, sched.id, intended) - return None - return "held-undecided" + return _AdmissionOutcome(None) + return _AdmissionOutcome("held-undecided") run_id = self.run_store.allocate_run_id() from wf_api.run_lifecycle import persist_admission @@ -471,8 +490,7 @@ class Scheduler: self.run_store.mark_pending_dispatch(run_id) materialize_admitted_view(store=self.run_store, admission=admission) - self._execute_guarded(run_id, now) - return run_id + return _AdmissionOutcome(run_id, dispatch_run_id=run_id) def _execute_guarded(self, run_id: str, now: datetime) -> None: """Dispatch one provably undispatched run behind a durable transition.