from __future__ import annotations
|
|
import ctypes
|
import hashlib
|
import os
|
import secrets
|
from pathlib import Path
|
from typing import Any, Callable, Mapping
|
|
from .constants import PENDING_SCHEMA
|
from .strict_json import canonical_bytes, loads
|
|
|
class DurabilityError(RuntimeError):
|
def __init__(self, code: str, message: str) -> None:
|
super().__init__(message)
|
self.code = code
|
|
|
def fsync_directory(path: Path) -> None:
|
if os.name != "nt":
|
fd = os.open(path, os.O_RDONLY)
|
try:
|
os.fsync(fd)
|
finally:
|
os.close(fd)
|
return
|
create_file = ctypes.windll.kernel32.CreateFileW
|
handle = create_file(str(path), 0x40000000, 0x7, None, 3, 0x02000000, None)
|
if handle == ctypes.c_void_p(-1).value:
|
raise OSError(ctypes.get_last_error(), "CreateFileW directory failed")
|
try:
|
if not ctypes.windll.kernel32.FlushFileBuffers(handle):
|
raise OSError(ctypes.get_last_error(), "FlushFileBuffers directory failed")
|
finally:
|
ctypes.windll.kernel32.CloseHandle(handle)
|
|
|
class PendingStore:
|
def __init__(self, path: Path, boundary_hook: Callable[[str], None] | None = None) -> None:
|
self.path = path
|
self._hook = boundary_hook or (lambda _phase: None)
|
|
def _boundary(self, name: str) -> None:
|
self._hook(name)
|
|
def load(self) -> dict[str, Any] | None:
|
if not self.path.exists():
|
return None
|
try:
|
value = loads(self.path.read_bytes())
|
except Exception as exc:
|
raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending state is unreadable") from exc
|
required = {
|
"schema", "run_id", "request_id", "phase", "action_budget_consumed", "may_have_dispatched",
|
"refresh_count", "retry_count", "permit_id", "permit_payload_sha256", "deadline_at",
|
}
|
if not isinstance(value, dict) or set(value) != required or value["schema"] != PENDING_SCHEMA:
|
raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending state schema differs")
|
return value
|
|
def write(self, value: Mapping[str, Any]) -> None:
|
payload = canonical_bytes(dict(value), newline=True)
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
partial = self.path.parent / f".{self.path.name}.{secrets.token_hex(8)}.partial"
|
self._boundary("before_open")
|
binary = getattr(os, "O_BINARY", 0)
|
fd = os.open(partial, os.O_CREAT | os.O_EXCL | os.O_WRONLY | binary, 0o600)
|
try:
|
self._boundary("after_open")
|
view = memoryview(payload)
|
while view:
|
written = os.write(fd, view)
|
if written <= 0:
|
raise OSError("short pending write")
|
view = view[written:]
|
self._boundary("after_write")
|
os.fsync(fd)
|
self._boundary("after_file_fsync")
|
finally:
|
os.close(fd)
|
os.replace(partial, self.path)
|
self._boundary("after_replace")
|
if self.path.read_bytes() != payload:
|
raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending readback differs")
|
self._boundary("after_readback")
|
fsync_directory(self.path.parent)
|
self._boundary("after_directory_fsync")
|
if self.path.read_bytes() != payload:
|
raise DurabilityError("E_DISPATCH_DURABILITY_AMBIGUOUS", "pending second readback differs")
|
self._boundary("after_second_readback")
|
|
def initialize(self, *, run_id: str, request_id: str, deadline_at: str) -> dict[str, Any]:
|
value = {
|
"schema": PENDING_SCHEMA,
|
"run_id": run_id,
|
"request_id": request_id,
|
"phase": "PREPARED",
|
"action_budget_consumed": False,
|
"may_have_dispatched": False,
|
"refresh_count": 0,
|
"retry_count": 0,
|
"permit_id": None,
|
"permit_payload_sha256": None,
|
"deadline_at": deadline_at,
|
}
|
self.write(value)
|
return value
|
|
def consume(self, value: Mapping[str, Any], *, permit_id: str, permit_payload: bytes) -> dict[str, Any]:
|
if value.get("phase") != "PREPARED" or value.get("action_budget_consumed") is not False:
|
raise DurabilityError("E_DISPATCH_REPLAY", "action budget is not available")
|
updated = dict(value)
|
updated.update(
|
phase="ACTION_BUDGET_CONSUMED",
|
action_budget_consumed=True,
|
may_have_dispatched=True,
|
refresh_count=1,
|
retry_count=0,
|
permit_id=permit_id,
|
permit_payload_sha256=hashlib.sha256(permit_payload).hexdigest(),
|
)
|
self.write(updated)
|
return updated
|
|
def recovery_projection(self) -> dict[str, Any]:
|
try:
|
value = self.load()
|
except DurabilityError:
|
return {"error_code": "E_DISPATCH_DURABILITY_AMBIGUOUS", "refresh_count": 1, "retry_count": 0, "may_have_dispatched": True}
|
if value is None or value["action_budget_consumed"] is False:
|
return {"error_code": "E_PRE_PERMIT_STOP", "refresh_count": 0, "retry_count": 0, "may_have_dispatched": False}
|
return {"error_code": "E_POST_PERMIT_UNCERTAIN", "refresh_count": 1, "retry_count": 0, "may_have_dispatched": True}
|
