Coverage for async_durable_execution/filesystem_serdes.py: 86%
420 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1"""Composable SerDes stage backed by a durable shared filesystem."""
3from __future__ import annotations
5import asyncio
6import errno
7import hashlib
8import inspect
9import json
10import os
11import re
12import stat
13import uuid
14from collections.abc import Awaitable, Callable
15from concurrent.futures import ThreadPoolExecutor
16from dataclasses import dataclass
17from enum import Enum
18from pathlib import Path
19from typing import Any
20from urllib.parse import quote
22from ._core.context import SerDesContext
23from ._core.exceptions import RetryableSerDesError, SerDesError
24from ._core.models import OperationType
25from .preview import PreviewConfig, build_preview
27_ENVELOPE_MARKER = "__durable_execution_filesystem_serdes"
28_ENVELOPE_VERSION = 1
29_PAYLOAD_TYPE = "STRING"
30_DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES = 256 * 1024 - 1024
31_DURABLE_EXECUTION_ARN_PATTERN = re.compile(
32 r"^arn:([^:]+):lambda:([^:]+):([^:]+):function:"
33 r"([^:/]+):([^:/]+)/durable-execution/([^/]+)/([^/]+)$"
34)
35_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
36_NON_RETRYABLE_FILESYSTEM_ERRNOS = {
37 errno.EACCES,
38 getattr(errno, "EDQUOT", errno.ENOSPC),
39 errno.EFBIG,
40 errno.ELOOP,
41 errno.EINVAL,
42 errno.EISDIR,
43 errno.ENAMETOOLONG,
44 errno.ENOSPC,
45 errno.ENOTDIR,
46 getattr(errno, "EOPNOTSUPP", errno.EINVAL),
47 getattr(errno, "ENOTSUP", getattr(errno, "EOPNOTSUPP", errno.EINVAL)),
48 errno.EPERM,
49 errno.EROFS,
50}
51_FILESYSTEM_EXECUTOR = ThreadPoolExecutor(
52 max_workers=4,
53 thread_name_prefix="durable-filesystem-serdes",
54)
56PreviewGenerator = Callable[
57 [str, SerDesContext],
58 dict[str, Any] | None | Awaitable[dict[str, Any] | None],
59]
60CrossExecutionReferencePolicy = Callable[[str, str, SerDesContext], bool]
63class FileSystemSerDesMode(str, Enum):
64 """Controls when payload strings are offloaded to the filesystem."""
66 ALWAYS = "ALWAYS"
67 OVERFLOW = "OVERFLOW"
70class FileSystemPathEncoding(str, Enum):
71 """Controls how execution and entity identifiers appear in paths."""
73 URI = "URI"
74 HASH = "HASH"
77@dataclass(frozen=True)
78class FileSystemSerDesStageConfig:
79 """Configuration for :class:`FileSystemSerDesStage`."""
81 storage_mode: FileSystemSerDesMode = FileSystemSerDesMode.ALWAYS
82 path_encoding: FileSystemPathEncoding = FileSystemPathEncoding.URI
83 checkpoint_envelope_limit_bytes: int = _DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES
84 generate_preview: PreviewGenerator | None = None
85 preview_config: PreviewConfig | None = None
86 cross_execution_reference_policy: CrossExecutionReferencePolicy | None = None
88 def __post_init__(self) -> None:
89 if not isinstance(self.storage_mode, FileSystemSerDesMode):
90 msg = "storage_mode must be a FileSystemSerDesMode."
91 raise TypeError(msg)
92 if not isinstance(self.path_encoding, FileSystemPathEncoding):
93 msg = "path_encoding must be a FileSystemPathEncoding."
94 raise TypeError(msg)
95 if self.checkpoint_envelope_limit_bytes <= 0:
96 msg = "checkpoint_envelope_limit_bytes must be positive."
97 raise ValueError(msg)
98 if self.generate_preview is not None and self.preview_config is not None:
99 msg = "Configure either generate_preview or preview_config, not both."
100 raise ValueError(msg)
101 if self.cross_execution_reference_policy is not None and not callable(
102 self.cross_execution_reference_policy
103 ):
104 msg = "cross_execution_reference_policy must be callable."
105 raise TypeError(msg)
108class FileSystemSerDesStage:
109 """Store a pipeline string on a durable shared filesystem.
111 Do not use Lambda's ephemeral ``/tmp`` directory. Use a shared durable
112 mount such as Amazon EFS or S3 Files.
114 Payload files are immutable and uniquely named. File contents and
115 directory metadata are synchronized before the versioned checkpoint
116 envelope is returned. The envelope records ownership and a SHA-256 digest.
117 Unrecognized input passes through unchanged.
118 """
120 def __init__(
121 self,
122 base_path: str | os.PathLike[str],
123 config: FileSystemSerDesStageConfig | None = None,
124 ) -> None:
125 if not os.fspath(base_path):
126 msg = "base_path must not be empty."
127 raise ValueError(msg)
128 self._base_path = Path(os.path.abspath(os.fspath(base_path)))
129 self._config = config or FileSystemSerDesStageConfig()
131 def execution_directory(self, durable_execution_arn: str) -> Path:
132 """Return the directory eligible for cleanup after history retention."""
133 return self._resolve_execution_directory(durable_execution_arn)
135 async def serialize(self, value: str, context: SerDesContext) -> str:
136 """Store or inline ``value`` and return a versioned envelope."""
137 self._require_context(context)
138 payload = value.encode("utf-8")
139 digest = hashlib.sha256(payload).hexdigest()
140 payload_size = len(payload)
142 if self._config.storage_mode is FileSystemSerDesMode.OVERFLOW:
143 inline_envelope = self._encode_envelope(
144 context=context,
145 digest=digest,
146 payload_size=payload_size,
147 data=value,
148 )
149 if self._fits_checkpoint(inline_envelope):
150 return inline_envelope
152 file_path = self._resolve_payload_path(context, digest)
153 preview = await self._generate_preview(value, context)
154 file_envelope = self._encode_envelope(
155 context=context,
156 digest=digest,
157 payload_size=payload_size,
158 file_path=file_path,
159 preview=preview,
160 )
161 if not self._fits_checkpoint(file_envelope):
162 msg = (
163 "Filesystem SerDes envelope exceeds the checkpoint payload "
164 f"limit for entity {self._entity_id(context)!r}."
165 )
166 raise SerDesError(msg)
168 try:
169 await asyncio.get_running_loop().run_in_executor(
170 _FILESYSTEM_EXECUTOR,
171 _write_payload,
172 self._base_path,
173 file_path,
174 payload,
175 )
176 except OSError as error:
177 msg = (
178 "Failed to store filesystem payload for entity "
179 f"{self._entity_id(context)!r}."
180 )
181 _raise_filesystem_error(msg, error)
182 return file_envelope
184 async def deserialize(self, data: str, context: SerDesContext) -> str:
185 """Resolve a recognized filesystem envelope or pass input through."""
186 try:
187 envelope = _strict_json_loads(data)
188 except (TypeError, ValueError, json.JSONDecodeError) as error:
189 if _contains_top_level_marker(data):
190 self._require_context(context)
191 raise self._malformed_envelope(context) from error
192 return data
194 if not isinstance(envelope, dict) or _ENVELOPE_MARKER not in envelope:
195 return data
197 self._require_context(context)
198 version = envelope.get(_ENVELOPE_MARKER)
199 if isinstance(version, bool) or not isinstance(version, int): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 raise self._malformed_envelope(context)
201 if version != _ENVELOPE_VERSION:
202 msg = (
203 f"Unsupported filesystem SerDes envelope version {version} "
204 f"for entity {self._entity_id(context)!r}."
205 )
206 raise SerDesError(msg)
208 parsed = self._validate_envelope(envelope, context)
209 digest = parsed["payloadDigest"]
210 payload_size = parsed["payloadSizeBytes"]
211 owner_arn = parsed["ownerDurableExecutionArn"]
212 owner_entity_id = parsed["ownerEntityId"]
213 self._validate_owner(owner_arn, owner_entity_id, context)
215 inline_data = parsed.get("data")
216 if isinstance(inline_data, str):
217 inline_payload = inline_data.encode("utf-8")
218 self._verify_payload_size(inline_payload, payload_size, context)
219 self._verify_digest(inline_payload, digest, context)
220 return inline_data
222 file_path = self._validate_file_path(
223 parsed["file"],
224 owner_arn,
225 owner_entity_id,
226 digest,
227 )
228 try:
229 payload = await asyncio.get_running_loop().run_in_executor(
230 _FILESYSTEM_EXECUTOR,
231 _read_payload,
232 self._base_path,
233 file_path,
234 payload_size,
235 )
236 except OSError as error:
237 msg = (
238 "Failed to load filesystem payload for entity "
239 f"{self._entity_id(context)!r}."
240 )
241 _raise_filesystem_error(msg, error, missing_is_permanent=True)
242 self._verify_digest(payload, digest, context)
243 try:
244 return payload.decode("utf-8")
245 except UnicodeDecodeError as error:
246 raise self._malformed_envelope(context) from error
248 async def _generate_preview(
249 self,
250 value: str,
251 context: SerDesContext,
252 ) -> dict[str, Any] | None:
253 generator = self._config.generate_preview
254 try:
255 if generator is not None:
256 preview = generator(value, context)
257 if inspect.isawaitable(preview): 257 ↛ 275line 257 didn't jump to line 275 because the condition on line 257 was always true
258 preview = await preview
259 elif self._config.preview_config is not None:
260 preview = build_preview(
261 json.loads(value),
262 self._config.preview_config,
263 )
264 else:
265 preview = None
266 except RetryableSerDesError:
267 raise
268 except Exception as error:
269 msg = (
270 "Failed to generate filesystem payload preview for entity "
271 f"{self._entity_id(context)!r}."
272 )
273 raise SerDesError(msg) from error
275 if preview is not None and not isinstance(preview, dict):
276 msg = "Filesystem SerDes preview generator must return a dict or None."
277 raise SerDesError(msg)
278 return preview
280 def _encode_envelope(
281 self,
282 *,
283 context: SerDesContext,
284 digest: str,
285 payload_size: int,
286 data: str | None = None,
287 file_path: Path | None = None,
288 preview: dict[str, Any] | None = None,
289 ) -> str:
290 envelope: dict[str, Any] = {
291 _ENVELOPE_MARKER: _ENVELOPE_VERSION,
292 "ownerDurableExecutionArn": context.durable_execution_arn,
293 "ownerEntityId": self._entity_id(context),
294 "payloadType": _PAYLOAD_TYPE,
295 "payloadDigest": digest,
296 "payloadSizeBytes": payload_size,
297 }
298 if data is not None:
299 envelope["data"] = data
300 elif file_path is not None: 300 ↛ 305line 300 didn't jump to line 305 because the condition on line 300 was always true
301 envelope["file"] = str(file_path)
302 if preview is not None:
303 envelope["preview"] = preview
304 else:
305 msg = "Filesystem SerDes envelope requires data or a file."
306 raise SerDesError(msg)
308 try:
309 return json.dumps(
310 envelope,
311 ensure_ascii=False,
312 separators=(",", ":"),
313 )
314 except (TypeError, ValueError) as error:
315 msg = (
316 "Failed to encode filesystem payload envelope for entity "
317 f"{self._entity_id(context)!r}."
318 )
319 raise SerDesError(msg) from error
321 def _validate_envelope(
322 self,
323 envelope: dict[str, Any],
324 context: SerDesContext,
325 ) -> dict[str, Any]:
326 has_data = isinstance(envelope.get("data"), str)
327 has_file = isinstance(envelope.get("file"), str)
328 has_preview = "preview" in envelope
329 expected_keys = {
330 _ENVELOPE_MARKER,
331 "ownerDurableExecutionArn",
332 "ownerEntityId",
333 "payloadType",
334 "payloadDigest",
335 "payloadSizeBytes",
336 "data" if has_data else "file",
337 }
338 if has_preview:
339 expected_keys.add("preview")
341 valid = (
342 has_data != has_file
343 and set(envelope) == expected_keys
344 and isinstance(envelope.get("ownerDurableExecutionArn"), str)
345 and bool(envelope.get("ownerDurableExecutionArn"))
346 and isinstance(envelope.get("ownerEntityId"), str)
347 and bool(envelope.get("ownerEntityId"))
348 and envelope.get("payloadType") == _PAYLOAD_TYPE
349 and isinstance(envelope.get("payloadDigest"), str)
350 and bool(_SHA256_PATTERN.fullmatch(envelope["payloadDigest"]))
351 and isinstance(envelope.get("payloadSizeBytes"), int)
352 and not isinstance(envelope.get("payloadSizeBytes"), bool)
353 and envelope["payloadSizeBytes"] >= 0
354 and (
355 not has_preview or (has_file and isinstance(envelope["preview"], dict))
356 )
357 )
358 if not valid: 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true
359 raise self._malformed_envelope(context)
360 return envelope
362 def _validate_owner(
363 self,
364 owner_arn: str,
365 owner_entity_id: str,
366 context: SerDesContext,
367 ) -> None:
368 same_owner = (
369 owner_arn == context.durable_execution_arn
370 and owner_entity_id == self._entity_id(context)
371 )
372 operation_type = context.operation_type
373 operation_type_value = (
374 operation_type.value
375 if isinstance(operation_type, OperationType)
376 else operation_type
377 )
378 if same_owner:
379 return
381 policy = self._config.cross_execution_reference_policy
382 if (
383 operation_type_value == OperationType.CHAINED_INVOKE.value
384 and policy is not None
385 ):
386 try:
387 decision = policy(owner_arn, owner_entity_id, context)
388 if inspect.isawaitable(decision):
389 if inspect.iscoroutine(decision): 389 ↛ 391line 389 didn't jump to line 391 because the condition on line 389 was always true
390 decision.close()
391 msg = (
392 "Filesystem SerDes cross-execution policy must return "
393 "a bool synchronously."
394 )
395 raise SerDesError(msg)
396 if not isinstance(decision, bool):
397 msg = "Filesystem SerDes cross-execution policy must return a bool."
398 raise SerDesError(msg)
399 if decision: 399 ↛ 407line 399 didn't jump to line 407 because the condition on line 399 was always true
400 return
401 except SerDesError:
402 raise
403 except Exception as error:
404 msg = "Filesystem SerDes cross-execution policy failed."
405 raise SerDesError(msg) from error
407 msg = "Filesystem SerDes file belongs to a different durable entity."
408 raise SerDesError(msg)
410 def _validate_file_path(
411 self,
412 file_value: str,
413 owner_arn: str,
414 owner_entity_id: str,
415 digest: str,
416 ) -> Path:
417 file_path = Path(os.path.abspath(file_value))
418 expected_directory = self._resolve_execution_directory(owner_arn)
419 encoded_entity = self._encode(owner_entity_id)
420 expected_name = re.compile(
421 rf"^{re.escape(encoded_entity)}-{digest}-"
422 r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-"
423 r"[0-9a-f]{4}-[0-9a-f]{12}\.payload$"
424 )
425 if file_path.parent != expected_directory or not expected_name.fullmatch( 425 ↛ 428line 425 didn't jump to line 428 because the condition on line 425 was never true
426 file_path.name
427 ):
428 msg = "Filesystem SerDes file is not valid for its declared entity."
429 raise SerDesError(msg)
430 return file_path
432 def _verify_digest(
433 self,
434 payload: bytes,
435 expected_digest: str,
436 context: SerDesContext,
437 ) -> None:
438 if hashlib.sha256(payload).hexdigest() != expected_digest:
439 msg = (
440 "Filesystem SerDes payload digest does not match stored "
441 f"content for entity {self._entity_id(context)!r}."
442 )
443 raise SerDesError(msg)
445 def _verify_payload_size(
446 self,
447 payload: bytes,
448 expected_size: int,
449 context: SerDesContext,
450 ) -> None:
451 if len(payload) != expected_size: 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 msg = (
453 "Filesystem SerDes payload size does not match the envelope "
454 f"for entity {self._entity_id(context)!r}."
455 )
456 raise SerDesError(msg)
458 def _resolve_payload_path(
459 self,
460 context: SerDesContext,
461 digest: str,
462 ) -> Path:
463 directory = self._resolve_execution_directory(context.durable_execution_arn)
464 filename = (
465 f"{self._encode(self._entity_id(context))}-{digest}-{uuid.uuid4()}.payload"
466 )
467 return directory / filename
469 def _resolve_execution_directory(self, durable_execution_arn: str) -> Path:
470 if not durable_execution_arn.strip():
471 msg = "durable_execution_arn must not be empty."
472 raise SerDesError(msg)
474 if self._config.path_encoding is FileSystemPathEncoding.URI:
475 match = _DURABLE_EXECUTION_ARN_PATTERN.fullmatch(durable_execution_arn)
476 if match is not None:
477 directory = self._base_path.joinpath(
478 *(self._encode(part) for part in match.groups())
479 )
480 return self._require_strict_descendant(directory)
481 directory = self._base_path / self._encode(durable_execution_arn)
482 return self._require_strict_descendant(directory)
484 def _encode(self, value: str) -> str:
485 if self._config.path_encoding is FileSystemPathEncoding.HASH:
486 return hashlib.sha256(value.encode("utf-8")).hexdigest()
487 encoded = quote(value, safe="-._~")
488 if encoded in {".", ".."}:
489 return "".join(f"%{byte:02X}" for byte in value.encode("utf-8"))
490 return encoded
492 def _require_strict_descendant(self, directory: Path) -> Path:
493 normalized = Path(os.path.abspath(directory))
494 try:
495 relative = normalized.relative_to(self._base_path)
496 except ValueError as error:
497 msg = "Filesystem SerDes execution directory is outside the base path."
498 raise SerDesError(msg) from error
499 if not relative.parts: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 msg = "Filesystem SerDes execution directory must be below the base path."
501 raise SerDesError(msg)
502 return normalized
504 def _fits_checkpoint(self, envelope: str) -> bool:
505 return (
506 len(envelope.encode("utf-8"))
507 <= self._config.checkpoint_envelope_limit_bytes
508 )
510 def _require_context(self, context: SerDesContext) -> None:
511 if not context.durable_execution_arn or not self._entity_id(context):
512 msg = (
513 "FileSystemSerDesStage requires an SDK-managed SerDesContext "
514 "with durable_execution_arn and entity_id."
515 )
516 raise SerDesError(msg)
518 @staticmethod
519 def _entity_id(context: SerDesContext) -> str:
520 if context.entity_id: 520 ↛ 522line 520 didn't jump to line 522 because the condition on line 520 was always true
521 return context.entity_id
522 if context.operation_id:
523 return f"operation/{context.operation_id}"
524 return ""
526 def _malformed_envelope(self, context: SerDesContext) -> SerDesError:
527 return SerDesError(
528 "Invalid filesystem SerDes envelope for entity "
529 f"{self._entity_id(context)!r}."
530 )
533def create_file_system_serdes_stage(
534 base_path: str | os.PathLike[str],
535 config: FileSystemSerDesStageConfig | None = None,
536) -> FileSystemSerDesStage:
537 """Create a filesystem stage for a composable SerDes pipeline."""
538 return FileSystemSerDesStage(base_path, config)
541def _raise_filesystem_error(
542 message: str,
543 error: OSError,
544 *,
545 missing_is_permanent: bool = False,
546) -> None:
547 if error.errno in _NON_RETRYABLE_FILESYSTEM_ERRNOS or (
548 missing_is_permanent and error.errno == errno.ENOENT
549 ):
550 raise SerDesError(message) from error
551 raise RetryableSerDesError(message) from error
554def _strict_json_loads(data: str) -> Any:
555 def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
556 result: dict[str, Any] = {}
557 for key, value in pairs:
558 if key in result:
559 msg = f"Duplicate JSON object key: {key}"
560 raise ValueError(msg)
561 result[key] = value
562 return result
564 return json.loads(data, object_pairs_hook=reject_duplicate_keys)
567def _contains_top_level_marker(data: str) -> bool:
568 index = 0
569 while index < len(data) and data[index].isspace(): 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true
570 index += 1
571 if index == len(data) or data[index] != "{":
572 return False
574 depth = 1
575 index += 1
576 while index < len(data) and depth > 0:
577 current = data[index]
578 if current in "{[":
579 depth += 1
580 index += 1
581 continue
582 if current in "}]": 582 ↛ 583line 582 didn't jump to line 583 because the condition on line 582 was never true
583 depth -= 1
584 index += 1
585 continue
586 if current != '"':
587 index += 1
588 continue
590 literal_start = index
591 index += 1
592 escaped = False
593 while index < len(data): 593 ↛ 602line 593 didn't jump to line 602 because the condition on line 593 was always true
594 current = data[index]
595 if escaped: 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true
596 escaped = False
597 elif current == "\\": 597 ↛ 598line 597 didn't jump to line 598 because the condition on line 597 was never true
598 escaped = True
599 elif current == '"':
600 break
601 index += 1
602 if index == len(data): 602 ↛ 603line 602 didn't jump to line 603 because the condition on line 602 was never true
603 return False
605 literal_end = index
606 delimiter = index + 1
607 while delimiter < len(data) and data[delimiter].isspace(): 607 ↛ 608line 607 didn't jump to line 608 because the condition on line 607 was never true
608 delimiter += 1
609 if depth == 1 and delimiter < len(data) and data[delimiter] == ":":
610 try:
611 field_name = json.loads(data[literal_start : literal_end + 1])
612 except json.JSONDecodeError:
613 return False
614 if field_name == _ENVELOPE_MARKER:
615 return True
616 index += 1
617 return False
620def _directory_open_flags() -> int:
621 return os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
624def _open_directory_path(
625 path: Path,
626 *,
627 create: bool,
628 required_root: Path | None = None,
629) -> int:
630 if not path.is_absolute(): 630 ↛ 631line 630 didn't jump to line 631 because the condition on line 630 was never true
631 msg = "Filesystem SerDes paths must be absolute."
632 raise SerDesError(msg)
634 if required_root is None:
635 root = Path(path.anchor)
636 parts = path.parts[1:]
637 directory_fd = os.open(root, _directory_open_flags())
638 else:
639 try:
640 relative = path.relative_to(required_root)
641 except ValueError as error:
642 msg = "Filesystem SerDes directory is outside the configured base path."
643 raise SerDesError(msg) from error
644 try:
645 directory_fd = _open_directory_path(required_root, create=False)
646 except FileNotFoundError as error:
647 msg = "Filesystem SerDes configured base path does not exist."
648 raise SerDesError(msg) from error
649 parts = relative.parts
651 try:
652 for part in parts:
653 if create:
654 try:
655 os.mkdir(part, mode=0o700, dir_fd=directory_fd)
656 except FileExistsError:
657 pass
658 os.fsync(directory_fd)
659 next_fd = os.open(part, _directory_open_flags(), dir_fd=directory_fd)
660 os.close(directory_fd)
661 directory_fd = next_fd
662 return directory_fd
663 except BaseException:
664 os.close(directory_fd)
665 raise
668def _write_payload(base_path: Path, file_path: Path, payload: bytes) -> None:
669 directory_fd = _open_directory_path(
670 file_path.parent,
671 create=True,
672 required_root=base_path,
673 )
674 file_fd: int | None = None
675 created = False
676 try:
677 flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
678 file_fd = os.open(
679 file_path.name,
680 flags,
681 mode=0o600,
682 dir_fd=directory_fd,
683 )
684 created = True
685 view = memoryview(payload)
686 while view:
687 written = os.write(file_fd, view)
688 view = view[written:]
689 os.fsync(file_fd)
690 os.fsync(directory_fd)
691 except BaseException:
692 if created:
693 try:
694 os.unlink(file_path.name, dir_fd=directory_fd)
695 except OSError:
696 pass
697 raise
698 finally:
699 if file_fd is not None: 699 ↛ 701line 699 didn't jump to line 701 because the condition on line 699 was always true
700 os.close(file_fd)
701 os.close(directory_fd)
704def _read_payload(base_path: Path, file_path: Path, expected_size: int) -> bytes:
705 directory_fd = _open_directory_path(
706 file_path.parent,
707 create=False,
708 required_root=base_path,
709 )
710 file_fd: int | None = None
711 try:
712 file_fd = os.open(
713 file_path.name,
714 os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
715 dir_fd=directory_fd,
716 )
717 file_stat = os.fstat(file_fd)
718 if not stat.S_ISREG(file_stat.st_mode): 718 ↛ 719line 718 didn't jump to line 719 because the condition on line 718 was never true
719 msg = "Filesystem SerDes envelope does not reference a regular file."
720 raise SerDesError(msg)
721 if file_stat.st_size != expected_size:
722 msg = "Filesystem SerDes payload size does not match the envelope."
723 raise SerDesError(msg)
725 chunks: list[bytes] = []
726 remaining = expected_size
727 while remaining:
728 chunk = os.read(file_fd, min(remaining, 1024 * 1024))
729 if not chunk: 729 ↛ 730line 729 didn't jump to line 730 because the condition on line 729 was never true
730 msg = "Filesystem SerDes payload ended before its declared size."
731 raise SerDesError(msg)
732 chunks.append(chunk)
733 remaining -= len(chunk)
734 if os.read(file_fd, 1): 734 ↛ 735line 734 didn't jump to line 735 because the condition on line 734 was never true
735 msg = "Filesystem SerDes payload exceeds its declared size."
736 raise SerDesError(msg)
737 return b"".join(chunks)
738 finally:
739 if file_fd is not None:
740 os.close(file_fd)
741 os.close(directory_fd)