Coverage for async_durable_execution/_core/exceptions.py: 98%
332 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1"""Exceptions for the Durable Executions SDK.
3Avoid any non-stdlib references in this module, it is at the bottom of the dependency chain.
4"""
6from __future__ import annotations
8import datetime
9import json
10import time
11from collections.abc import Callable
12from dataclasses import dataclass
13from enum import Enum
14from typing import Any, NoReturn, TypedDict, cast
16BAD_REQUEST_ERROR: int = 400
17TOO_MANY_REQUESTS_ERROR: int = 429
18SERVICE_ERROR: int = 500
19INVALID_PARAMETER_VALUE_EXCEPTION: str = "InvalidParameterValueException"
20INVALID_CHECKPOINT_TOKEN_PREFIX: str = "Invalid Checkpoint Token"
21_SDK_ERROR_DATA_KEY: str = "__async_durable_execution_error__"
22_SDK_ERROR_DATA_VERSION: int = 1
23_SDK_INVOCATION_ERROR_PAYLOAD_VERSION: int = 1
24_SDK_EXECUTION_ERROR_PAYLOAD_VERSION: int = 1
27@dataclass(frozen=True)
28class _SdkControlErrorCodec:
29 exception_type: type[ExecutionError]
30 encode_payload: Callable[[ExecutionError], str | None] | None
31 restore: Callable[[str, str | None], ExecutionError]
34_SDK_CONTROL_ERROR_CODECS_BY_TYPE: dict[
35 type[ExecutionError], _SdkControlErrorCodec
36] = {}
37_SDK_CONTROL_ERROR_CODECS_BY_NAME: dict[str, _SdkControlErrorCodec] = {}
39# Non-retryable customer error codes that arrive as non-4xx (e.g. HTTP 502) from Lambda.
40# Unlike typical 5xx errors, these require customer intervention (e.g., fixing
41# a KMS key configuration) and will never succeed on retry.
42# Add new non-retryable error codes here — they are automatically classified
43# as EXECUTION (non-retryable) by _classify_error_category().
44_NON_RETRYABLE_CUSTOMER_ERROR_CODES: frozenset[str] = frozenset(
45 {
46 "KMSAccessDeniedException",
47 "KMSDisabledException",
48 "KMSInvalidStateException",
49 "KMSNotFoundException",
50 }
51)
54def _encode_sdk_error_data(
55 exception_type: type[Exception],
56 payload: str | None = None,
57) -> str:
58 """Encode SDK-owned exception metadata for durable replay."""
59 return json.dumps(
60 {
61 _SDK_ERROR_DATA_KEY: _SDK_ERROR_DATA_VERSION,
62 "exception_type": (
63 f"{exception_type.__module__}.{exception_type.__qualname__}"
64 ),
65 "payload": payload,
66 },
67 separators=(",", ":"),
68 sort_keys=True,
69 )
72def _decode_sdk_error_data_envelope(
73 data: str | None,
74) -> tuple[str, str | None] | None:
75 """Decode validated SDK exception metadata without assuming a specific type."""
76 if data is None:
77 return None
79 try:
80 decoded = json.loads(data)
81 except (TypeError, ValueError):
82 return None
84 if not isinstance(decoded, dict):
85 return None
87 version = decoded.get(_SDK_ERROR_DATA_KEY)
88 if type(version) is not int or version != _SDK_ERROR_DATA_VERSION:
89 return None
91 exception_type_name = decoded.get("exception_type")
92 if not isinstance(exception_type_name, str) or not exception_type_name: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 return None
95 payload = decoded.get("payload")
96 if payload is not None and not isinstance(payload, str):
97 return None
99 return exception_type_name, payload
102def _decode_sdk_error_data(
103 data: str | None,
104 expected_exception_type: type[Exception],
105 *,
106 legacy_exception_type_names: tuple[str, ...] = (),
107) -> tuple[bool, str | None]:
108 """Return whether data identifies the expected SDK exception and its payload.
110 Args:
111 data: Encoded exception metadata.
112 expected_exception_type: Current exception type to identify.
113 legacy_exception_type_names: Previous qualified names accepted for replay.
114 """
115 decoded = _decode_sdk_error_data_envelope(data)
116 if decoded is None:
117 return False, None
119 encoded_name, payload = decoded
120 expected_name = (
121 f"{expected_exception_type.__module__}.{expected_exception_type.__qualname__}"
122 )
123 if (
124 encoded_name != expected_name
125 and encoded_name not in legacy_exception_type_names
126 ):
127 return False, None
129 return True, payload
132def _register_sdk_control_error_type(
133 exception_type: type[ExecutionError],
134 *,
135 encode_payload: Callable[[ExecutionError], str | None] | None = None,
136 restore: Callable[[str, str | None], ExecutionError],
137 legacy_exception_type_names: tuple[str, ...] = (),
138) -> None:
139 """Register an operation-specific execution control error codec."""
140 if not issubclass(exception_type, ExecutionError): 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 msg = "SDK control error codecs require an ExecutionError subclass."
142 raise TypeError(msg)
144 codec = _SdkControlErrorCodec(
145 exception_type=exception_type,
146 encode_payload=encode_payload,
147 restore=restore,
148 )
149 current_name = f"{exception_type.__module__}.{exception_type.__qualname__}"
150 for exception_type_name in (current_name, *legacy_exception_type_names):
151 existing = _SDK_CONTROL_ERROR_CODECS_BY_NAME.get(exception_type_name)
152 if existing is not None and existing.exception_type is not exception_type: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 msg = f"SDK control error type name is already registered: {exception_type_name}"
154 raise ValueError(msg)
155 _SDK_CONTROL_ERROR_CODECS_BY_NAME[exception_type_name] = codec
156 _SDK_CONTROL_ERROR_CODECS_BY_TYPE[exception_type] = codec
159def _encode_sdk_control_error_data(
160 error: Exception,
161 *,
162 invocation_retryable: bool | None = None,
163) -> str | None:
164 """Encode the SDK-owned control category of an exception, if any."""
165 if isinstance(error, ExecutionError):
166 codec = _SDK_CONTROL_ERROR_CODECS_BY_TYPE.get(type(error))
167 if codec is not None:
168 return _encode_sdk_error_data(
169 codec.exception_type,
170 (
171 codec.encode_payload(error)
172 if codec.encode_payload is not None
173 else None
174 ),
175 )
177 if isinstance(error, InvocationError):
178 return _encode_sdk_error_data(
179 InvocationError,
180 _encode_sdk_invocation_error_payload(
181 error,
182 retryable=invocation_retryable,
183 ),
184 )
185 if isinstance(error, ExecutionError):
186 return _encode_sdk_error_data(
187 ExecutionError,
188 _encode_sdk_execution_error_payload(
189 error,
190 error.termination_reason,
191 ),
192 )
193 if isinstance(error, SerDesError):
194 return _encode_sdk_error_data(
195 SerDesError,
196 _encode_sdk_execution_error_payload(
197 error,
198 TerminationReason.SERIALIZATION_ERROR,
199 ),
200 )
201 return None
204class AwsErrorObj(TypedDict):
205 """Subset of a boto-style AWS error payload."""
207 Code: str | None
208 Message: str | None
211class AwsErrorMetadata(TypedDict, total=False):
212 """Subset of boto response metadata used for retry classification."""
214 RequestId: str | None
215 HostId: str | None
216 HTTPStatusCode: int | None
217 HTTPHeaders: str | None
218 RetryAttempts: str | None
221class TerminationReason(Enum):
222 """Reasons why a durable execution terminated."""
224 UNHANDLED_ERROR = "UNHANDLED_ERROR"
225 INVOCATION_ERROR = "INVOCATION_ERROR"
226 EXECUTION_ERROR = "EXECUTION_ERROR"
227 CHECKPOINT_FAILED = "CHECKPOINT_FAILED"
228 NON_DETERMINISTIC_EXECUTION = "NON_DETERMINISTIC_EXECUTION"
229 STEP_INTERRUPTED = "STEP_INTERRUPTED"
230 CALLBACK_ERROR = "CALLBACK_ERROR"
231 SERIALIZATION_ERROR = "SERIALIZATION_ERROR"
234class DurableExecutionsError(Exception):
235 """Base class for Durable Executions exceptions"""
238class UnrecoverableError(DurableExecutionsError):
239 """Base class for errors that terminate execution."""
241 def __init__(self, message: str, termination_reason: TerminationReason) -> None:
242 super().__init__(message)
243 self.termination_reason = termination_reason
246class ExecutionError(UnrecoverableError):
247 """Error that returns FAILED status without retry."""
249 def __init__(
250 self,
251 message: str,
252 termination_reason: TerminationReason = TerminationReason.EXECUTION_ERROR,
253 ) -> None:
254 super().__init__(message, termination_reason)
257class _RestoredExecutionError(ExecutionError):
258 """Execution control error restored without importing its original class."""
260 def __init__(
261 self,
262 message: str,
263 *,
264 original_error_type: str,
265 termination_reason: TerminationReason,
266 ) -> None:
267 super().__init__(message, termination_reason)
268 self.original_error_type = original_error_type
271class InvocationError(UnrecoverableError):
272 """Error that should cause Lambda retry by throwing from handler."""
274 def __init__(
275 self,
276 message: str,
277 termination_reason: TerminationReason = TerminationReason.INVOCATION_ERROR,
278 ) -> None:
279 super().__init__(message, termination_reason)
281 def is_retryable(self) -> bool:
282 """Whether this error is retryable. Returns True by default.
284 Subclasses override to implement classification logic based on
285 error codes and HTTP status codes.
286 """
287 return True
289 def build_logger_extras(self) -> dict:
290 """Return structured logging extras for retryable invocation errors."""
291 return {}
294class RetryableSerDesError(InvocationError):
295 """Transient serialization failure that should enter retry handling."""
297 def __init__(self, message: str) -> None:
298 super().__init__(
299 message,
300 termination_reason=TerminationReason.SERIALIZATION_ERROR,
301 )
304class DurableApiErrorCategory(Enum):
305 """Whether a durable API failure should retry the Lambda or fail execution."""
307 INVOCATION = "INVOCATION"
308 EXECUTION = "EXECUTION"
311# Backward-compatible alias
312CheckpointErrorCategory = DurableApiErrorCategory
315class BotoClientError(InvocationError):
316 """Error from a Lambda API call (e.g., CheckpointDurableExecution, GetDurableExecutionState).
318 Extends InvocationError because the default behavior for API failures is to retry
319 the Lambda invocation. However, some errors are non-retryable (e.g., 4xx client errors,
320 KMS key misconfiguration) and should fail the execution instead. The error_category field
321 and is_retryable() method distinguish these cases at runtime.
322 """
324 def __init__(
325 self,
326 message: str,
327 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION,
328 error: AwsErrorObj | None = None,
329 response_metadata: AwsErrorMetadata | None = None,
330 termination_reason=TerminationReason.INVOCATION_ERROR,
331 ) -> None:
332 super().__init__(message=message, termination_reason=termination_reason)
333 self.error: AwsErrorObj | None = error
334 self.response_metadata: AwsErrorMetadata | None = response_metadata
335 self.error_category: DurableApiErrorCategory = error_category
337 @classmethod
338 def from_exception(cls, exception: Exception) -> BotoClientError:
339 response = getattr(exception, "response", {})
340 response_metadata = response.get("ResponseMetadata")
341 error = response.get("Error")
342 error_category = BotoClientError._classify_error_category(
343 error, response_metadata
344 )
345 return cls(
346 message=str(exception),
347 error_category=error_category,
348 error=error,
349 response_metadata=response_metadata,
350 )
352 @staticmethod
353 def _classify_error_category(
354 error: AwsErrorObj | None,
355 response_metadata: AwsErrorMetadata | None,
356 ) -> DurableApiErrorCategory:
357 """Classify a Durable API error as retryable (INVOCATION) or non-retryable (EXECUTION).
359 Classification rules:
360 - Non-retryable customer error codes (e.g., KMS key issues) → EXECUTION
361 These arrive as HTTP 502 but require customer intervention to fix.
362 - 4xx errors → EXECUTION, except:
363 - 429 (TooManyRequests) → INVOCATION (throttling is transient)
364 - InvalidParameterValueException with "Invalid Checkpoint Token" → INVOCATION
365 (stale token from a concurrent checkpoint; next invocation gets a fresh token)
366 - 5xx, network errors → INVOCATION
367 """
368 error_code: str | None = error.get("Code") if error else None
369 if error_code and error_code in _NON_RETRYABLE_CUSTOMER_ERROR_CODES:
370 return DurableApiErrorCategory.EXECUTION
372 status_code: int | None = (
373 response_metadata.get("HTTPStatusCode") if response_metadata else None
374 )
376 if (
377 status_code
378 and BAD_REQUEST_ERROR <= status_code < SERVICE_ERROR
379 and status_code != TOO_MANY_REQUESTS_ERROR
380 and error
381 and not (
382 (error.get("Code") or "") == INVALID_PARAMETER_VALUE_EXCEPTION
383 and (error.get("Message") or "").startswith(
384 INVALID_CHECKPOINT_TOKEN_PREFIX
385 )
386 )
387 ):
388 return DurableApiErrorCategory.EXECUTION
390 return DurableApiErrorCategory.INVOCATION
392 def is_retryable(self) -> bool:
393 """Whether this error is retryable based on error_category."""
394 return self.error_category == DurableApiErrorCategory.INVOCATION
396 # Backward-compatible alias
397 is_retriable = is_retryable
399 def build_logger_extras(self) -> dict:
400 extras: dict = {}
401 # preserve PascalCase to be consistent with other langauges
402 if error := self.error:
403 extras["Error"] = error
404 if response_metadata := self.response_metadata:
405 extras["ResponseMetadata"] = response_metadata
406 return extras
409class _RestoredInvocationError(InvocationError):
410 """Invocation control error restored without importing its original class."""
412 def __init__(
413 self,
414 message: str,
415 *,
416 original_error_type: str,
417 retryable: bool,
418 termination_reason: TerminationReason,
419 error_category: DurableApiErrorCategory | None = None,
420 error: AwsErrorObj | None = None,
421 response_metadata: AwsErrorMetadata | None = None,
422 ) -> None:
423 super().__init__(message, termination_reason)
424 self.original_error_type = original_error_type
425 self.retryable = retryable
426 self.error_category = error_category
427 self.error = error
428 self.response_metadata = response_metadata
430 def is_retryable(self) -> bool:
431 return self.retryable
433 def build_logger_extras(self) -> dict:
434 extras: dict = {}
435 if self.error is not None:
436 extras["Error"] = self.error
437 if self.response_metadata is not None:
438 extras["ResponseMetadata"] = self.response_metadata
439 return extras
442def _sdk_error_type_name(error: Exception) -> str:
443 """Return the original type name for a restored SDK control error."""
444 if isinstance(error, _RestoredInvocationError | _RestoredExecutionError):
445 return error.original_error_type
446 return type(error).__name__
449def _encode_sdk_execution_error_payload(
450 error: Exception,
451 termination_reason: TerminationReason,
452) -> str:
453 """Encode execution-control details needed after durable replay."""
454 return json.dumps(
455 {
456 "version": _SDK_EXECUTION_ERROR_PAYLOAD_VERSION,
457 "error_type": _sdk_error_type_name(error),
458 "termination_reason": termination_reason.value,
459 },
460 separators=(",", ":"),
461 sort_keys=True,
462 )
465def _restore_sdk_execution_error(
466 message: str,
467 error_type: str | None,
468 payload: str | None,
469 *,
470 default_termination_reason: TerminationReason,
471) -> ExecutionError:
472 """Restore an execution control error from current or legacy metadata."""
473 original_error_type = error_type or ExecutionError.__name__
474 termination_reason = default_termination_reason
476 try:
477 termination_reason = TerminationReason(payload)
478 except (TypeError, ValueError):
479 try:
480 decoded = json.loads(payload) if payload is not None else None
481 except (TypeError, ValueError):
482 decoded = None
483 if (
484 isinstance(decoded, dict)
485 and decoded.get("version") == _SDK_EXECUTION_ERROR_PAYLOAD_VERSION
486 ):
487 encoded_error_type = decoded.get("error_type")
488 if isinstance(encoded_error_type, str) and encoded_error_type:
489 original_error_type = encoded_error_type
490 try:
491 termination_reason = TerminationReason(
492 decoded.get("termination_reason")
493 )
494 except (TypeError, ValueError):
495 pass
497 return _RestoredExecutionError(
498 message,
499 original_error_type=original_error_type,
500 termination_reason=termination_reason,
501 )
504def _encode_sdk_invocation_error_payload(
505 error: InvocationError,
506 *,
507 retryable: bool | None = None,
508) -> str:
509 """Encode retry behavior needed to safely restore an invocation error."""
510 details: dict[str, Any] = {
511 "version": _SDK_INVOCATION_ERROR_PAYLOAD_VERSION,
512 "error_type": _sdk_error_type_name(error),
513 "retryable": error.is_retryable() if retryable is None else retryable,
514 "termination_reason": error.termination_reason.value,
515 }
516 if isinstance(error, BotoClientError | _RestoredInvocationError):
517 if error.error_category is not None:
518 details["error_category"] = error.error_category.value
519 details["error"] = error.error
520 details["response_metadata"] = error.response_metadata
522 try:
523 return json.dumps(details, separators=(",", ":"), sort_keys=True)
524 except (TypeError, ValueError):
525 # Retry behavior is control data; diagnostic boto payloads are best effort.
526 details.pop("error", None)
527 details.pop("response_metadata", None)
528 return json.dumps(details, separators=(",", ":"), sort_keys=True)
531def _restore_sdk_invocation_error(
532 message: str,
533 error_type: str | None,
534 payload: str | None,
535) -> InvocationError:
536 """Restore replay-critical invocation semantics from SDK-owned metadata."""
537 original_error_type = error_type or InvocationError.__name__
538 # Legacy envelopes had no payload and represented retryable InvocationError.
539 # A present but invalid payload fails closed to avoid an unbounded retry loop.
540 retryable = payload is None
541 termination_reason = TerminationReason.INVOCATION_ERROR
542 error_category: DurableApiErrorCategory | None = None
543 error: AwsErrorObj | None = None
544 response_metadata: AwsErrorMetadata | None = None
546 try:
547 decoded = json.loads(payload) if payload is not None else None
548 except (TypeError, ValueError):
549 decoded = None
551 if (
552 isinstance(decoded, dict)
553 and decoded.get("version") == _SDK_INVOCATION_ERROR_PAYLOAD_VERSION
554 ):
555 encoded_error_type = decoded.get("error_type")
556 if isinstance(encoded_error_type, str) and encoded_error_type:
557 original_error_type = encoded_error_type
559 encoded_retryable = decoded.get("retryable")
560 if type(encoded_retryable) is bool:
561 retryable = encoded_retryable
563 try:
564 termination_reason = TerminationReason(decoded.get("termination_reason"))
565 except (TypeError, ValueError):
566 pass
568 try:
569 error_category = DurableApiErrorCategory(decoded.get("error_category"))
570 except (TypeError, ValueError):
571 pass
573 encoded_error = decoded.get("error")
574 if isinstance(encoded_error, dict):
575 error = cast("AwsErrorObj", encoded_error)
576 encoded_response_metadata = decoded.get("response_metadata")
577 if isinstance(encoded_response_metadata, dict):
578 response_metadata = cast(
579 "AwsErrorMetadata",
580 encoded_response_metadata,
581 )
583 return _RestoredInvocationError(
584 message,
585 original_error_type=original_error_type,
586 retryable=retryable,
587 termination_reason=termination_reason,
588 error_category=error_category,
589 error=error,
590 response_metadata=response_metadata,
591 )
594def _restore_sdk_control_error(
595 message: str,
596 error_type: str | None,
597 data: str | None,
598) -> ExecutionError | InvocationError | None:
599 """Restore a control error identified by SDK-owned checkpoint metadata."""
600 decoded = _decode_sdk_error_data_envelope(data)
601 if decoded is not None:
602 exception_type_name, payload = decoded
603 codec = _SDK_CONTROL_ERROR_CODECS_BY_NAME.get(exception_type_name)
604 if codec is not None and error_type == codec.exception_type.__name__:
605 return codec.restore(message, payload)
607 is_invocation_error, payload = _decode_sdk_error_data(
608 data,
609 InvocationError,
610 legacy_exception_type_names=(
611 "async_durable_execution.exceptions.InvocationError",
612 "async_durable_execution.core.exceptions.InvocationError",
613 ),
614 )
615 if is_invocation_error:
616 return _restore_sdk_invocation_error(message, error_type, payload)
618 is_execution_error, payload = _decode_sdk_error_data(
619 data,
620 ExecutionError,
621 legacy_exception_type_names=(
622 "async_durable_execution.exceptions.ExecutionError",
623 "async_durable_execution.core.exceptions.ExecutionError",
624 ),
625 )
626 if is_execution_error:
627 return _restore_sdk_execution_error(
628 message,
629 error_type,
630 payload,
631 default_termination_reason=TerminationReason.EXECUTION_ERROR,
632 )
634 is_serdes_error, payload = _decode_sdk_error_data(
635 data,
636 SerDesError,
637 legacy_exception_type_names=(
638 "async_durable_execution.exceptions.SerDesError",
639 "async_durable_execution.core.exceptions.SerDesError",
640 ),
641 )
642 if is_serdes_error:
643 return _restore_sdk_execution_error(
644 message,
645 error_type,
646 payload,
647 default_termination_reason=TerminationReason.SERIALIZATION_ERROR,
648 )
650 return None
653class NonDeterministicExecutionError(ExecutionError):
654 """Error when execution is non-deterministic."""
656 def __init__(self, message: str, step_id: str | None = None) -> None:
657 super().__init__(message, TerminationReason.NON_DETERMINISTIC_EXECUTION)
658 self.step_id = step_id
661class CheckpointError(BotoClientError):
662 """Failure to checkpoint. Will terminate the lambda."""
664 def __init__(
665 self,
666 message: str,
667 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION,
668 error: AwsErrorObj | None = None,
669 response_metadata: AwsErrorMetadata | None = None,
670 ) -> None:
671 super().__init__(
672 message,
673 error_category,
674 error,
675 response_metadata,
676 termination_reason=TerminationReason.CHECKPOINT_FAILED,
677 )
680class ValidationError(DurableExecutionsError):
681 """Incorrect arguments to a Durable Function operation."""
684class GetExecutionStateError(BotoClientError):
685 """Raised when failing to retrieve execution state"""
687 def __init__(
688 self,
689 message: str,
690 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION,
691 error: AwsErrorObj | None = None,
692 response_metadata: AwsErrorMetadata | None = None,
693 ) -> None:
694 super().__init__(
695 message,
696 error_category,
697 error,
698 response_metadata,
699 termination_reason=TerminationReason.INVOCATION_ERROR,
700 )
703class InvalidStateError(DurableExecutionsError):
704 """Raised when an operation is attempted on an object in an invalid state."""
707class UserlandError(DurableExecutionsError):
708 """Failure in user-land - i.e code passed into durable executions from the caller."""
711class CallableRuntimeError(UserlandError):
712 """This error wraps any failure from inside the callable code that you pass to a Durable Function operation."""
714 def __init__(
715 self,
716 message: str | None,
717 error_type: str | None,
718 data: str | None,
719 stack_trace: list[str] | None,
720 ) -> None:
721 super().__init__(message)
722 self.message = message
723 self.error_type = error_type
724 self.data = data
725 self.stack_trace = stack_trace
727 @classmethod
728 def from_error_object(cls, error_object: Any) -> CallableRuntimeError:
729 return cls(
730 message=error_object.message,
731 error_type=error_object.type,
732 data=error_object.data,
733 stack_trace=error_object.stack_trace,
734 )
737class BackgroundThreadError(BaseException):
738 """Critical error from background checkpoint thread.
740 Derives from BaseException to bypass normal exception handlers.
741 Similar to KeyboardInterrupt or SystemExit - this is a system-level
742 error that should terminate execution immediately without attempting
743 to checkpoint or process the error.
745 This exception is raised in the user thread when the background
746 checkpoint processing thread encounters a fatal error. It propagates
747 through the awaiting checkpoint future to interrupt blocked user code.
749 Attributes:
750 source_exception: The original exception from the background thread
751 """
753 def __init__(self, message: str, source_exception: Exception) -> None:
754 super().__init__(message)
755 self.source_exception = source_exception
758class OrphanedChildException(BaseException):
759 """Raised when an operation checkpoints after its parent context completed.
761 This inherits from BaseException so user code does not accidentally catch it
762 with broad exception handlers like ``except Exception``.
763 """
765 def __init__(self, message: str, operation_id: str) -> None:
766 super().__init__(message)
767 self.operation_id = operation_id
770class SuspendExecution(BaseException):
771 """Raise this exception to suspend the current execution by returning PENDING to DAR.
773 Note this derives from BaseException - in keeping with system-exiting exceptions like
774 KeyboardInterrupt or SystemExit.
775 """
777 def __init__(self, message: str) -> None:
778 super().__init__(message)
781class TimedSuspendExecution(SuspendExecution):
782 """Suspend execution until a specific timestamp.
784 This is a specialized form of SuspendExecution that includes a scheduled resume time.
786 Attributes:
787 scheduled_timestamp (float): Unix timestamp in seconds at which to resume.
788 """
790 def __init__(self, message: str, scheduled_timestamp: float) -> None:
791 super().__init__(message)
792 self.scheduled_timestamp = scheduled_timestamp
794 @classmethod
795 def from_delay(cls, message: str, delay_seconds: int) -> TimedSuspendExecution:
796 """Create a timed suspension with the delay calculated from now.
798 Args:
799 message: Descriptive message for the suspension
800 delay_seconds: Number of seconds to suspend from current time
802 Returns:
803 TimedSuspendExecution: Instance with calculated resume time
805 Example:
806 >>> exception = TimedSuspendExecution.from_delay("Waiting for callback", 30)
807 >>> # Will suspend for 30 seconds from now
808 """
809 resume_time = time.time() + delay_seconds
810 return cls(message, scheduled_timestamp=resume_time)
812 @classmethod
813 def from_datetime(
814 cls, message: str, datetime_timestamp: datetime.datetime
815 ) -> TimedSuspendExecution:
816 """Create a timed suspension with the delay calculated from now.
818 Args:
819 message: Descriptive message for the suspension
820 datetime_timestamp: Unix datetime timestamp in seconds at which to resume
822 Returns:
823 TimedSuspendExecution: Instance with calculated resume time
824 """
825 return cls(message, scheduled_timestamp=datetime_timestamp.timestamp())
828def suspend_with_optional_resume_timestamp(
829 msg: str, datetime_timestamp: datetime.datetime | None = None
830) -> NoReturn:
831 """Suspend execution with an optional target resume timestamp."""
833 if datetime_timestamp is None:
834 msg = f"No timestamp provided. Suspending without retry timestamp. Original operation: [{msg}]"
835 raise SuspendExecution(msg)
837 if datetime_timestamp < datetime.datetime.now(tz=datetime.timezone.utc):
838 msg = f"Invalid timestamp {datetime_timestamp}, suspending with immediate retry, original operation: [{msg}]"
839 raise TimedSuspendExecution.from_datetime(
840 msg, datetime.datetime.now(tz=datetime.timezone.utc)
841 )
843 raise TimedSuspendExecution.from_datetime(msg, datetime_timestamp)
846def suspend_with_optional_resume_delay(
847 msg: str, delay_seconds: int | None = None
848) -> NoReturn:
849 """Suspend execution with an optional delay before resuming."""
851 if delay_seconds is None:
852 msg = f"No delay_seconds provided, suspending without retry timestamp, original operation: [{msg}]"
853 raise SuspendExecution(msg)
855 if delay_seconds < 0:
856 msg = f"Invalid delay_seconds {delay_seconds}, suspending with delay 0, original operation: [{msg}]"
857 raise TimedSuspendExecution.from_delay(msg, 0)
859 raise TimedSuspendExecution.from_delay(msg, delay_seconds)
862@dataclass(frozen=True)
863class CallableRuntimeErrorSerializableDetails:
864 """Serializable error details."""
866 type: str
867 message: str
869 @classmethod
870 def from_exception(
871 cls, exception: Exception
872 ) -> CallableRuntimeErrorSerializableDetails:
873 """Create an instance from an Exception, using its type and message.
875 Args:
876 exception: An Exception instance
878 Returns:
879 A CallableRuntimeErrorDetails instance with the exception's type name and message
880 """
881 return cls(type=exception.__class__.__name__, message=str(exception))
883 def __str__(self) -> str:
884 """
885 Return a string representation of the object.
887 Returns:
888 A string in the format "type: message"
889 """
890 return f"{self.type}: {self.message}"
893class SerDesError(DurableExecutionsError):
894 """Raised when serialization fails."""