Coverage for async-durable-execution/src/async_durable_execution/exceptions.py: 97%

1"""Exceptions for the Durable Executions SDK. 

2 

3Avoid any non-stdlib references in this module, it is at the bottom of the dependency chain. 

4""" 

5 

6from __future__ import annotations 

7 

8import datetime 

9import time 

10from dataclasses import dataclass 

11from enum import Enum 

12from typing import Any, NoReturn, TypedDict 

13 

14BAD_REQUEST_ERROR: int = 400 

15TOO_MANY_REQUESTS_ERROR: int = 429 

16SERVICE_ERROR: int = 500 

17INVALID_PARAMETER_VALUE_EXCEPTION: str = "InvalidParameterValueException" 

18INVALID_CHECKPOINT_TOKEN_PREFIX: str = "Invalid Checkpoint Token" 

19 

20# Non-retryable customer error codes that arrive as non-4xx (e.g. HTTP 502) from Lambda. 

21# Unlike typical 5xx errors, these require customer intervention (e.g., fixing 

22# a KMS key configuration) and will never succeed on retry. 

23# Add new non-retryable error codes here — they are automatically classified 

24# as EXECUTION (non-retryable) by _classify_error_category(). 

25_NON_RETRYABLE_CUSTOMER_ERROR_CODES: frozenset[str] = frozenset( 

26 { 

27 "KMSAccessDeniedException", 

28 "KMSDisabledException", 

29 "KMSInvalidStateException", 

30 "KMSNotFoundException", 

31 } 

32) 

33 

34 

35class AwsErrorObj(TypedDict): 

36 """Subset of a boto-style AWS error payload.""" 

37 

38 Code: str | None 

39 Message: str | None 

40 

41 

42class AwsErrorMetadata(TypedDict, total=False): 

43 """Subset of boto response metadata used for retry classification.""" 

44 

45 RequestId: str | None 

46 HostId: str | None 

47 HTTPStatusCode: int | None 

48 HTTPHeaders: str | None 

49 RetryAttempts: str | None 

50 

51 

52class TerminationReason(Enum): 

53 """Reasons why a durable execution terminated.""" 

54 

55 UNHANDLED_ERROR = "UNHANDLED_ERROR" 

56 INVOCATION_ERROR = "INVOCATION_ERROR" 

57 EXECUTION_ERROR = "EXECUTION_ERROR" 

58 CHECKPOINT_FAILED = "CHECKPOINT_FAILED" 

59 NON_DETERMINISTIC_EXECUTION = "NON_DETERMINISTIC_EXECUTION" 

60 STEP_INTERRUPTED = "STEP_INTERRUPTED" 

61 CALLBACK_ERROR = "CALLBACK_ERROR" 

62 SERIALIZATION_ERROR = "SERIALIZATION_ERROR" 

63 

64 

65class DurableExecutionsError(Exception): 

66 """Base class for Durable Executions exceptions""" 

67 

68 

69class UnrecoverableError(DurableExecutionsError): 

70 """Base class for errors that terminate execution.""" 

71 

72 def __init__(self, message: str, termination_reason: TerminationReason): 

73 super().__init__(message) 

74 self.termination_reason = termination_reason 

75 

76 

77class ExecutionError(UnrecoverableError): 

78 """Error that returns FAILED status without retry.""" 

79 

80 def __init__( 

81 self, 

82 message: str, 

83 termination_reason: TerminationReason = TerminationReason.EXECUTION_ERROR, 

84 ): 

85 super().__init__(message, termination_reason) 

86 

87 

88class InvocationError(UnrecoverableError): 

89 """Error that should cause Lambda retry by throwing from handler.""" 

90 

91 def __init__( 

92 self, 

93 message: str, 

94 termination_reason: TerminationReason = TerminationReason.INVOCATION_ERROR, 

95 ): 

96 super().__init__(message, termination_reason) 

97 

98 def is_retryable(self) -> bool: 

99 """Whether this error is retryable. Returns True by default. 

100 

101 Subclasses override to implement classification logic based on 

102 error codes and HTTP status codes. 

103 """ 

104 return True 

105 

106 def build_logger_extras(self) -> dict: 

107 """Return structured logging extras for retryable invocation errors.""" 

108 return {} 

109 

110 

111class DurableApiErrorCategory(Enum): 

112 """Whether a durable API failure should retry the Lambda or fail execution.""" 

113 

114 INVOCATION = "INVOCATION" 

115 EXECUTION = "EXECUTION" 

116 

117 

118# Backward-compatible alias 

119CheckpointErrorCategory = DurableApiErrorCategory 

120 

121 

122class BotoClientError(InvocationError): 

123 """Error from a Lambda API call (e.g., CheckpointDurableExecution, GetDurableExecutionState). 

124 

125 Extends InvocationError because the default behavior for API failures is to retry 

126 the Lambda invocation. However, some errors are non-retryable (e.g., 4xx client errors, 

127 KMS key misconfiguration) and should fail the execution instead. The error_category field 

128 and is_retryable() method distinguish these cases at runtime. 

129 """ 

130 

131 def __init__( 

132 self, 

133 message: str, 

134 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION, 

135 error: AwsErrorObj | None = None, 

136 response_metadata: AwsErrorMetadata | None = None, 

137 termination_reason=TerminationReason.INVOCATION_ERROR, 

138 ): 

139 super().__init__(message=message, termination_reason=termination_reason) 

140 self.error: AwsErrorObj | None = error 

141 self.response_metadata: AwsErrorMetadata | None = response_metadata 

142 self.error_category: DurableApiErrorCategory = error_category 

143 

144 @classmethod 

145 def from_exception(cls, exception: Exception) -> BotoClientError: 

146 response = getattr(exception, "response", {}) 

147 response_metadata = response.get("ResponseMetadata") 

148 error = response.get("Error") 

149 error_category = BotoClientError._classify_error_category( 

150 error, response_metadata 

151 ) 

152 return cls( 

153 message=str(exception), 

154 error_category=error_category, 

155 error=error, 

156 response_metadata=response_metadata, 

157 ) 

158 

159 @staticmethod 

160 def _classify_error_category( 

161 error: AwsErrorObj | None, 

162 response_metadata: AwsErrorMetadata | None, 

163 ) -> DurableApiErrorCategory: 

164 """Classify a Durable API error as retryable (INVOCATION) or non-retryable (EXECUTION). 

165 

166 Classification rules: 

167 - Non-retryable customer error codes (e.g., KMS key issues) → EXECUTION 

168 These arrive as HTTP 502 but require customer intervention to fix. 

169 - 4xx errors → EXECUTION, except: 

170 - 429 (TooManyRequests) → INVOCATION (throttling is transient) 

171 - InvalidParameterValueException with "Invalid Checkpoint Token" → INVOCATION 

172 (stale token from a concurrent checkpoint; next invocation gets a fresh token) 

173 - 5xx, network errors → INVOCATION 

174 """ 

175 error_code: str | None = error.get("Code") if error else None 

176 if error_code and error_code in _NON_RETRYABLE_CUSTOMER_ERROR_CODES: 

177 return DurableApiErrorCategory.EXECUTION 

178 

179 status_code: int | None = ( 

180 response_metadata.get("HTTPStatusCode") if response_metadata else None 

181 ) 

182 

183 if ( 

184 status_code 

185 and BAD_REQUEST_ERROR <= status_code < SERVICE_ERROR 

186 and status_code != TOO_MANY_REQUESTS_ERROR 

187 and error 

188 and not ( 

189 (error.get("Code") or "") == INVALID_PARAMETER_VALUE_EXCEPTION 

190 and (error.get("Message") or "").startswith( 

191 INVALID_CHECKPOINT_TOKEN_PREFIX 

192 ) 

193 ) 

194 ): 

195 return DurableApiErrorCategory.EXECUTION 

196 

197 return DurableApiErrorCategory.INVOCATION 

198 

199 def is_retryable(self) -> bool: 

200 """Whether this error is retryable based on error_category.""" 

201 return self.error_category == DurableApiErrorCategory.INVOCATION 

202 

203 # Backward-compatible alias 

204 is_retriable = is_retryable 

205 

206 def build_logger_extras(self) -> dict: 

207 extras: dict = {} 

208 # preserve PascalCase to be consistent with other langauges 

209 if error := self.error: 

210 extras["Error"] = error 

211 if response_metadata := self.response_metadata: 

212 extras["ResponseMetadata"] = response_metadata 

213 return extras 

214 

215 

216class NonDeterministicExecutionError(ExecutionError): 

217 """Error when execution is non-deterministic.""" 

218 

219 def __init__(self, message: str, step_id: str | None = None): 

220 super().__init__(message, TerminationReason.NON_DETERMINISTIC_EXECUTION) 

221 self.step_id = step_id 

222 

223 

224class CheckpointError(BotoClientError): 

225 """Failure to checkpoint. Will terminate the lambda.""" 

226 

227 def __init__( 

228 self, 

229 message: str, 

230 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION, 

231 error: AwsErrorObj | None = None, 

232 response_metadata: AwsErrorMetadata | None = None, 

233 ): 

234 super().__init__( 

235 message, 

236 error_category, 

237 error, 

238 response_metadata, 

239 termination_reason=TerminationReason.CHECKPOINT_FAILED, 

240 ) 

241 

242 

243class ValidationError(DurableExecutionsError): 

244 """Incorrect arguments to a Durable Function operation.""" 

245 

246 

247class GetExecutionStateError(BotoClientError): 

248 """Raised when failing to retrieve execution state""" 

249 

250 def __init__( 

251 self, 

252 message: str, 

253 error_category: DurableApiErrorCategory = DurableApiErrorCategory.INVOCATION, 

254 error: AwsErrorObj | None = None, 

255 response_metadata: AwsErrorMetadata | None = None, 

256 ): 

257 super().__init__( 

258 message, 

259 error_category, 

260 error, 

261 response_metadata, 

262 termination_reason=TerminationReason.INVOCATION_ERROR, 

263 ) 

264 

265 

266class InvalidStateError(DurableExecutionsError): 

267 """Raised when an operation is attempted on an object in an invalid state.""" 

268 

269 

270class UserlandError(DurableExecutionsError): 

271 """Failure in user-land - i.e code passed into durable executions from the caller.""" 

272 

273 

274class CallableRuntimeError(UserlandError): 

275 """This error wraps any failure from inside the callable code that you pass to a Durable Function operation.""" 

276 

277 def __init__( 

278 self, 

279 message: str | None, 

280 error_type: str | None, 

281 data: str | None, 

282 stack_trace: list[str] | None, 

283 ) -> None: 

284 super().__init__(message) 

285 self.message = message 

286 self.error_type = error_type 

287 self.data = data 

288 self.stack_trace = stack_trace 

289 

290 @classmethod 

291 def from_error_object(cls, error_object: Any) -> CallableRuntimeError: 

292 return cls( 

293 message=error_object.message, 

294 error_type=error_object.type, 

295 data=error_object.data, 

296 stack_trace=error_object.stack_trace, 

297 ) 

298 

299 

300class BackgroundThreadError(BaseException): 

301 """Critical error from background checkpoint thread. 

302 

303 Derives from BaseException to bypass normal exception handlers. 

304 Similar to KeyboardInterrupt or SystemExit - this is a system-level 

305 error that should terminate execution immediately without attempting 

306 to checkpoint or process the error. 

307 

308 This exception is raised in the user thread when the background 

309 checkpoint processing thread encounters a fatal error. It propagates 

310 through the awaiting checkpoint future to interrupt blocked user code. 

311 

312 Attributes: 

313 source_exception: The original exception from the background thread 

314 """ 

315 

316 def __init__(self, message: str, source_exception: Exception): 

317 super().__init__(message) 

318 self.source_exception = source_exception 

319 

320 

321class SuspendExecution(BaseException): 

322 """Raise this exception to suspend the current execution by returning PENDING to DAR. 

323 

324 Note this derives from BaseException - in keeping with system-exiting exceptions like 

325 KeyboardInterrupt or SystemExit. 

326 """ 

327 

328 def __init__(self, message: str): 

329 super().__init__(message) 

330 

331 

332class TimedSuspendExecution(SuspendExecution): 

333 """Suspend execution until a specific timestamp. 

334 

335 This is a specialized form of SuspendExecution that includes a scheduled resume time. 

336 

337 Attributes: 

338 scheduled_timestamp (float): Unix timestamp in seconds at which to resume. 

339 """ 

340 

341 def __init__(self, message: str, scheduled_timestamp: float): 

342 super().__init__(message) 

343 self.scheduled_timestamp = scheduled_timestamp 

344 

345 @classmethod 

346 def from_delay(cls, message: str, delay_seconds: int) -> TimedSuspendExecution: 

347 """Create a timed suspension with the delay calculated from now. 

348 

349 Args: 

350 message: Descriptive message for the suspension 

351 delay_seconds: Number of seconds to suspend from current time 

352 

353 Returns: 

354 TimedSuspendExecution: Instance with calculated resume time 

355 

356 Example: 

357 >>> exception = TimedSuspendExecution.from_delay("Waiting for callback", 30) 

358 >>> # Will suspend for 30 seconds from now 

359 """ 

360 resume_time = time.time() + delay_seconds 

361 return cls(message, scheduled_timestamp=resume_time) 

362 

363 @classmethod 

364 def from_datetime( 

365 cls, message: str, datetime_timestamp: datetime.datetime 

366 ) -> TimedSuspendExecution: 

367 """Create a timed suspension with the delay calculated from now. 

368 

369 Args: 

370 message: Descriptive message for the suspension 

371 datetime_timestamp: Unix datetime timestamp in seconds at which to resume 

372 

373 Returns: 

374 TimedSuspendExecution: Instance with calculated resume time 

375 """ 

376 return cls(message, scheduled_timestamp=datetime_timestamp.timestamp()) 

377 

378 

379def suspend_with_optional_resume_timestamp( 

380 msg: str, datetime_timestamp: datetime.datetime | None = None 

381) -> NoReturn: 

382 """Suspend execution with an optional target resume timestamp.""" 

383 

384 if datetime_timestamp is None: 

385 msg = f"No timestamp provided. Suspending without retry timestamp. Original operation: [{msg}]" 

386 raise SuspendExecution(msg) 

387 

388 if datetime_timestamp < datetime.datetime.now(tz=datetime.timezone.utc): 

389 msg = f"Invalid timestamp {datetime_timestamp}, suspending with immediate retry, original operation: [{msg}]" 

390 raise TimedSuspendExecution.from_datetime( 

391 msg, datetime.datetime.now(tz=datetime.timezone.utc) 

392 ) 

393 

394 raise TimedSuspendExecution.from_datetime(msg, datetime_timestamp) 

395 

396 

397def suspend_with_optional_resume_delay( 

398 msg: str, delay_seconds: int | None = None 

399) -> NoReturn: 

400 """Suspend execution with an optional delay before resuming.""" 

401 

402 if delay_seconds is None: 

403 msg = f"No delay_seconds provided, suspending without retry timestamp, original operation: [{msg}]" 

404 raise SuspendExecution(msg) 

405 

406 if delay_seconds < 0: 

407 msg = f"Invalid delay_seconds {delay_seconds}, suspending with delay 0, original operation: [{msg}]" 

408 raise TimedSuspendExecution.from_delay(msg, 0) 

409 

410 raise TimedSuspendExecution.from_delay(msg, delay_seconds) 

411 

412 

413@dataclass(frozen=True) 

414class CallableRuntimeErrorSerializableDetails: 

415 """Serializable error details.""" 

416 

417 type: str 

418 message: str 

419 

420 @classmethod 

421 def from_exception( 

422 cls, exception: Exception 

423 ) -> CallableRuntimeErrorSerializableDetails: 

424 """Create an instance from an Exception, using its type and message. 

425 

426 Args: 

427 exception: An Exception instance 

428 

429 Returns: 

430 A CallableRuntimeErrorDetails instance with the exception's type name and message 

431 """ 

432 return cls(type=exception.__class__.__name__, message=str(exception)) 

433 

434 def __str__(self) -> str: 

435 """ 

436 Return a string representation of the object. 

437 

438 Returns: 

439 A string in the format "type: message" 

440 """ 

441 return f"{self.type}: {self.message}" 

442 

443 

444class SerDesError(DurableExecutionsError): 

445 """Raised when serialization fails."""