"""Registered-canonical descriptor authority for the schedule reminder ledger.""" from __future__ import annotations from anyio import CancelScope from anyio.to_thread import run_sync import fcntl import hashlib import json import os import threading from collections.abc import AsyncGenerator, Generator from contextlib import asynccontextmanager, contextmanager from dataclasses import dataclass, replace from datetime import date from pathlib import Path from typing import Final, Literal, Never, assert_never, final from .customer_schedule import ( abandon_missing_checkin_reminder_for_terminal_morning_response, initialize_schedule_delivery_fence, mark_customer_task_abandoned, mark_customer_task_delivered, mark_customer_task_known_failure, mark_customer_task_sent_audited, mark_customer_task_unknown, mark_weekly_reminder_sending, reserve_missing_checkin_reminder, schedule_delivery_ledger, ) from .weekly_operations_schedule_host_authority_r4 import ( weekly_schedule_authority_fence, weekly_schedule_descriptor_root, weekly_schedule_provider_admission, ) from .weekly_operations_schedule_host_models_r4 import ( ScheduledDeliveryReceipt, WeeklyReminderReservationAuthority, ) from .weekly_operations import CustomerIdentityDigest, WeeklyOperationsConflict from .weekly_reminder_ledger_identity import ( canonical_root_identity, directory_identity, file_identity, verify_named, ) from .weekly_operations_customer_authority import CanonicalCheckinCustomerAuthority from .weekly_operations_store import WeeklyOperationsStore _SCHEMA: Final = "weekly-reminder-ledger-authority-v2" _DATA: Final = "data" _LEDGER: Final = "scheduled-deliveries.jsonl" _LOCK: Final = ".scheduled-deliveries.lock" _CLAIMS: Final = "customer-schedule-claims" _DIR_FLAGS: Final = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC _PROCESS_LOCKS: dict[tuple[int, int], threading.Lock] = {} _PROCESS_LOCKS_GUARD: Final = threading.Lock() @dataclass(frozen=True, slots=True) class WeeklyReminderLedgerBinding: schema: str customer_identity_digest: CustomerIdentityDigest registration_row_digest: str registered_binding_digest: str canonical_root_identity: tuple[int, int, int, int, int] profile_root_identity: tuple[int, int, int, int, int] data_root_identity: tuple[int, int, int, int, int] sidecar_authority_digest: str sidecar_customer_digest: CustomerIdentityDigest ledger_identity: tuple[int, int, int, int, int] lock_identity: tuple[int, int, int, int, int] binding_digest: str @dataclass(frozen=True, slots=True) class LedgerReservation: customer_identity_digest: CustomerIdentityDigest day: date destination: dict[str, str | None] registry_digest: str config_digest: str authority_digest: str canonical_sequence: int canonical_digest: str weekly_authority: WeeklyReminderReservationAuthority @dataclass(frozen=True, slots=True) class LedgerTransition: action: Literal["abandon", "answered", "sending", "known_failure", "unknown", "delivered", "audited"] receipt: ScheduledDeliveryReceipt reason: str = "" provider_receipt: str = "" message_id: str = "" authority: WeeklyReminderReservationAuthority | None = None @final class WeeklyReminderLedgerAuthority: """No-path capability bound to registered canonical and Todo 3 authorities.""" def __init__( self, binding: WeeklyReminderLedgerBinding, source: CanonicalCheckinCustomerAuthority, store: WeeklyOperationsStore, profile_fd: int, data_fd: int, ledger_fd: int, lock_fd: int, ) -> None: self.binding = binding self._source = source self._store = store self._profile_fd = profile_fd self._data_fd = data_fd self._ledger_fd = ledger_fd self._lock_fd = lock_fd self.verify() @property def customer_identity_digest(self) -> CustomerIdentityDigest: return self.binding.customer_identity_digest @property def binding_digest(self) -> str: return self.binding.binding_digest def verify(self) -> None: self._source.verify() self._store.authority.verify() registered = self._source.registered_binding if ( registered.registration_row_digest != self.binding.registration_row_digest or registered.binding_digest != self.binding.registered_binding_digest or registered.authority_binding_digest != self.binding.sidecar_authority_digest or self._source.customer_identity_digest != self.binding.customer_identity_digest or self._store.customer_identity_digest != self.binding.sidecar_customer_digest or self._store.authority_binding_digest != self.binding.sidecar_authority_digest or canonical_root_identity(self._source) != self.binding.canonical_root_identity or directory_identity(os.fstat(self._profile_fd)) != self.binding.profile_root_identity or directory_identity(os.fstat(self._data_fd)) != self.binding.data_root_identity or file_identity(os.fstat(self._ledger_fd)) != self.binding.ledger_identity or file_identity(os.fstat(self._lock_fd)) != self.binding.lock_identity or _binding_digest(self.binding) != self.binding.binding_digest ): raise WeeklyOperationsConflict("weekly reminder ledger binding drift") verify_named(self._profile_fd, _DATA, self._data_fd, directory=True) verify_named(self._data_fd, _LEDGER, self._ledger_fd, directory=False) verify_named(self._data_fd, _LOCK, self._lock_fd, directory=False) def _root(self) -> Path: return Path(f"/proc/self/fd/{self._profile_fd}") @contextmanager def _operation(self) -> Generator[Path]: self.verify() with weekly_schedule_descriptor_root(self._profile_fd): with weekly_schedule_authority_fence(self.verify): yield self._root() self.verify() def receipts(self) -> tuple[ScheduledDeliveryReceipt, ...]: with self._operation() as root: return schedule_delivery_ledger(root) def reserve(self, value: LedgerReservation) -> ScheduledDeliveryReceipt: with self._operation() as root: return reserve_missing_checkin_reminder( root, str(value.customer_identity_digest), value.day, value.destination, registry_digest=value.registry_digest, config_digest=value.config_digest, operator_approval=value.authority_digest, canonical_sequence=value.canonical_sequence, canonical_digest=value.canonical_digest, weekly_authority=value.weekly_authority, ) def transition(self, value: LedgerTransition) -> ScheduledDeliveryReceipt: with self._operation() as root: action = str(value.action) match action: case "abandon": return mark_customer_task_abandoned(root, value.receipt, value.reason) case "answered": return abandon_missing_checkin_reminder_for_terminal_morning_response(root, value.receipt) case "sending": if value.authority is None: raise WeeklyOperationsConflict("weekly sending authority absent") return mark_weekly_reminder_sending(root, value.receipt, value.authority) case "known_failure": return mark_customer_task_known_failure(root, value.receipt, value.reason) case "unknown": return mark_customer_task_unknown(root, value.receipt, value.reason) case "delivered": return mark_customer_task_delivered(root, value.receipt, value.provider_receipt, value.message_id) case "audited": return mark_customer_task_sent_audited(root, value.receipt) case _ as unreachable: assert_never(_invalid_action(unreachable)) @asynccontextmanager async def provider_admission(self) -> AsyncGenerator[None]: """Hold one exact flock across final reread, provider call, and terminal fsync.""" identity = self.binding.lock_identity[:2] with _PROCESS_LOCKS_GUARD: process_lock = _PROCESS_LOCKS.setdefault(identity, threading.Lock()) acquired = await run_sync(process_lock.acquire) if not acquired: raise WeeklyOperationsConflict("weekly reminder process lock unavailable") try: await run_sync(fcntl.flock, self._lock_fd, fcntl.LOCK_EX) try: with weekly_schedule_provider_admission(self._lock_fd, self.verify): yield finally: # Cancellation may already be active when the provider exits. # Shield only this bounded mandatory unlock; the original # cancellation resumes as soon as cleanup completes. with CancelScope(shield=True): await run_sync( fcntl.flock, self._lock_fd, fcntl.LOCK_UN ) finally: _ = process_lock.release() def close(self) -> None: for descriptor in (self._lock_fd, self._ledger_fd, self._data_fd, self._profile_fd): os.close(descriptor) def acquire_registered_weekly_reminder_ledger_authority( source: CanonicalCheckinCustomerAuthority, store: WeeklyOperationsStore, ) -> WeeklyReminderLedgerAuthority: """Acquire only from an already-opened registered Todo 4 source and Todo 3 store.""" source.verify() registered = source.registered_binding if source.customer_identity_digest != store.customer_identity_digest: raise WeeklyOperationsConflict("weekly reminder registered authorities disagree") store_authority_digest = store.authority_binding_digest if registered.authority_binding_digest != store_authority_digest: raise WeeklyOperationsConflict( "weekly reminder registered sidecar authority disagrees" ) opened: list[int] = [] try: customer_fd = source.duplicate_registered_root_descriptor(); opened.append(customer_fd) customers_fd = os.open("..", _DIR_FLAGS, dir_fd=customer_fd); opened.append(customers_fd) data_fd = os.open("..", _DIR_FLAGS, dir_fd=customers_fd); opened.append(data_fd) profile_fd = os.open("..", _DIR_FLAGS, dir_fd=data_fd); opened.append(profile_fd) root = Path(f"/proc/self/fd/{profile_fd}") with weekly_schedule_descriptor_root(profile_fd): _ = initialize_schedule_delivery_fence(root) try: os.mkdir(_CLAIMS, mode=0o700, dir_fd=data_fd) except FileExistsError: claims = os.stat(_CLAIMS, dir_fd=data_fd, follow_symlinks=False) _ = directory_identity(claims) ledger_fd = _open_regular(data_fd, _LEDGER, create=True); opened.append(ledger_fd) lock_fd = _open_regular(data_fd, _LOCK); opened.append(lock_fd) binding = WeeklyReminderLedgerBinding( _SCHEMA, source.customer_identity_digest, registered.registration_row_digest, registered.binding_digest, canonical_root_identity(source), directory_identity(os.fstat(profile_fd)), directory_identity(os.fstat(data_fd)), store_authority_digest, store.customer_identity_digest, file_identity(os.fstat(ledger_fd)), file_identity(os.fstat(lock_fd)), "", ) binding = replace(binding, binding_digest=_binding_digest(binding)) authority = WeeklyReminderLedgerAuthority( binding, source, store, profile_fd, data_fd, ledger_fd, lock_fd ) os.close(customers_fd); os.close(customer_fd); opened.clear() return authority except (OSError, WeeklyOperationsConflict) as error: for descriptor in reversed(opened): os.close(descriptor) raise WeeklyOperationsConflict("registered weekly reminder ledger acquisition failed") from error def _open_regular(parent: int, name: str, *, create: bool = False) -> int: flags = os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC if create: flags |= os.O_CREAT descriptor = os.open(name, flags, 0o600, dir_fd=parent) if create: os.fchmod(descriptor, 0o600) return descriptor def _binding_digest(value: WeeklyReminderLedgerBinding) -> str: payload = {name: getattr(value, name) for name in value.__dataclass_fields__ if name != "binding_digest"} return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def _invalid_action(value: str) -> Never: raise AssertionError(f"unexpected weekly ledger action: {value}")