Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/orchestration/core_policy.py: 30%
657 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1"""
2Orchestration logic that fires on state transitions.
4`CoreFlowPolicy` and `CoreTaskPolicy` contain all default orchestration rules that
5Prefect enforces on a state transition.
6"""
8from __future__ import annotations
10import datetime
11import logging
12import math
13from typing import Any, Union, cast
14from uuid import UUID, uuid4
16import sqlalchemy as sa
17from packaging.version import Version
18from sqlalchemy import select
19from sqlalchemy.ext.asyncio import AsyncSession
21from prefect.logging import get_logger
22from prefect.server import models
23from prefect.server.concurrency.lease_storage import (
24 ConcurrencyLeaseHolder,
25 ConcurrencyLimitLeaseMetadata,
26 get_concurrency_lease_storage,
27)
28from prefect.server.database import PrefectDBInterface, orm_models
29from prefect.server.database.dependencies import db_injector, provide_database_interface
30from prefect.server.exceptions import ObjectNotFoundError
31from prefect.server.models import concurrency_limits, concurrency_limits_v2, deployments
32from prefect.server.orchestration.dependencies import (
33 MIN_CLIENT_VERSION_FOR_CONCURRENCY_LIMIT_LEASING,
34 WORKER_VERSIONS_THAT_MANAGE_DEPLOYMENT_CONCURRENCY,
35)
36from prefect.server.orchestration.policies import (
37 FlowRunOrchestrationPolicy,
38 TaskRunOrchestrationPolicy,
39)
40from prefect.server.orchestration.rules import (
41 ALL_ORCHESTRATION_STATES,
42 TERMINAL_STATES,
43 BaseOrchestrationRule,
44 BaseUniversalTransform,
45 FlowOrchestrationContext,
46 FlowRunOrchestrationRule,
47 FlowRunUniversalTransform,
48 GenericOrchestrationRule,
49 OrchestrationContext,
50 TaskRunOrchestrationRule,
51 TaskRunUniversalTransform,
52)
53from prefect.server.schemas import core, filters, states
54from prefect.server.schemas.states import StateType
55from prefect.server.task_queue import TaskQueue
56from prefect.settings import (
57 get_current_settings,
58)
59from prefect.types._datetime import now
60from prefect.utilities.math import clamped_poisson_interval
62from .instrumentation_policies import InstrumentFlowRunStateTransitions
64logger: logging.Logger = get_logger(__name__)
67def _has_persisted_result(data: Any) -> bool:
68 """Return True if *data* looks like persisted Prefect result metadata.
70 `StateCreate.data` accepts `Any`, so callers must never assume
71 dict-like access. This helper centralises the check so that both
72 `HandleFlowTerminalStateTransitions` and `PreventResultDataLoss`
73 use the same logic.
74 """
75 return isinstance(data, dict) and data.get("type") != "unpersisted"
78async def _release_concurrency_lease(
79 session: AsyncSession,
80 lease_id: UUID,
81 fallback_concurrency_limit_ids: list[UUID] | None = None,
82 fallback_slots: int = 1,
83) -> None:
84 lease_storage = get_concurrency_lease_storage()
85 lease = await lease_storage.read_lease(lease_id=lease_id)
86 if not lease:
87 logger.warning(f"Lease {lease_id} not found during release")
88 if fallback_concurrency_limit_ids:
89 slots_released = await concurrency_limits_v2.bulk_decrement_active_slots(
90 session=session,
91 concurrency_limit_ids=fallback_concurrency_limit_ids,
92 slots=fallback_slots,
93 )
94 if not slots_released:
95 raise RuntimeError(
96 f"Failed to release concurrency slots for lease {lease_id}"
97 )
98 return
100 slots_released = await concurrency_limits_v2.bulk_decrement_active_slots(
101 session=session,
102 concurrency_limit_ids=lease.resource_ids,
103 slots=lease.metadata.slots if lease.metadata else 1,
104 )
105 if not slots_released:
106 raise RuntimeError(f"Failed to release concurrency slots for lease {lease_id}")
108 await lease_storage.revoke_lease(lease_id=lease.id)
111class CoreFlowPolicy(FlowRunOrchestrationPolicy):
112 """
113 Orchestration rules that run against flow-run-state transitions in priority order.
114 """
116 @staticmethod
117 def priority() -> list[
118 Union[
119 type[BaseUniversalTransform[orm_models.FlowRun, core.FlowRunPolicy]],
120 type[BaseOrchestrationRule[orm_models.FlowRun, core.FlowRunPolicy]],
121 ]
122 ]:
123 return cast(
124 list[
125 Union[
126 type[
127 BaseUniversalTransform[orm_models.FlowRun, core.FlowRunPolicy]
128 ],
129 type[BaseOrchestrationRule[orm_models.FlowRun, core.FlowRunPolicy]],
130 ]
131 ],
132 [
133 PreserveDeploymentConcurrencyLeaseId,
134 PreventDuplicateTransitions,
135 HandleFlowTerminalStateTransitions,
136 EnforceCancellingToCancelledTransition,
137 BypassCancellingFlowRunsWithNoInfra,
138 PreventPendingTransitions,
139 ValidateDeploymentConcurrencyAtRunning,
140 SecureFlowConcurrencySlots,
141 RemoveDeploymentConcurrencyLeaseForOldClientVersions,
142 EnsureOnlyScheduledFlowsMarkedLate,
143 HandlePausingFlows,
144 HandleResumingPausedFlows,
145 CopyScheduledTime,
146 WaitForScheduledTime,
147 RetryFailedFlows,
148 InstrumentFlowRunStateTransitions,
149 ReleaseFlowConcurrencySlots,
150 ],
151 )
154class CoreTaskPolicy(TaskRunOrchestrationPolicy):
155 """
156 Orchestration rules that run against task-run-state transitions in priority order.
157 """
159 @staticmethod
160 def priority() -> list[
161 Union[
162 type[BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]],
163 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
164 ]
165 ]:
166 return cast(
167 list[
168 Union[
169 type[
170 BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]
171 ],
172 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
173 ]
174 ],
175 [
176 CacheRetrieval,
177 HandleTaskTerminalStateTransitions,
178 PreventRunningTasksFromStoppedFlows,
179 SecureTaskConcurrencySlots, # retrieve cached states even if slots are full
180 CopyScheduledTime,
181 WaitForScheduledTime,
182 RetryFailedTasks,
183 RenameReruns,
184 UpdateFlowRunTrackerOnTasks,
185 CacheInsertion,
186 ReleaseTaskConcurrencySlots,
187 ],
188 )
191class ClientSideTaskOrchestrationPolicy(TaskRunOrchestrationPolicy):
192 """
193 Orchestration rules that run against task-run-state transitions in priority order,
194 specifically for clients doing client-side orchestration.
195 """
197 @staticmethod
198 def priority() -> list[
199 Union[
200 type[BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]],
201 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
202 ]
203 ]:
204 return cast(
205 list[
206 Union[
207 type[
208 BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]
209 ],
210 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
211 ]
212 ],
213 [
214 CacheRetrieval,
215 HandleTaskTerminalStateTransitions,
216 PreventRunningTasksFromStoppedFlows,
217 CopyScheduledTime,
218 WaitForScheduledTime,
219 RetryFailedTasks,
220 RenameReruns,
221 UpdateFlowRunTrackerOnTasks,
222 CacheInsertion,
223 ReleaseTaskConcurrencySlots,
224 ],
225 )
228class BackgroundTaskPolicy(TaskRunOrchestrationPolicy):
229 """
230 Orchestration rules that run against task-run-state transitions in priority order.
231 """
233 @staticmethod
234 def priority() -> list[
235 type[BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]]
236 | type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]]
237 ]:
238 return cast(
239 list[
240 Union[
241 type[
242 BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]
243 ],
244 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
245 ]
246 ],
247 [
248 PreventPendingTransitions,
249 CacheRetrieval,
250 HandleTaskTerminalStateTransitions,
251 # SecureTaskConcurrencySlots, # retrieve cached states even if slots are full
252 CopyScheduledTime,
253 CopyTaskParametersID,
254 WaitForScheduledTime,
255 RetryFailedTasks,
256 RenameReruns,
257 UpdateFlowRunTrackerOnTasks,
258 CacheInsertion,
259 ReleaseTaskConcurrencySlots,
260 EnqueueScheduledTasks,
261 ],
262 )
265class MinimalFlowPolicy(FlowRunOrchestrationPolicy):
266 @staticmethod
267 def priority() -> list[
268 Union[
269 type[BaseUniversalTransform[orm_models.FlowRun, core.FlowRunPolicy]],
270 type[BaseOrchestrationRule[orm_models.FlowRun, core.FlowRunPolicy]],
271 ]
272 ]:
273 return [
274 PreventResultDataLoss,
275 BypassCancellingFlowRunsWithNoInfra, # cancel scheduled or suspended runs from the UI
276 InstrumentFlowRunStateTransitions,
277 ReleaseFlowConcurrencySlots,
278 ]
281class MarkLateRunsPolicy(FlowRunOrchestrationPolicy):
282 @staticmethod
283 def priority() -> list[
284 Union[
285 type[BaseUniversalTransform[orm_models.FlowRun, core.FlowRunPolicy]],
286 type[BaseOrchestrationRule[orm_models.FlowRun, core.FlowRunPolicy]],
287 ]
288 ]:
289 return [
290 EnsureOnlyScheduledFlowsMarkedLate,
291 EnforceDeploymentConcurrencyOnLate,
292 InstrumentFlowRunStateTransitions,
293 ]
296class MinimalTaskPolicy(TaskRunOrchestrationPolicy):
297 @staticmethod
298 def priority() -> list[
299 Union[
300 type[BaseUniversalTransform[orm_models.TaskRun, core.TaskRunPolicy]],
301 type[BaseOrchestrationRule[orm_models.TaskRun, core.TaskRunPolicy]],
302 ]
303 ]:
304 return [
305 ReleaseTaskConcurrencySlots, # always release concurrency slots
306 ]
309class SecureTaskConcurrencySlots(TaskRunOrchestrationRule):
310 """
311 Checks relevant concurrency slots are available before entering a Running state.
313 This rule checks if concurrency limits have been set on the tags associated with a
314 TaskRun. If so, a concurrency slot will be secured against each concurrency limit
315 before being allowed to transition into a running state. If a concurrency limit has
316 been reached, the client will be instructed to delay the transition for the duration
317 specified by the "PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS" setting
318 before trying again. If the concurrency limit set on a tag is 0, the transition will
319 be aborted to prevent deadlocks.
320 """
322 FROM_STATES = ALL_ORCHESTRATION_STATES
323 TO_STATES = {StateType.RUNNING}
325 async def before_transition(
326 self,
327 initial_state: states.State[Any] | None,
328 proposed_state: states.State[Any] | None,
329 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
330 ) -> None:
331 settings = get_current_settings()
332 self._applied_limits: list[str] = []
333 self._acquired_v2_lease_ids: list[UUID] = []
334 v1_limits = (
335 await concurrency_limits.filter_concurrency_limits_for_orchestration(
336 context.session, tags=context.run.tags
337 )
338 )
339 v2_names = [f"tag:{tag}" for tag in context.run.tags]
340 v2_limits = await concurrency_limits_v2.bulk_read_concurrency_limits(
341 context.session, names=v2_names
342 )
344 # Handle V2 limits first (if they exist)
345 v2_tags: set[str] = set() # Track which tags have V2 limits
346 if v2_limits:
347 lease_storage = get_concurrency_lease_storage()
348 # Track which tags have V2 limits to exclude from V1 processing
349 v2_tags = {
350 limit.name.removeprefix("tag:") for limit in v2_limits if limit.active
351 }
353 # Check for zero limits that would deadlock
354 for limit in v2_limits:
355 if limit.active and limit.limit == 0:
356 # Clean up any already acquired V2 leases
357 for lease_id in self._acquired_v2_lease_ids:
358 try:
359 await _release_concurrency_lease(
360 session=context.session, lease_id=lease_id
361 )
362 except Exception:
363 logger.warning(
364 f"Failed to clean up lease {lease_id} during abort",
365 exc_info=True,
366 )
367 raise
369 await self.abort_transition(
370 reason=f'The concurrency limit on tag "{limit.name.removeprefix("tag:")}" is 0 and will deadlock if the task tries to run again.',
371 )
373 # Try to acquire V2 slots with lease (exclude zero limits as they're handled above)
374 active_v2_limits = [
375 limit for limit in v2_limits if limit.active and limit.limit > 0
376 ]
377 if active_v2_limits:
378 # Attempt to acquire slots
379 async with provide_database_interface().session_context(
380 begin_transaction=True
381 ) as session:
382 acquired = await concurrency_limits_v2.bulk_increment_active_slots(
383 session=session,
384 concurrency_limit_ids=[limit.id for limit in active_v2_limits],
385 slots=1,
386 )
387 if not acquired:
388 await session.rollback()
389 # Use avg_slot_occupancy_seconds from the most
390 # contended limit, capped at the configured max, to
391 # avoid fixed-delay batching where all waiting tasks
392 # wake up simultaneously.
393 max_wait = (
394 settings.server.tasks.tag_concurrency_slot_wait_seconds
395 )
396 blocking_limit = max(
397 active_v2_limits,
398 key=lambda lim: lim.active_slots / lim.limit,
399 )
400 average_interval = min(
401 blocking_limit.avg_slot_occupancy_seconds or max_wait,
402 max_wait,
403 )
404 delay_seconds = clamped_poisson_interval(
405 average_interval=average_interval,
406 )
407 await self.delay_transition(
408 delay_seconds=round(delay_seconds),
409 reason=f"Concurrency limit reached for tags: {', '.join([limit.name.removeprefix('tag:') for limit in active_v2_limits])}",
410 )
411 return
413 # Create lease for acquired slots with minimal metadata first
414 lease = await lease_storage.create_lease(
415 resource_ids=[limit.id for limit in active_v2_limits],
416 ttl=concurrency_limits.V1_LEASE_TTL,
417 metadata=ConcurrencyLimitLeaseMetadata(
418 slots=1,
419 holder=ConcurrencyLeaseHolder(
420 type="task_run",
421 id=context.run.id,
422 ),
423 ),
424 )
426 self._acquired_v2_lease_ids.append(lease.id)
428 remaining_v1_limits = [limit for limit in v1_limits if limit.tag not in v2_tags]
429 if remaining_v1_limits:
430 run_limits = {limit.tag: limit for limit in v1_limits}
431 for tag, cl in run_limits.items():
432 limit = cl.concurrency_limit
433 if limit == 0:
434 # limits of 0 will deadlock, and the transition needs to abort
435 for stale_tag in self._applied_limits:
436 stale_limit = run_limits.get(stale_tag, None)
437 if stale_limit:
438 active_slots: set[str] = set(stale_limit.active_slots)
439 active_slots.discard(str(context.run.id))
440 stale_limit.active_slots = list(active_slots)
442 await self.abort_transition(
443 reason=(
444 f'The concurrency limit on tag "{tag}" is 0 and will deadlock'
445 " if the task tries to run again."
446 ),
447 )
448 elif len(cl.active_slots) >= limit:
449 # if the limit has already been reached, delay the transition
450 for stale_tag in self._applied_limits:
451 stale_limit = run_limits.get(stale_tag, None)
452 if stale_limit:
453 active_slots = set(stale_limit.active_slots)
454 active_slots.discard(str(context.run.id))
455 stale_limit.active_slots = list(active_slots)
457 await self.delay_transition(
458 delay_seconds=int(
459 settings.server.tasks.tag_concurrency_slot_wait_seconds
460 ),
461 # PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS.value(),
462 reason=f"Concurrency limit for the {tag} tag has been reached",
463 )
464 else:
465 # log the TaskRun ID to active_slots
466 self._applied_limits.append(tag)
467 active_slots = set(cl.active_slots)
468 active_slots.add(str(context.run.id))
469 cl.active_slots = list(active_slots)
471 async def cleanup(
472 self,
473 initial_state: states.State[Any] | None,
474 validated_state: states.State[Any] | None,
475 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
476 ) -> None:
477 # Clean up V2 leases
478 for lease_id in self._acquired_v2_lease_ids:
479 try:
480 await _release_concurrency_lease(
481 session=context.session, lease_id=lease_id
482 )
483 except Exception:
484 logger.warning(f"Failed to clean up lease {lease_id}", exc_info=True)
485 raise
487 for tag in self._applied_limits:
488 cl = await concurrency_limits.read_concurrency_limit_by_tag(
489 context.session, tag
490 )
491 if cl:
492 active_slots = set(cl.active_slots)
493 active_slots.discard(str(context.run.id))
494 cl.active_slots = list(active_slots)
497class ReleaseTaskConcurrencySlots(TaskRunUniversalTransform):
498 """
499 Releases any concurrency slots held by a run upon exiting a Running or
500 Cancelling state.
501 """
503 async def after_transition(
504 self,
505 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
506 ) -> None:
507 if self.nullified_transition(): 507 ↛ 508line 507 didn't jump to line 508 because the condition on line 507 was never true
508 return
510 if context.validated_state and context.validated_state.type not in [
511 states.StateType.RUNNING,
512 states.StateType.CANCELLING,
513 ]:
514 v2_names = [f"tag:{tag}" for tag in context.run.tags]
515 v2_limits = await concurrency_limits_v2.bulk_read_concurrency_limits(
516 context.session, names=v2_names
517 )
518 # Release V2 leases for this task run
519 if v2_limits:
520 lease_storage = get_concurrency_lease_storage()
521 lease_ids_to_reconcile: set[UUID] = set()
522 for v2_limit in v2_limits:
523 # Find holders for this limit
524 holders_with_leases: list[
525 tuple[UUID, ConcurrencyLeaseHolder]
526 ] = await lease_storage.list_holders_for_limit(
527 limit_id=v2_limit.id,
528 )
529 # Find leases that belong to this task run
530 for lease_id, holder in holders_with_leases:
531 if holder.id == context.run.id:
532 lease_ids_to_reconcile.add(lease_id)
534 # Reconcile all found leases
535 for lease_id in lease_ids_to_reconcile:
536 try:
537 await _release_concurrency_lease(
538 session=context.session, lease_id=lease_id
539 )
540 except Exception:
541 logger.warning(
542 f"Failed to reconcile lease {lease_id} during release",
543 exc_info=True,
544 )
545 raise
547 v1_limits = (
548 await concurrency_limits.filter_concurrency_limits_for_orchestration(
549 context.session, tags=context.run.tags
550 )
551 )
552 for cl in v1_limits:
553 active_slots = set(cl.active_slots)
554 active_slots.discard(str(context.run.id))
555 cl.active_slots = list(active_slots)
558class SecureFlowConcurrencySlots(FlowRunOrchestrationRule):
559 """
560 Enforce deployment concurrency limits.
562 This rule enforces concurrency limits on deployments. If a deployment has a concurrency limit,
563 this rule will prevent more than that number of flow runs from being submitted concurrently
564 based on the concurrency limit behavior configured for the deployment.
566 We use the PENDING state as the target transition because this allows workers to secure a slot
567 before provisioning dynamic infrastructure to run a flow. If a slot isn't available, the worker
568 won't provision infrastructure.
570 A lease is created for the concurrency limit. The client will be responsible for maintaining the lease.
571 """
573 FROM_STATES = ALL_ORCHESTRATION_STATES - {
574 states.StateType.PENDING,
575 states.StateType.RUNNING,
576 states.StateType.CANCELLING,
577 }
578 TO_STATES = {states.StateType.PENDING}
580 async def before_transition(
581 self,
582 initial_state: states.State[Any] | None,
583 proposed_state: states.State[Any] | None,
584 context: FlowOrchestrationContext,
585 ) -> None:
586 self._acquired_deployment_concurrency_limit_id: UUID | None = None
587 self._acquired_deployment_concurrency_lease_id: UUID | None = None
589 if ( 589 ↛ 598line 589 didn't jump to line 598 because the condition on line 589 was always true
590 not context.session
591 or not context.run.deployment_id
592 or not proposed_state
593 or context.client_version
594 in WORKER_VERSIONS_THAT_MANAGE_DEPLOYMENT_CONCURRENCY
595 ):
596 return
598 deployment = await deployments.read_deployment(
599 session=context.session,
600 deployment_id=context.run.deployment_id,
601 )
602 if not deployment:
603 await self.abort_transition("Deployment not found.")
604 return
606 if (
607 not deployment.global_concurrency_limit
608 or not deployment.concurrency_limit_id
609 ):
610 return
612 if deployment.global_concurrency_limit.limit == 0:
613 await self.abort_transition(
614 "The deployment concurrency limit is 0. The flow will deadlock if submitted again."
615 )
616 return
618 acquired = await concurrency_limits_v2.bulk_increment_active_slots(
619 session=context.session,
620 concurrency_limit_ids=[deployment.concurrency_limit_id],
621 slots=1,
622 )
623 if acquired:
624 lease_storage = get_concurrency_lease_storage()
625 settings = get_current_settings()
627 concurrency_options = deployment.concurrency_options
628 grace_period = None
629 if concurrency_options is not None:
630 if isinstance(concurrency_options, dict):
631 concurrency_options = core.ConcurrencyOptions.model_validate(
632 concurrency_options
633 )
634 grace_period = concurrency_options.grace_period_seconds
635 # Fall back to server setting if grace_period_seconds is not explicitly set
636 if grace_period is None:
637 grace_period = (
638 settings.server.concurrency.initial_deployment_lease_duration
639 )
641 lease = await lease_storage.create_lease(
642 resource_ids=[deployment.concurrency_limit_id],
643 metadata=ConcurrencyLimitLeaseMetadata(
644 slots=1,
645 ),
646 ttl=datetime.timedelta(seconds=grace_period),
647 )
648 proposed_state.state_details.deployment_concurrency_lease_id = lease.id
649 self._acquired_deployment_concurrency_limit_id = (
650 deployment.concurrency_limit_id
651 )
652 self._acquired_deployment_concurrency_lease_id = lease.id
654 else:
655 concurrency_options = (
656 deployment.concurrency_options
657 or core.ConcurrencyOptions(
658 collision_strategy=core.ConcurrencyLimitStrategy.ENQUEUE
659 )
660 )
662 if (
663 concurrency_options.collision_strategy
664 == core.ConcurrencyLimitStrategy.ENQUEUE
665 ):
666 settings = get_current_settings()
667 await self.reject_transition(
668 state=states.Scheduled(
669 name="AwaitingConcurrencySlot",
670 scheduled_time=now("UTC")
671 + datetime.timedelta(
672 seconds=settings.server.deployments.concurrency_slot_wait_seconds
673 ),
674 ),
675 reason="Deployment concurrency limit reached.",
676 )
677 elif (
678 concurrency_options.collision_strategy
679 == core.ConcurrencyLimitStrategy.CANCEL_NEW
680 ):
681 await self.reject_transition(
682 state=states.Cancelled(
683 message="Deployment concurrency limit reached."
684 ),
685 reason="Deployment concurrency limit reached.",
686 )
688 async def cleanup( # type: ignore
689 self,
690 initial_state: states.State[Any] | None,
691 validated_state: states.State[Any] | None,
692 context: FlowOrchestrationContext,
693 ) -> None:
694 concurrency_limit_id = self._acquired_deployment_concurrency_limit_id
695 lease_id = self._acquired_deployment_concurrency_lease_id
696 if not context.session or not concurrency_limit_id: 696 ↛ 699line 696 didn't jump to line 699 because the condition on line 696 was always true
697 return
699 if lease_id:
700 await _release_concurrency_lease(
701 session=context.session,
702 lease_id=lease_id,
703 fallback_concurrency_limit_ids=[concurrency_limit_id],
704 )
705 if validated_state:
706 validated_state.state_details.deployment_concurrency_lease_id = None
707 else:
708 slots_released = await concurrency_limits_v2.bulk_decrement_active_slots(
709 session=context.session,
710 concurrency_limit_ids=[concurrency_limit_id],
711 slots=1,
712 )
713 if not slots_released:
714 raise RuntimeError(
715 "Failed to release deployment concurrency slots during cleanup"
716 )
719class ValidateDeploymentConcurrencyAtRunning(FlowRunOrchestrationRule):
720 """
721 Validates and renews deployment concurrency leases at the PENDING→RUNNING transition.
723 This prevents concurrency violations that occur when the lease reaper reclaims slots
724 from PENDING flows. Without this validation, a flow can lose its slot while provisioning
725 infrastructure and still transition to RUNNING, violating the concurrency limit.
726 """
728 FROM_STATES = {states.StateType.PENDING}
729 TO_STATES = {states.StateType.RUNNING}
731 async def before_transition(
732 self,
733 initial_state: states.State[Any] | None,
734 proposed_state: states.State[Any] | None,
735 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
736 ) -> None:
737 if initial_state is None or proposed_state is None:
738 return
740 # Copy lease ID to proposed state to maintain it through the transition
741 lease_id = initial_state.state_details.deployment_concurrency_lease_id
742 proposed_state.state_details.deployment_concurrency_lease_id = lease_id
744 if not lease_id:
745 return
747 # Only validate leases for clients that can maintain them (3.4.11+)
748 # Older clients don't renew leases, so validating would cause false failures
749 client_version = context.client_version or Version("2.0.0")
750 if isinstance(client_version, str):
751 client_version = Version(client_version)
752 if client_version < MIN_CLIENT_VERSION_FOR_CONCURRENCY_LIMIT_LEASING:
753 return
755 # Need deployment context to validate and potentially re-acquire slots
756 if (
757 not context.session
758 or not isinstance(context, FlowOrchestrationContext)
759 or not context.run.deployment_id
760 ):
761 return
763 deployment = await deployments.read_deployment(
764 session=context.session,
765 deployment_id=context.run.deployment_id,
766 )
768 if not deployment or not deployment.global_concurrency_limit:
769 return
771 concurrency_options = deployment.concurrency_options
772 if concurrency_options is not None:
773 if isinstance(concurrency_options, dict):
774 concurrency_options = core.ConcurrencyOptions.model_validate(
775 concurrency_options
776 )
777 grace_period = concurrency_options.grace_period_seconds
778 else:
779 grace_period = None
781 # Fall back to server setting if grace_period_seconds is not explicitly set
782 if grace_period is None:
783 settings = get_current_settings()
784 grace_period = settings.server.concurrency.initial_deployment_lease_duration
786 # Attempt atomic renewal to prevent race conditions where the lease
787 # exists but hasn't expired yet
788 lease_storage = get_concurrency_lease_storage()
789 renewed = await lease_storage.renew_lease(
790 lease_id=lease_id,
791 ttl=datetime.timedelta(seconds=grace_period),
792 )
794 if renewed:
795 return
797 # Lease was reaped - attempt re-acquisition for resilience. If no other
798 # flows need the slot, allow this flow to continue rather than failing it
799 slots_acquired = await concurrency_limits_v2.bulk_increment_active_slots(
800 session=context.session,
801 concurrency_limit_ids=[deployment.concurrency_limit_id],
802 slots=1,
803 )
805 if slots_acquired:
806 # Slot available - create new lease and continue. This provides
807 # resilience when a flow's lease expires but no other flows are competing
808 await (
809 context.session.flush()
810 ) # Ensure DB update is visible before creating lease
811 try:
812 new_lease = await lease_storage.create_lease(
813 resource_ids=[deployment.concurrency_limit_id],
814 ttl=datetime.timedelta(seconds=grace_period),
815 metadata=ConcurrencyLimitLeaseMetadata(
816 slots=1,
817 holder=ConcurrencyLeaseHolder(
818 type="flow_run", id=str(context.run.id)
819 ),
820 ),
821 )
822 proposed_state.state_details.deployment_concurrency_lease_id = (
823 new_lease.id
824 )
825 return
826 except Exception:
827 # Lease creation failed - release the slot we acquired
828 await concurrency_limits_v2.bulk_decrement_active_slots(
829 session=context.session,
830 concurrency_limit_ids=[deployment.concurrency_limit_id],
831 slots=1,
832 )
833 raise
835 # No slots available. Must cancel because infrastructure is already
836 # provisioned and we cannot reschedule at this point
837 await self.reject_transition(
838 state=states.Cancelled(
839 message="Deployment concurrency slot lost during provisioning - "
840 "no slots available to continue execution"
841 ),
842 reason="Deployment concurrency limit reached after lease expiry.",
843 )
846class RemoveDeploymentConcurrencyLeaseForOldClientVersions(FlowRunOrchestrationRule):
847 """
848 Removes a deployment concurrency lease if the client version is less than the minimum version for leasing.
849 """
851 FROM_STATES = {states.StateType.PENDING}
852 TO_STATES = {states.StateType.RUNNING, states.StateType.CANCELLING}
854 async def after_transition(
855 self,
856 initial_state: states.State[Any] | None,
857 validated_state: states.State[Any] | None,
858 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
859 ) -> None:
860 if not initial_state or (
861 context.client_version
862 and Version(context.client_version)
863 >= MIN_CLIENT_VERSION_FOR_CONCURRENCY_LIMIT_LEASING
864 ):
865 return
867 if lease_id := initial_state.state_details.deployment_concurrency_lease_id:
868 lease_storage = get_concurrency_lease_storage()
869 await lease_storage.revoke_lease(
870 lease_id=lease_id,
871 )
874class ReleaseFlowConcurrencySlots(FlowRunUniversalTransform):
875 """
876 Releases deployment concurrency slots held by a flow run.
878 This rule releases a concurrency slot for a deployment when a flow run
879 transitions out of the Running or Cancelling state.
880 """
882 async def after_transition(
883 self,
884 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
885 ) -> None:
886 if self.nullified_transition():
887 return
889 initial_state_type = (
890 context.initial_state.type if context.initial_state else None
891 )
892 proposed_state_type = (
893 context.proposed_state.type if context.proposed_state else None
894 )
896 # Check if the transition is valid for releasing concurrency slots.
897 # This should happen within `after_transition` because BaseUniversalTransforms
898 # don't know how to "fizzle" themselves if they encounter a transition that
899 # shouldn't apply to them, even if they use FROM_STATES and TO_STATES.
900 if not (
901 initial_state_type
902 in {
903 states.StateType.RUNNING,
904 states.StateType.CANCELLING,
905 states.StateType.PENDING,
906 }
907 and proposed_state_type
908 not in {
909 states.StateType.PENDING,
910 states.StateType.RUNNING,
911 states.StateType.CANCELLING,
912 }
913 ):
914 return
915 if not context.session or not context.run.deployment_id:
916 return
918 if ( 918 ↛ 922line 918 didn't jump to line 922 because the condition on line 918 was never true
919 context.initial_state
920 and context.initial_state.state_details.deployment_concurrency_lease_id
921 ):
922 fallback_concurrency_limit_ids = None
923 deployment = await deployments.read_deployment(
924 session=context.session,
925 deployment_id=context.run.deployment_id,
926 )
927 if deployment and deployment.concurrency_limit_id:
928 fallback_concurrency_limit_ids = [deployment.concurrency_limit_id]
930 await _release_concurrency_lease(
931 session=context.session,
932 lease_id=context.initial_state.state_details.deployment_concurrency_lease_id,
933 fallback_concurrency_limit_ids=fallback_concurrency_limit_ids,
934 )
935 return
937 deployment = await deployments.read_deployment(
938 session=context.session,
939 deployment_id=context.run.deployment_id,
940 )
941 if not deployment or not deployment.concurrency_limit_id:
942 return
944 slots_released = await concurrency_limits_v2.bulk_decrement_active_slots(
945 session=context.session,
946 concurrency_limit_ids=[deployment.concurrency_limit_id],
947 slots=1,
948 )
949 if not slots_released:
950 raise RuntimeError("Failed to release deployment concurrency slots")
953class CacheInsertion(TaskRunOrchestrationRule):
954 """
955 Caches completed states with cache keys after they are validated.
956 """
958 FROM_STATES = ALL_ORCHESTRATION_STATES
959 TO_STATES = {StateType.COMPLETED}
961 async def before_transition(
962 self,
963 initial_state: states.State[Any] | None,
964 proposed_state: states.State[Any] | None,
965 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
966 ) -> None:
967 if proposed_state is None:
968 return
970 settings = get_current_settings()
971 cache_key = proposed_state.state_details.cache_key
972 if cache_key and len(cache_key) > settings.server.tasks.max_cache_key_length:
973 await self.reject_transition(
974 state=proposed_state,
975 reason=f"Cache key exceeded maximum allowed length of {settings.server.tasks.max_cache_key_length} characters.",
976 )
977 return
979 @db_injector
980 async def after_transition(
981 self,
982 db: PrefectDBInterface,
983 initial_state: states.State[Any] | None,
984 validated_state: states.State[Any] | None,
985 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
986 ) -> None:
987 if not validated_state or not context.session:
988 return
990 cache_key = validated_state.state_details.cache_key
991 if cache_key:
992 new_cache_item = db.TaskRunStateCache(
993 cache_key=cache_key,
994 cache_expiration=validated_state.state_details.cache_expiration,
995 task_run_state_id=validated_state.id,
996 )
997 context.session.add(new_cache_item)
1000class CacheRetrieval(TaskRunOrchestrationRule):
1001 """
1002 Rejects running states if a completed state has been cached.
1004 This rule rejects transitions into a running state with a cache key if the key
1005 has already been associated with a completed state in the cache table. The client
1006 will be instructed to transition into the cached completed state instead.
1007 """
1009 FROM_STATES = ALL_ORCHESTRATION_STATES
1010 TO_STATES = {StateType.RUNNING}
1012 @db_injector
1013 async def before_transition(
1014 self,
1015 db: PrefectDBInterface,
1016 initial_state: states.State[Any] | None,
1017 proposed_state: states.State[Any] | None,
1018 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1019 ) -> None:
1020 if not proposed_state:
1021 return
1023 cache_key = proposed_state.state_details.cache_key
1024 if cache_key and not proposed_state.state_details.refresh_cache:
1025 # Check for cached states matching the cache key
1026 cached_state_id = (
1027 select(db.TaskRunStateCache.task_run_state_id)
1028 .where(
1029 sa.and_(
1030 db.TaskRunStateCache.cache_key == cache_key,
1031 sa.or_(
1032 db.TaskRunStateCache.cache_expiration.is_(None),
1033 db.TaskRunStateCache.cache_expiration > now("UTC"),
1034 ),
1035 ),
1036 )
1037 .order_by(db.TaskRunStateCache.created.desc())
1038 .limit(1)
1039 ).scalar_subquery()
1040 query = select(db.TaskRunState).where(db.TaskRunState.id == cached_state_id)
1041 cached_state = (await context.session.execute(query)).scalar()
1042 if cached_state:
1043 new_state = cached_state.as_state().fresh_copy()
1044 new_state.name = "Cached"
1045 await self.reject_transition(
1046 state=new_state, reason="Retrieved state from cache"
1047 )
1050class RetryFailedFlows(FlowRunOrchestrationRule):
1051 """
1052 Rejects failed states and schedules a retry if the retry limit has not been reached.
1054 This rule rejects transitions into a failed state if `retries` has been
1055 set and the run count has not reached the specified limit. The client will be
1056 instructed to transition into a scheduled state to retry flow execution.
1057 """
1059 FROM_STATES = {StateType.RUNNING}
1060 TO_STATES = {StateType.FAILED}
1062 async def before_transition(
1063 self,
1064 initial_state: states.State[Any] | None,
1065 proposed_state: states.State[Any] | None,
1066 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1067 ) -> None:
1068 if initial_state is None or proposed_state is None: 1068 ↛ 1069line 1068 didn't jump to line 1069 because the condition on line 1068 was never true
1069 return
1071 run_settings = context.run_settings
1072 run_count = context.run.run_count
1074 if run_settings.retries is None or run_count > run_settings.retries: 1074 ↛ 1082line 1074 didn't jump to line 1082 because the condition on line 1074 was always true
1075 # Clear retry type to allow for future infrastructure level retries (e.g. via the UI)
1076 updated_policy = context.run.empirical_policy.model_dump()
1077 updated_policy["retry_type"] = None
1078 context.run.empirical_policy = core.FlowRunPolicy(**updated_policy)
1080 return # Retry count exceeded, allow transition to failed
1082 scheduled_start_time = now("UTC") + datetime.timedelta(
1083 seconds=run_settings.retry_delay or 0
1084 )
1086 # support old-style flow run retries for older clients
1087 # older flow retries require us to loop over failed tasks to update their state
1088 # this is not required after API version 0.8.3
1089 api_version = context.parameters.get("api-version", None)
1090 if api_version and api_version < Version("0.8.3"):
1091 failed_task_runs = await models.task_runs.read_task_runs(
1092 context.session,
1093 flow_run_filter=filters.FlowRunFilter(
1094 id=filters.FlowRunFilterId(any_=[context.run.id])
1095 ),
1096 task_run_filter=filters.TaskRunFilter(
1097 state=filters.TaskRunFilterState(
1098 type=filters.TaskRunFilterStateType(any_=[StateType.FAILED])
1099 )
1100 ),
1101 )
1102 for run in failed_task_runs:
1103 await models.task_runs.set_task_run_state(
1104 context.session,
1105 run.id,
1106 state=states.AwaitingRetry(scheduled_time=scheduled_start_time),
1107 force=True,
1108 )
1109 # Reset the run count so that the task run retries still work correctly
1110 run.run_count = 0
1112 # Reset pause metadata on retry
1113 # Pauses as a concept only exist after API version 0.8.4
1114 api_version = context.parameters.get("api-version", None)
1115 if api_version is None or api_version >= Version("0.8.4"):
1116 updated_policy = context.run.empirical_policy.model_dump()
1117 updated_policy["resuming"] = False
1118 updated_policy["pause_keys"] = set()
1119 updated_policy["retry_type"] = "in_process"
1120 context.run.empirical_policy = core.FlowRunPolicy(**updated_policy)
1122 # Generate a new state for the flow
1123 retry_state = states.AwaitingRetry(
1124 scheduled_time=scheduled_start_time,
1125 message=proposed_state.message,
1126 data=proposed_state.data,
1127 )
1128 await self.reject_transition(state=retry_state, reason="Retrying")
1131class RetryFailedTasks(TaskRunOrchestrationRule):
1132 """
1133 Rejects failed states and schedules a retry if the retry limit has not been reached.
1135 This rule rejects transitions into a failed state if `retries` has been
1136 set, the run count has not reached the specified limit, and the client
1137 asserts it is a retriable task run. The client will be instructed to
1138 transition into a scheduled state to retry task execution.
1139 """
1141 FROM_STATES = {StateType.RUNNING}
1142 TO_STATES = {StateType.FAILED}
1144 async def before_transition(
1145 self,
1146 initial_state: states.State[Any] | None,
1147 proposed_state: states.State[Any] | None,
1148 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1149 ) -> None:
1150 if initial_state is None or proposed_state is None:
1151 return
1153 run_settings = context.run_settings
1154 run_count = context.run.run_count
1155 delay = run_settings.retry_delay
1157 if isinstance(delay, list):
1158 base_delay = delay[min(run_count - 1, len(delay) - 1)]
1159 else:
1160 base_delay = delay or 0
1162 # guard against negative relative jitter inputs; also guard against
1163 # base_delay == 0: clamped_poisson_interval(0, ...) divides by zero.
1164 if run_settings.retry_jitter_factor and base_delay > 0:
1165 delay = clamped_poisson_interval(
1166 base_delay, clamping_factor=run_settings.retry_jitter_factor
1167 )
1168 else:
1169 delay = base_delay
1171 # set by user to conditionally retry a task using @task(retry_condition_fn=...)
1172 if getattr(proposed_state.state_details, "retriable", True) is False:
1173 return
1175 if run_settings.retries is not None and run_count <= run_settings.retries:
1176 retry_state = states.AwaitingRetry(
1177 scheduled_time=now("UTC") + datetime.timedelta(seconds=delay),
1178 message=proposed_state.message,
1179 data=proposed_state.data,
1180 )
1181 await self.reject_transition(state=retry_state, reason="Retrying")
1184class EnqueueScheduledTasks(TaskRunOrchestrationRule):
1185 """
1186 Enqueues background task runs when they are scheduled
1187 """
1189 FROM_STATES = ALL_ORCHESTRATION_STATES
1190 TO_STATES = {StateType.SCHEDULED}
1192 async def after_transition(
1193 self,
1194 initial_state: states.State[Any] | None,
1195 validated_state: states.State[Any] | None,
1196 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1197 ) -> None:
1198 if not validated_state:
1199 # Only if the transition was valid
1200 return
1202 if not validated_state.state_details.deferred:
1203 # Only for tasks that are deferred
1204 return
1206 task_run: core.TaskRun = core.TaskRun.model_validate(context.run)
1207 queue: TaskQueue = TaskQueue.for_key(task_run.task_key)
1209 if validated_state.name == "AwaitingRetry":
1210 await queue.retry(task_run)
1211 else:
1212 await queue.enqueue(task_run)
1215class RenameReruns(GenericOrchestrationRule):
1216 """
1217 Name the states if they have run more than once.
1219 In the special case where the initial state is an "AwaitingRetry" scheduled state,
1220 the proposed state will be renamed to "Retrying" instead.
1221 """
1223 FROM_STATES = ALL_ORCHESTRATION_STATES
1224 TO_STATES = {StateType.RUNNING}
1226 async def before_transition(
1227 self,
1228 initial_state: states.State[Any] | None,
1229 proposed_state: states.State[Any] | None,
1230 context: OrchestrationContext[
1231 orm_models.Run, core.TaskRunPolicy | core.FlowRunPolicy
1232 ],
1233 ) -> None:
1234 if initial_state is None or proposed_state is None:
1235 return
1237 run_count = context.run.run_count
1238 if run_count > 0:
1239 if initial_state.name == "AwaitingRetry":
1240 await self.rename_state("Retrying")
1241 else:
1242 await self.rename_state("Rerunning")
1245class CopyScheduledTime(
1246 BaseOrchestrationRule[orm_models.Run, Union[core.TaskRunPolicy, core.FlowRunPolicy]]
1247):
1248 """
1249 Ensures scheduled time is copied from scheduled states to pending states.
1251 If a new scheduled time has been proposed on the pending state, the scheduled time
1252 on the scheduled state will be ignored.
1253 """
1255 FROM_STATES = {StateType.SCHEDULED}
1256 TO_STATES = {StateType.PENDING}
1258 async def before_transition(
1259 self,
1260 initial_state: states.State[Any] | None,
1261 proposed_state: states.State[Any] | None,
1262 context: OrchestrationContext[
1263 orm_models.Run, core.TaskRunPolicy | core.FlowRunPolicy
1264 ],
1265 ) -> None:
1266 if initial_state is None or proposed_state is None:
1267 return
1269 if not proposed_state.state_details.scheduled_time:
1270 proposed_state.state_details.scheduled_time = (
1271 initial_state.state_details.scheduled_time
1272 )
1275class WaitForScheduledTime(
1276 BaseOrchestrationRule[orm_models.Run, Union[core.TaskRunPolicy, core.FlowRunPolicy]]
1277):
1278 """
1279 Prevents transitions to running states from happening too early.
1281 This rule enforces that all scheduled states will only start with the machine clock
1282 used by the Prefect REST API instance. This rule will identify transitions from scheduled
1283 states that are too early and nullify them. Instead, no state will be written to the
1284 database and the client will be sent an instruction to wait for `delay_seconds`
1285 before attempting the transition again.
1286 """
1288 FROM_STATES = {StateType.SCHEDULED, StateType.PENDING}
1289 TO_STATES = {StateType.RUNNING}
1291 async def before_transition(
1292 self,
1293 initial_state: states.State[Any] | None,
1294 proposed_state: states.State[Any] | None,
1295 context: OrchestrationContext[
1296 orm_models.Run, core.TaskRunPolicy | core.FlowRunPolicy
1297 ],
1298 ) -> None:
1299 if initial_state is None or proposed_state is None:
1300 return
1302 scheduled_time = initial_state.state_details.scheduled_time
1303 if not scheduled_time:
1304 return
1306 # At this moment, we round delay to the nearest second as the API schema
1307 # specifies an integer return value.
1308 delay = scheduled_time - now("UTC")
1309 delay_seconds = math.floor(delay.total_seconds())
1310 delay_seconds += round(delay.microseconds / 1e6)
1311 if delay_seconds > 0:
1312 await self.delay_transition(
1313 delay_seconds, reason="Scheduled time is in the future"
1314 )
1317class CopyTaskParametersID(TaskRunOrchestrationRule):
1318 """
1319 Ensures a task's parameters ID is copied from Scheduled to Pending and from
1320 Pending to Running states.
1322 If a parameters ID has been included on the proposed state, the parameters ID
1323 on the initial state will be ignored.
1324 """
1326 FROM_STATES = {StateType.SCHEDULED, StateType.PENDING}
1327 TO_STATES = {StateType.PENDING, StateType.RUNNING}
1329 async def before_transition(
1330 self,
1331 initial_state: states.State[Any] | None,
1332 proposed_state: states.State[Any] | None,
1333 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1334 ) -> None:
1335 if initial_state is None or proposed_state is None:
1336 return
1338 if not proposed_state.state_details.task_parameters_id:
1339 proposed_state.state_details.task_parameters_id = (
1340 initial_state.state_details.task_parameters_id
1341 )
1344class HandlePausingFlows(FlowRunOrchestrationRule):
1345 """
1346 Governs runs attempting to enter a Paused/Suspended state
1347 """
1349 FROM_STATES = ALL_ORCHESTRATION_STATES
1350 TO_STATES = {StateType.PAUSED}
1352 async def before_transition(
1353 self,
1354 initial_state: states.State[Any] | None,
1355 proposed_state: states.State[Any] | None,
1356 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1357 ) -> None:
1358 if proposed_state is None:
1359 return
1361 verb = "suspend" if proposed_state.name == "Suspended" else "pause"
1363 if initial_state is None:
1364 await self.abort_transition(f"Cannot {verb} flows with no state.")
1365 return
1367 if not initial_state.is_running():
1368 await self.reject_transition(
1369 state=None,
1370 reason=f"Cannot {verb} flows that are not currently running.",
1371 )
1372 return
1374 self.key = proposed_state.state_details.pause_key
1375 if self.key is None:
1376 # if no pause key is provided, default to a UUID
1377 self.key = str(uuid4())
1379 pause_keys = context.run.empirical_policy.pause_keys or set()
1380 if self.key in pause_keys:
1381 await self.reject_transition(
1382 state=None, reason=f"This {verb} has already fired."
1383 )
1384 return
1386 if proposed_state.state_details.pause_reschedule:
1387 if context.run.parent_task_run_id:
1388 await self.abort_transition(
1389 reason=f"Cannot {verb} subflows.",
1390 )
1391 return
1393 if context.run.deployment_id is None:
1394 await self.abort_transition(
1395 reason=f"Cannot {verb} flows without a deployment.",
1396 )
1397 return
1399 async def after_transition(
1400 self,
1401 initial_state: states.State[Any] | None,
1402 validated_state: states.State[Any] | None,
1403 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1404 ) -> None:
1405 updated_policy = context.run.empirical_policy.model_dump()
1406 updated_policy["pause_keys"].add(self.key)
1407 context.run.empirical_policy = core.FlowRunPolicy(**updated_policy)
1410class HandleResumingPausedFlows(FlowRunOrchestrationRule):
1411 """
1412 Governs runs attempting to leave a Paused state
1413 """
1415 FROM_STATES = {StateType.PAUSED}
1416 TO_STATES = ALL_ORCHESTRATION_STATES
1418 async def before_transition(
1419 self,
1420 initial_state: states.State[Any] | None,
1421 proposed_state: states.State[Any] | None,
1422 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1423 ) -> None:
1424 if initial_state is None or proposed_state is None: 1424 ↛ 1425line 1424 didn't jump to line 1425 because the condition on line 1424 was never true
1425 return
1427 if not (
1428 proposed_state
1429 and (
1430 proposed_state.is_running()
1431 or proposed_state.is_scheduled()
1432 or proposed_state.is_final()
1433 )
1434 ):
1435 await self.reject_transition(
1436 state=None,
1437 reason=(
1438 f"This run cannot transition to the {proposed_state.type} state"
1439 f" from the {initial_state.type} state."
1440 ),
1441 )
1442 return
1444 verb = "suspend" if proposed_state.name == "Suspended" else "pause"
1446 display_state_name = (
1447 proposed_state.name.lower()
1448 if proposed_state.name
1449 else proposed_state.type.value.lower()
1450 )
1452 if initial_state.state_details.pause_reschedule: 1452 ↛ 1453line 1452 didn't jump to line 1453 because the condition on line 1452 was never true
1453 if not context.run.deployment_id:
1454 await self.reject_transition(
1455 state=None,
1456 reason=(
1457 f"Cannot reschedule a {display_state_name} flow run"
1458 " without a deployment."
1459 ),
1460 )
1461 return
1462 pause_timeout = initial_state.state_details.pause_timeout
1463 if pause_timeout and pause_timeout < now("UTC"): 1463 ↛ 1464line 1463 didn't jump to line 1464 because the condition on line 1463 was never true
1464 pause_timeout_failure = states.Failed(
1465 message=(f"The flow was {display_state_name} and never resumed."),
1466 )
1467 await self.reject_transition(
1468 state=pause_timeout_failure,
1469 reason=f"The flow run {verb} has timed out and can no longer resume.",
1470 )
1471 return
1473 async def after_transition(
1474 self,
1475 initial_state: states.State[Any] | None,
1476 validated_state: states.State[Any] | None,
1477 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1478 ) -> None:
1479 updated_policy = context.run.empirical_policy.model_dump()
1480 updated_policy["resuming"] = True
1481 context.run.empirical_policy = core.FlowRunPolicy(**updated_policy)
1484class UpdateFlowRunTrackerOnTasks(TaskRunOrchestrationRule):
1485 """
1486 Tracks the flow run attempt a task run state is associated with.
1487 """
1489 FROM_STATES = ALL_ORCHESTRATION_STATES
1490 TO_STATES = {StateType.RUNNING}
1492 async def after_transition(
1493 self,
1494 initial_state: states.State[Any] | None,
1495 validated_state: states.State[Any] | None,
1496 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1497 ) -> None:
1498 if context.run.flow_run_id is not None:
1499 self.flow_run: orm_models.FlowRun | None = await context.flow_run()
1500 if self.flow_run:
1501 context.run.flow_run_run_count = self.flow_run.run_count
1502 else:
1503 raise ObjectNotFoundError(
1504 (
1505 "Unable to read flow run associated with task run:"
1506 f" {context.run.id}, this flow run might have been deleted"
1507 ),
1508 )
1511class HandleTaskTerminalStateTransitions(TaskRunOrchestrationRule):
1512 """
1513 We do not allow tasks to leave terminal states if:
1514 - The task is completed and has a persisted result
1515 - The task is going to CANCELLING / PAUSED / CRASHED
1517 We reset the run count when a task leaves a terminal state for a non-terminal state
1518 which resets task run retries; this is particularly relevant for flow run retries.
1519 """
1521 FROM_STATES: set[states.StateType | None] = TERMINAL_STATES # pyright: ignore[reportAssignmentType] technically TERMINAL_STATES doesn't contain None
1522 TO_STATES: set[states.StateType | None] = ALL_ORCHESTRATION_STATES
1524 async def before_transition(
1525 self,
1526 initial_state: states.State[Any] | None,
1527 proposed_state: states.State[Any] | None,
1528 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1529 ) -> None:
1530 if initial_state is None or proposed_state is None: 1530 ↛ 1531line 1530 didn't jump to line 1531 because the condition on line 1530 was never true
1531 return
1533 self.original_run_count: int = context.run.run_count
1535 # Do not allow runs to be marked as crashed, paused, or cancelling if already terminal
1536 if proposed_state.type in { 1536 ↛ 1541line 1536 didn't jump to line 1541 because the condition on line 1536 was never true
1537 StateType.CANCELLING,
1538 StateType.PAUSED,
1539 StateType.CRASHED,
1540 }:
1541 await self.abort_transition(f"Run is already {initial_state.type.value}.")
1542 return
1544 # Only allow departure from a happily completed state if the result is not persisted
1545 if ( 1545 ↛ 1550line 1545 didn't jump to line 1550 because the condition on line 1545 was never true
1546 initial_state.is_completed()
1547 and initial_state.data
1548 and initial_state.data.get("type") != "unpersisted"
1549 ):
1550 await self.reject_transition(None, "This run is already completed.")
1551 return
1553 if not proposed_state.is_final(): 1553 ↛ 1555line 1553 didn't jump to line 1555 because the condition on line 1553 was never true
1554 # Reset run count to reset retries
1555 context.run.run_count = 0
1557 # Change the name of the state to retrying if its a flow run retry
1558 if proposed_state.is_running() and context.run.flow_run_id is not None: 1558 ↛ 1559line 1558 didn't jump to line 1559 because the condition on line 1558 was never true
1559 self.flow_run: orm_models.FlowRun | None = await context.flow_run()
1560 if self.flow_run is not None:
1561 flow_retrying = context.run.flow_run_run_count < self.flow_run.run_count
1562 if flow_retrying:
1563 await self.rename_state("Retrying")
1565 async def cleanup(
1566 self,
1567 initial_state: states.State[Any] | None,
1568 validated_state: states.State[Any] | None,
1569 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1570 ) -> None:
1571 # reset run count
1572 context.run.run_count = self.original_run_count
1575class HandleFlowTerminalStateTransitions(FlowRunOrchestrationRule):
1576 """
1577 We do not allow flows to leave terminal states if:
1578 - The flow is completed and has a persisted result
1579 - The flow is going to CANCELLING / PAUSED / CRASHED
1580 - The flow is going to scheduled and has no deployment
1582 We reset the pause metadata when a flow leaves a terminal state for a non-terminal
1583 state. This resets pause behavior during manual flow run retries.
1584 """
1586 FROM_STATES: set[states.StateType | None] = TERMINAL_STATES # pyright: ignore[reportAssignmentType] technically TERMINAL_STATES doesn't contain None
1587 TO_STATES: set[states.StateType | None] = ALL_ORCHESTRATION_STATES
1589 async def before_transition(
1590 self,
1591 initial_state: states.State[Any] | None,
1592 proposed_state: states.State[Any] | None,
1593 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1594 ) -> None:
1595 if initial_state is None or proposed_state is None: 1595 ↛ 1596line 1595 didn't jump to line 1596 because the condition on line 1595 was never true
1596 return
1598 self.original_flow_policy: dict[str, Any] = (
1599 context.run.empirical_policy.model_dump()
1600 )
1602 # Do not allow runs to be marked as crashed, paused, or cancelling if already terminal
1603 if proposed_state.type in {
1604 StateType.CANCELLING,
1605 StateType.PAUSED,
1606 StateType.CRASHED,
1607 }:
1608 await self.abort_transition(
1609 f"Run is already in terminal state {initial_state.type.value}."
1610 )
1611 return
1613 # Only allow departure from a happily completed state if the result is not
1614 # persisted and the a rerun is being proposed
1615 if ( 1615 ↛ 1621line 1615 didn't jump to line 1621 because the condition on line 1615 was never true
1616 initial_state.is_completed()
1617 and not proposed_state.is_final()
1618 and initial_state.data
1619 and _has_persisted_result(initial_state.data)
1620 ):
1621 await self.reject_transition(None, "Run is already COMPLETED.")
1622 return
1624 # Prevent COMPLETED → COMPLETED transitions that would discard result data.
1625 # See https://github.com/PrefectHQ/prefect/issues/21955
1626 if ( 1626 ↛ 1633line 1626 didn't jump to line 1633 because the condition on line 1626 was never true
1627 initial_state.is_completed()
1628 and proposed_state.is_completed()
1629 and initial_state.data
1630 and _has_persisted_result(initial_state.data)
1631 and not _has_persisted_result(proposed_state.data)
1632 ):
1633 await self.reject_transition(
1634 None,
1635 "Cannot overwrite a COMPLETED state that carries persisted result "
1636 "data with one that does not.",
1637 )
1638 return
1640 # Do not allows runs to be rescheduled without a deployment
1641 if proposed_state.is_scheduled() and not context.run.deployment_id: 1641 ↛ 1642line 1641 didn't jump to line 1642 because the condition on line 1641 was never true
1642 await self.abort_transition(
1643 "Cannot reschedule a run without an associated deployment."
1644 )
1645 return
1647 if not proposed_state.is_final():
1648 # Reset pause metadata when leaving a terminal state
1649 api_version = context.parameters.get("api-version", None)
1650 if api_version is None or api_version >= Version("0.8.4"): 1650 ↛ exitline 1650 didn't return from function 'before_transition' because the condition on line 1650 was always true
1651 updated_policy = context.run.empirical_policy.model_dump()
1652 updated_policy["resuming"] = False
1653 updated_policy["pause_keys"] = set()
1654 if proposed_state.is_scheduled(): 1654 ↛ 1655line 1654 didn't jump to line 1655 because the condition on line 1654 was never true
1655 updated_policy["retry_type"] = "reschedule"
1656 else:
1657 updated_policy["retry_type"] = None
1658 context.run.empirical_policy = core.FlowRunPolicy(**updated_policy)
1660 async def cleanup(
1661 self,
1662 initial_state: states.State[Any] | None,
1663 validated_state: states.State[Any] | None,
1664 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1665 ) -> None:
1666 context.run.empirical_policy = core.FlowRunPolicy(**self.original_flow_policy)
1669class PreventResultDataLoss(FlowRunOrchestrationRule):
1670 """Reject terminal-to-terminal transitions that would discard persisted result data.
1672 This is intentionally lightweight so it can be included in MinimalFlowPolicy
1673 (used by force=True transitions) without pulling in the full
1674 HandleFlowTerminalStateTransitions rule.
1676 See https://github.com/PrefectHQ/prefect/issues/21955
1677 """
1679 FROM_STATES: set[states.StateType | None] = TERMINAL_STATES # pyright: ignore[reportAssignmentType]
1680 TO_STATES: set[states.StateType | None] = TERMINAL_STATES # pyright: ignore[reportAssignmentType]
1682 async def before_transition(
1683 self,
1684 initial_state: states.State[Any] | None,
1685 proposed_state: states.State[Any] | None,
1686 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1687 ) -> None:
1688 if initial_state is None or proposed_state is None:
1689 return
1691 if (
1692 initial_state.is_completed()
1693 and proposed_state.is_completed()
1694 and initial_state.data
1695 and _has_persisted_result(initial_state.data)
1696 and not _has_persisted_result(proposed_state.data)
1697 ):
1698 await self.reject_transition(
1699 None,
1700 "Cannot overwrite a COMPLETED state that carries persisted result "
1701 "data with one that does not.",
1702 )
1703 return
1706class PreventPendingTransitions(GenericOrchestrationRule):
1707 """
1708 Prevents transitions to PENDING.
1710 This rule is only used for flow runs.
1712 This is intended to prevent race conditions during duplicate submissions of runs.
1713 Before a run is submitted to its execution environment, it should be placed in a
1714 PENDING state. If two workers attempt to submit the same run, one of them should
1715 encounter a PENDING -> PENDING transition and abort orchestration of the run.
1717 Similarly, if the execution environment starts quickly the run may be in a RUNNING
1718 state when the second worker attempts the PENDING transition. We deny these state
1719 changes as well to prevent duplicate submission. If a run has transitioned to a
1720 RUNNING state a worker should not attempt to submit it again unless it has moved
1721 into a terminal state.
1723 CANCELLING and CANCELLED runs should not be allowed to transition to PENDING.
1724 For re-runs of deployed runs, they should transition to SCHEDULED first.
1725 For re-runs of ad-hoc runs, they should transition directly to RUNNING.
1726 """
1728 FROM_STATES = {
1729 StateType.PENDING,
1730 StateType.CANCELLING,
1731 StateType.RUNNING,
1732 StateType.CANCELLED,
1733 }
1734 TO_STATES = {StateType.PENDING}
1736 async def before_transition(
1737 self,
1738 initial_state: states.State[Any] | None,
1739 proposed_state: states.State[Any] | None,
1740 context: OrchestrationContext[
1741 orm_models.Run, Union[core.FlowRunPolicy, core.TaskRunPolicy]
1742 ],
1743 ) -> None:
1744 if initial_state is None or proposed_state is None:
1745 return
1747 # Allow PENDING→PENDING transitions when the state name changes and
1748 # the proposed state is not the default "Pending" name. This enables
1749 # progression through named sub-states (e.g. Pending → Submitting →
1750 # InfrastructurePending) while still blocking a second worker from
1751 # re-proposing Pending after the run has already advanced.
1752 if (
1753 initial_state.type == StateType.PENDING
1754 and proposed_state.type == StateType.PENDING
1755 and initial_state.name != proposed_state.name
1756 and proposed_state.name != "Pending"
1757 ):
1758 # Carry forward state_details that were set by earlier
1759 # orchestration rules on the initial PENDING state.
1760 proposed_state.state_details.scheduled_time = (
1761 initial_state.state_details.scheduled_time
1762 )
1763 proposed_state.state_details.deployment_concurrency_lease_id = (
1764 initial_state.state_details.deployment_concurrency_lease_id
1765 )
1766 return
1768 await self.abort_transition(
1769 reason=(
1770 f"This run is in a {initial_state.type.name} state and cannot"
1771 " transition to a PENDING state."
1772 )
1773 )
1776class EnsureOnlyScheduledFlowsMarkedLate(FlowRunOrchestrationRule):
1777 FROM_STATES = ALL_ORCHESTRATION_STATES
1778 TO_STATES = {StateType.SCHEDULED}
1780 async def before_transition(
1781 self,
1782 initial_state: states.State[Any] | None,
1783 proposed_state: states.State[Any] | None,
1784 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1785 ) -> None:
1786 if initial_state is None or proposed_state is None:
1787 return
1789 marking_flow_late = (
1790 proposed_state.is_scheduled() and proposed_state.name == "Late"
1791 )
1792 if marking_flow_late:
1793 if not initial_state.is_scheduled():
1794 await self.reject_transition(
1795 state=None, reason="Only scheduled flows can be marked late."
1796 )
1797 elif initial_state.name == "Late":
1798 await self.reject_transition(
1799 state=None, reason="This flow run is already marked late."
1800 )
1803class EnforceDeploymentConcurrencyOnLate(FlowRunOrchestrationRule):
1804 """Enforce the CANCEL_NEW deployment concurrency strategy when marking runs late.
1806 When a flow run would be marked Late and its deployment uses the CANCEL_NEW
1807 collision strategy with a fully occupied concurrency limit, this rule rejects
1808 the Late transition and replaces it with a Cancelled state.
1810 This closes the gap where CANCEL_NEW is normally enforced at the * -> PENDING
1811 transition (by SecureFlowConcurrencySlots), but runs that never reach PENDING
1812 because they go late would accumulate in a Late state instead of being cancelled.
1813 """
1815 FROM_STATES = {StateType.SCHEDULED}
1816 TO_STATES = {StateType.SCHEDULED}
1818 async def before_transition(
1819 self,
1820 initial_state: states.State[Any] | None,
1821 proposed_state: states.State[Any] | None,
1822 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1823 ) -> None:
1824 if initial_state is None or proposed_state is None:
1825 return
1827 if not (proposed_state.is_scheduled() and proposed_state.name == "Late"):
1828 return
1830 if not context.run.deployment_id:
1831 return
1833 deployment = await deployments.read_deployment(
1834 session=context.session,
1835 deployment_id=context.run.deployment_id,
1836 )
1837 if not deployment or not deployment.concurrency_limit_id:
1838 return
1840 concurrency_options = deployment.concurrency_options
1841 if isinstance(concurrency_options, dict):
1842 concurrency_options = core.ConcurrencyOptions.model_validate(
1843 concurrency_options
1844 )
1845 if (
1846 not concurrency_options
1847 or concurrency_options.collision_strategy
1848 != core.ConcurrencyLimitStrategy.CANCEL_NEW
1849 ):
1850 return
1852 limit = deployment.global_concurrency_limit
1853 if not limit:
1854 return
1856 if limit.active_slots >= limit.limit:
1857 await self.reject_transition(
1858 state=states.Cancelled(message="Deployment concurrency limit reached."),
1859 reason=(
1860 "Deployment concurrency limit is full and uses the"
1861 " CANCEL_NEW strategy."
1862 ),
1863 )
1866class PreventRunningTasksFromStoppedFlows(TaskRunOrchestrationRule):
1867 """
1868 Prevents running tasks from stopped flows.
1870 A running state implies execution, but also the converse. This rule ensures that a
1871 flow's tasks cannot be run unless the flow is also running.
1872 """
1874 FROM_STATES = ALL_ORCHESTRATION_STATES
1875 TO_STATES = {StateType.RUNNING}
1877 async def before_transition(
1878 self,
1879 initial_state: states.State[Any] | None,
1880 proposed_state: states.State[Any] | None,
1881 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1882 ) -> None:
1883 flow_run = await context.flow_run()
1884 if flow_run is not None:
1885 if flow_run.state is None:
1886 await self.abort_transition(
1887 reason="The enclosing flow must be running to begin task execution."
1888 )
1889 elif flow_run.state.type == StateType.PAUSED:
1890 # Use the flow run's Paused state details to preserve data like
1891 # timeouts.
1892 paused_state = states.Paused(
1893 name="NotReady",
1894 pause_expiration_time=flow_run.state.state_details.pause_timeout,
1895 reschedule=flow_run.state.state_details.pause_reschedule,
1896 )
1897 await self.reject_transition(
1898 state=paused_state,
1899 reason=(
1900 "The flow is paused, new tasks can execute after resuming flow"
1901 f" run: {flow_run.id}."
1902 ),
1903 )
1904 elif not flow_run.state.type == StateType.RUNNING:
1905 # task runners should abort task run execution
1906 await self.abort_transition(
1907 reason=(
1908 "The enclosing flow must be running to begin task execution."
1909 ),
1910 )
1913class EnforceCancellingToCancelledTransition(TaskRunOrchestrationRule):
1914 """
1915 Rejects transitions from Cancelling to any terminal state except for Cancelled.
1916 """
1918 FROM_STATES = {StateType.CANCELLING}
1919 TO_STATES = ALL_ORCHESTRATION_STATES - {StateType.CANCELLED}
1921 async def before_transition(
1922 self,
1923 initial_state: states.State[Any] | None,
1924 proposed_state: states.State[Any] | None,
1925 context: OrchestrationContext[orm_models.TaskRun, core.TaskRunPolicy],
1926 ) -> None:
1927 await self.reject_transition(
1928 state=None,
1929 reason=(
1930 "Cannot transition flows that are cancelling to a state other "
1931 "than Cancelled."
1932 ),
1933 )
1934 return
1937class BypassCancellingFlowRunsWithNoInfra(FlowRunOrchestrationRule):
1938 """Rejects transitions from Scheduled to Cancelling, and instead sets the state to Cancelled,
1939 if the flow run has no associated infrastructure process ID. Also Rejects transitions from
1940 Paused to Cancelling if the Paused state's details indicates the flow run has been suspended,
1941 exiting the flow and tearing down infra.
1943 The `Cancelling` state is used to clean up infrastructure. If there is not infrastructure
1944 to clean up, we can transition directly to `Cancelled`. Runs that are `Resuming` are in a
1945 `Scheduled` state that were previously `Suspended` and do not yet have infrastructure.
1947 Runs that are `AwaitingRetry` are a `Scheduled` state that may have associated infrastructure.
1948 """
1950 FROM_STATES = {StateType.SCHEDULED, StateType.PAUSED}
1951 TO_STATES = {StateType.CANCELLING}
1953 async def before_transition(
1954 self,
1955 initial_state: states.State[Any] | None,
1956 proposed_state: states.State[Any] | None,
1957 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1958 ) -> None:
1959 if initial_state is None or proposed_state is None:
1960 return
1962 if (
1963 initial_state.type == states.StateType.SCHEDULED
1964 and not context.run.infrastructure_pid
1965 or initial_state.name == "Resuming"
1966 ):
1967 await self.reject_transition(
1968 state=states.Cancelled(),
1969 reason="Scheduled flow run has no infrastructure to terminate.",
1970 )
1971 elif ( 1971 ↛ anywhereline 1971 didn't jump anywhere: it always raised an exception.
1972 initial_state.type == states.StateType.PAUSED
1973 and initial_state.state_details.pause_reschedule
1974 ):
1975 await self.reject_transition(
1976 state=states.Cancelled(),
1977 reason="Suspended flow run has no infrastructure to terminate.",
1978 )
1981class PreserveDeploymentConcurrencyLeaseId(FlowRunUniversalTransform):
1982 """
1983 Preserves the deployment concurrency lease ID across state transitions.
1985 Workers send deployment_concurrency_lease_id: null in the proposed state JSON
1986 body (e.g., for PENDING→PENDING(Submitting)). Pydantic v2 treats null JSON
1987 fields as explicitly set, so the lease ID would otherwise be silently dropped.
1988 This transform copies the lease ID forward whenever the initial state has one
1989 and the proposed state does not.
1990 """
1992 async def before_transition(
1993 self,
1994 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
1995 ) -> None:
1996 if context.initial_state is None or context.proposed_state is None: 1996 ↛ 1997line 1996 didn't jump to line 1997 because the condition on line 1996 was never true
1997 return
1998 lease_id = context.initial_state.state_details.deployment_concurrency_lease_id
1999 if ( 1999 ↛ 2004line 1999 didn't jump to line 2004 because the condition on line 1999 was never true
2000 lease_id is not None
2001 and context.proposed_state.state_details.deployment_concurrency_lease_id
2002 is None
2003 ):
2004 context.proposed_state.state_details.deployment_concurrency_lease_id = (
2005 lease_id
2006 )
2009class PreventDuplicateTransitions(FlowRunOrchestrationRule):
2010 """
2011 Prevent duplicate transitions from being made right after one another.
2013 This rule allows for clients to set an optional transition_id on a state. If the
2014 run's next transition has the same transition_id, the transition will be
2015 rejected and the existing state will be returned.
2017 This allows for clients to make state transition requests without worrying about
2018 the following case:
2019 - A client making a state transition request
2020 - The server accepts transition and commits the transition
2021 - The client is unable to receive the response and retries the request
2022 """
2024 FROM_STATES: set[states.StateType | None] = ALL_ORCHESTRATION_STATES
2025 TO_STATES: set[states.StateType | None] = ALL_ORCHESTRATION_STATES
2027 async def before_transition(
2028 self,
2029 initial_state: states.State[Any] | None,
2030 proposed_state: states.State[Any] | None,
2031 context: OrchestrationContext[orm_models.FlowRun, core.FlowRunPolicy],
2032 ) -> None:
2033 if initial_state is None or proposed_state is None: 2033 ↛ 2034line 2033 didn't jump to line 2034 because the condition on line 2033 was never true
2034 return
2036 initial_transition_id = getattr(
2037 initial_state.state_details, "transition_id", None
2038 )
2039 proposed_transition_id = getattr(
2040 proposed_state.state_details, "transition_id", None
2041 )
2042 if ( 2042 ↛ 2047line 2042 didn't jump to line 2047 because the condition on line 2042 was never true
2043 initial_transition_id is not None
2044 and proposed_transition_id is not None
2045 and initial_transition_id == proposed_transition_id
2046 ):
2047 await self.reject_transition(
2048 # state=None will return the initial (current) state
2049 state=None,
2050 reason="This run has already made this state transition.",
2051 )