"""Profile-owned flat-root storage capability for weekly operations.""" from __future__ import annotations import fcntl import os import sys from threading import get_ident from types import TracebackType from typing import Final, Never, Self from checkin_cli.weekly_operations import WeeklyOperationsAuthorityCompromise, WeeklyOperationsCorruption, WeeklyOperationsError, WeeklyOperationsInputError from checkin_cli.weekly_operations_layout import MARKER_NAME from checkin_cli.weekly_operations_canonical_registry_history import REGISTRY_NAME from checkin_cli.weekly_operations_parent import AuthorityId, WeeklyOperationsAuthorityBinding, WeeklyOperationsParentAuthority, authority_marker_bytes, canonical_authority_marker, construct_authority_binding, validate_authority_binding from checkin_cli.weekly_operations_publish import MarkerNameOccupied, PreparedMarker, classify_existing_marker, commit_authority_marker, prepare_authority_marker from checkin_cli.weekly_operations_authority_root import ( WeeklyOperationsAuthorityRoot as WeeklyOperationsAuthorityRoot, verify_authority_descriptor, ) from checkin_cli.weekly_operations_registry_bootstrap import EMPTY_REGISTRY_DIGEST, PreparedRegistry, commit_registry, prepare_registry, remove_owned_registry from checkin_cli.weekly_operations_signals import InitializationCancellation, InitializationCommitContext, InitializationSignalGuard, WeeklyOperationsInitializationCommitted __all__: Final = ( "AuthorityId", "WeeklyOperationsAuthorityBinding", "WeeklyOperationsInitializationCommitted", "WeeklyOperationsInitializationTransaction", "begin_authority_initialization", "construct_authority_binding", ) class WeeklyOperationsInitializationTransaction: """Same-thread binding handoff that keeps catchable cancellation blocked.""" __slots__: Final = ("_parent", "_authority_id", "_creator", "_guard", "_entered", "_active", "_acknowledged", "_binding", "_authority") def __init__(self, parent: WeeklyOperationsParentAuthority, authority_id: AuthorityId) -> None: if len(authority_id) != 64 or any(character not in "0123456789abcdef" for character in authority_id): raise WeeklyOperationsInputError("authority id") self._parent: WeeklyOperationsParentAuthority = parent self._authority_id: AuthorityId = authority_id self._creator: int = get_ident() self._guard: InitializationSignalGuard = InitializationSignalGuard() self._entered: bool = False self._active: bool = False self._acknowledged: bool = False self._binding: WeeklyOperationsAuthorityBinding | None = None self._authority: WeeklyOperationsAuthorityRoot | None = None def _require_creator(self) -> None: if get_ident() != self._creator: raise WeeklyOperationsInputError("initialization transaction thread mismatch") def _require_active(self) -> None: self._require_creator() if not self._active: raise WeeklyOperationsInputError("initialization transaction is inactive") def __enter__(self) -> Self: self._require_creator() if self._entered: raise WeeklyOperationsInputError("initialization transaction cannot be reused") self._entered = True self._guard.block() self._active = True fcntl.flock(self._parent.descriptor, fcntl.LOCK_EX) prepared: PreparedMarker | None = None registry_prepared: PreparedRegistry | None = None descriptor: int | None = None completed = False occupied = False cancellation: InitializationCancellation | None = None cleanup_failure: BaseException | None = None try: try: if MARKER_NAME in os.listdir(self._parent.descriptor): occupied = True raise MarkerNameOccupied("authority marker name is occupied") registry_prepared = prepare_registry(self._parent) registry = commit_registry(self._parent, registry_prepared) payload = authority_marker_bytes(self._authority_id, registry) prepared = prepare_authority_marker( self._parent, payload, canonical_authority_marker, allowed=frozenset((REGISTRY_NAME,)), ) descriptor = os.dup(self._parent.descriptor) published = commit_authority_marker(self._parent, prepared) self._binding = construct_authority_binding( self._authority_id, os.fstat(descriptor), published.marker, published.marker_digest, registry, EMPTY_REGISTRY_DIGEST, ) self._authority = WeeklyOperationsAuthorityRoot(self._parent, descriptor, self._binding) descriptor = None os.close(prepared.descriptor) prepared = None os.close(registry_prepared.descriptor) registry_prepared = None completed = True except MarkerNameOccupied: occupied = True finally: fcntl.flock(self._parent.descriptor, fcntl.LOCK_UN) if not completed: if descriptor is not None: os.close(descriptor) if prepared is not None: os.close(prepared.descriptor) if registry_prepared is not None: if MARKER_NAME not in os.listdir(self._parent.descriptor): try: remove_owned_registry(self._parent, registry_prepared) except (OSError, WeeklyOperationsError) as cleanup_error: cleanup_failure = cleanup_error os.close(registry_prepared.descriptor) self._active = False if occupied: cancellation = self._guard.restore() else: error = ( cleanup_failure if isinstance(cleanup_failure, WeeklyOperationsAuthorityCompromise) else sys.exception() or cleanup_failure or WeeklyOperationsCorruption("initialization transaction enter failed") ) _ = self._guard.restore(error) if occupied: try: classify_existing_marker(self._parent.descriptor, canonical_authority_marker) except WeeklyOperationsError as error: if cancellation is not None: raise error from cancellation raise raise AssertionError("existing marker classifier returned") return self @property def binding(self) -> WeeklyOperationsAuthorityBinding: self._require_active() if self._binding is None: raise WeeklyOperationsInputError("initialization binding is unavailable") return self._binding @property def authority(self) -> WeeklyOperationsAuthorityRoot: self._require_active() if self._authority is None: raise WeeklyOperationsInputError("initialized authority is unavailable") return self._authority def acknowledge_binding(self) -> None: self._require_active() _ = self.binding self._acknowledged = True def __exit__(self, _kind: type[BaseException] | None, error: BaseException | None, _traceback: TracebackType | None) -> bool: self._require_active() binding = self.binding authority = self.authority self._active = False if not self._acknowledged: authority.close() self._authority = None if self._acknowledged: cancellation = self._guard.restore(error) if cancellation is not None: raise cancellation return False cancellation = self._guard.restore() original = error if original is not None: context = InitializationCommitContext.BODY_EXCEPTION elif cancellation is not None: context = InitializationCommitContext.CANCELLATION else: context = InitializationCommitContext.MISSING_ACKNOWLEDGEMENT committed = WeeklyOperationsInitializationCommitted( binding, context, None if cancellation is None else cancellation.cancellation_kind, original, ) if original is not None: raise committed from original if cancellation is not None: raise committed from cancellation raise committed def __copy__(self) -> Never: raise WeeklyOperationsInputError("initialization transaction cannot be copied") def __deepcopy__(self, _memo: dict[int, Self]) -> Never: raise WeeklyOperationsInputError("initialization transaction cannot be copied") def begin_authority_initialization(parent: WeeklyOperationsParentAuthority, authority_id: AuthorityId) -> WeeklyOperationsInitializationTransaction: """Construct an inert transaction; signal blocking starts only on enter.""" return WeeklyOperationsInitializationTransaction(parent, authority_id) def open_authority_root(parent: WeeklyOperationsParentAuthority, binding: WeeklyOperationsAuthorityBinding) -> WeeklyOperationsAuthorityRoot: """Open the caller-bound dedicated authority root.""" validate_authority_binding(binding) with parent.locked(): descriptor = os.dup(parent.descriptor) try: verify_authority_descriptor(descriptor, binding) return WeeklyOperationsAuthorityRoot(parent, descriptor, binding) except WeeklyOperationsCorruption: os.close(descriptor) raise from checkin_cli.weekly_operations_authority_access import ( LegacySidecarInventory as LegacySidecarInventory, WeeklyOperationsMigrationRequired as WeeklyOperationsMigrationRequired, discover_legacy_sidecars as discover_legacy_sidecars, issue_repair_authority as issue_repair_authority, open_authority_root_for_repair as open_authority_root_for_repair, )