diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6d57c00..4b62552 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -121,9 +121,11 @@ jobs: echo "authority: $AUTHORITY ($AUTHORITY_NA not applicable)" echo "templates: $TEMPLATES ($TEMPLATES_NA not applicable)" test "$AUTHORITY" = "verified 11/11" - test "$AUTHORITY_NA" = "0" + # G13 is N/A on SQLite, the action's default store: SQLite has no clock of its own + # to diverge from (SPEC-v0.7 §8.9). + test "$AUTHORITY_NA" = "1" test "$TEMPLATES" = "verified 6/6" - test "$TEMPLATES_NA" = "5" + test "$TEMPLATES_NA" = "6" test -s verify-badge.json test -s verify-report.json test -s verify-report.xml diff --git a/CHANGELOG.md b/CHANGELOG.md index f5ce25f..c669ba0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,9 +15,28 @@ any change to one appears here. 3.11, 3.12, 3.13 and 3.14, and the package classifiers name all four. The floor is unchanged: `requires-python` stays `>=3.11`, and mypy and ruff still check against 3.11. No library code changed; the one test fix is below. +- **Clock-skew detection** (SPEC-v0.7 §3, item 1). `PostgresStateStore` measures its server's + clock against the application's at open, and again when an expired lease is declared + `AMBIGUOUS` (at most once per `DEFAULT_LEASE`), in one round trip whose half is the + measurement's bound, so latency alone is never reported as skew. It keeps its latest + measurement as the optional, read-only `clock_skew` attribute, a `ctrlrun.state.ClockSkew`; + `Control` reads it at the start of every `execute` and `resume` and after an `AmbiguousEffect`, + and appends one new event type, `CLOCK_SKEW_DETECTED`, for a measurement past + `clock_skew_threshold` (default one second, at most `DEFAULT_LEASE`, and no value switches it + off). **It observes and reports, and changes no decision**: every lease is still evaluated + against the application clock exactly as at 0.6.1, no reservation outcome changes, and a + measurement that fails is logged and changes nothing. Verify gains G13, graded against a + Postgres `--store-url` and `N/A` on SQLite, and the catalogue moves to + `ctrlrun.guarantees/v3`; the store conformance suite gains a `clock` case, `not_applicable` + on SQLite and the in-memory store because neither has a clock of its own. ### Fixed +- **`PostgresStateStore.events()` read a missing `action_id` back as the string `"None"`.** An + event about no action (the three `DELEGATION_*` types, and now `CLOCK_SKEW_DETECTED` reported at + open) named a proposal called "None" on Postgres alone; SQLite and the in-memory store returned + `None`. It now returns `None` on all three. + - **The migration tests' release fixtures could not build a venv on some interpreters.** `venv.create` copies the interpreter by default, and a copied binary from a shared-libpython build (uv's CPython 3.14 on macOS) aborted inside `ensurepip`, so all five release fixtures diff --git a/docs/SPEC-v0.6.md b/docs/SPEC-v0.6.md index c958e55..8abe5da 100644 --- a/docs/SPEC-v0.6.md +++ b/docs/SPEC-v0.6.md @@ -1818,6 +1818,10 @@ A fixture that fails nothing is a failure; a fixture whose named suite passed is exception of `reservation`'s cross-process case for `InMemoryStateStore`, which is `not_applicable` with §2.4's reason. No other N/A is accepted, from either backend. +*Amended by `SPEC-v0.7.md` §8 T214 (its §9.6, item 7):* the `clock` suite's `skew-measured` case +is also `not_applicable` on both, with T214's reason, because neither has a clock of its own to +measure. That is the one further N/A accepted. + #### T142 — The report refuses a degenerate run Every case `not_applicable` → `report.ok` is `False`. `run(backend, only=…)` naming a case that is not in the registry **raises**, rather than silently running everything or nothing. diff --git a/docs/SPEC-v0.7.md b/docs/SPEC-v0.7.md index c608f99..e72bddc 100644 --- a/docs/SPEC-v0.7.md +++ b/docs/SPEC-v0.7.md @@ -2450,6 +2450,109 @@ decided, not afterwards. ### 12.1 Item 1: clock skew +**Item 1 observes and reports.** No lease is evaluated differently, no reservation outcome changes and no +store write changes: the measurement is one `SELECT clock_timestamp()` and one attribute assignment, and T213 +compares every decision and every record afterwards with both `plan_reservation` and a SQLite store driven +through the same steps. + +**Where the `E3` re-measurement lives.** In `_ambiguate`, after the kept `AMBIGUOUS` write commits and before +`_plan` raises the refusal, on that write's own fresh connection, which the commit has just left outside any +transaction. So `_plan`, the reservation transaction and the lost-commit paths are untouched (item 3a changes +those), and a measurement that fails cannot abort the transaction the refusal belongs to. A kept write that +itself fails raises as at 0.6.1 and measures nothing, because it is no longer §3.5's moment. + +**The rate limit counts attempts, on the application clock.** A re-measurement is due unless the last one was +at or before `now` and less than `DEFAULT_LEASE` ago. The attempt counts, not the success, so a failing query is +not retried on every refusal either; a clock that moved backwards makes one due, since that is itself worth a +reading. The application clock and not a monotonic one, because it is the clock the lease that just expired was +judged by, and it is the only one a test can move (T215). + +**A round trip the application clock measured as negative** (an injected or stepped clock) is taken by its size: +`bound = |t1 - t0| / 2`, `midpoint = min(t0, t1) + bound`. The doubt is the same whichever way the reads came. + +**`ClockSkew` checks its fields at construction**: `timedelta` for the three durations, a non-negative `bound`, a +positive `threshold`, an aware `measured_at` and a `trigger` in the closed pair. A third-party store that builds a +malformed one then fails where it built it, rather than inside the `Control` that would report it. `Control` still +treats any exception from reading or rendering the value as §3.6's "read raised", because a subclass can override +`exceeded`. The pair is private (`_CLOCK_SKEW_TRIGGERS`), so no public name is added beyond §9.2's. + +**"Not the measurement it last reported" is equality, not identity.** A store whose property builds a fresh +`ClockSkew` on every read with the same fields is then reported once, not once per action. Two measurements that +differ in any field are two reports, which is what T217's third step asserts. + +**"Once per store per kind" is two kinds**: the value is not a `ClockSkew` (whatever its type), and the read or +its rendering raised. They are keyed on the store object, weakly, so a process that builds a `Control` per request +around one store still logs each kind once; a store that cannot be weakly referenced falls back to the reading +`Control`'s own set. + +**`getattr(store, "clock_skew", None)` treats a property that raises `AttributeError` as absent**, which is +§3.6's literal read and is kept. The conformance case is where that store's author finds out: it asks for the +attribute with `inspect.getattr_static` first, so a present property whose read raises `AttributeError` fails the +case by name rather than earning the `not_applicable` reserved for an absent one. A forwarding wrapper whose +`__getattr__` reaches a real attribute counts as exposing it. + +**The pull is the first statement of `execute` and of `resume`**, before argument checks and before +`take_continuation`, so an at-open report precedes the first `ACTION_PROPOSED` (G13's observable). The pull after +an `AmbiguousEffect` is the first statement of `_secure`'s handler, before reconciliation and before the refusal's +own event, and observe mode's reservation refusal pulls too, because observe mode reserves and so meets `E3`. +`evaluate`, `delegate` and `revoke` do not pull: §3.6 names `execute` and `resume`, and none of the three meets a +lease. + +**`clock_timestamp()` against `now()` is an equivalent mutant as built**, and the mutation table says so rather +than claiming it closed. Both measurements run outside any transaction (the store's connections are autocommit, and +the `E3` one runs after its commit), so `now()` is the single statement's start and agrees with +`clock_timestamp()` to within the statement. `clock_timestamp()` stays, per §3.4, so that a later caller who +measures inside a transaction does not inherit an error the tests cannot see. + +**`data.measured_at` uses the event-data timestamp convention** (`iso_timestamp`, milliseconds, `Z`), as +`lease_expires_at` does. The three numbers are integer microseconds, exact, as §3.6 requires. + +**A defect the event exposed, fixed here.** `PostgresStateStore.events()` read a NULL `action_id` back as the +string `"None"`, so the three `DELEGATION_*` events have named a proposal called "None" on Postgres since that +store shipped. +T217's comparison of what a sink was handed with what `events()` returns found it. The fix is on the read path +only; nothing is written differently. + +**The conformance case.** Suite `clock`, case `skew-measured`. It aligns by a first measurement against the +host's real clock, as G13 does, and grades a store's retained measurement (`exceeded`), because the suite grades +stores and not `Control`. Four broken-store fixtures keep each check live: a look-alike type, a read that raises, a +detector that never fires and one that always fires. + +**G13 needs nothing from the document but the store.** It proposes an action no document names, so the policy +denies it with `unknown_action`; the report under test is taken at the start of `execute`, before any decision, so +a denial reaches it as surely as an allow. On Postgres it is therefore never `N/A`, which is what §8.9's +*Requires* line says. Its catalogue title is *clock divergence is named*, short enough for the report's column. + +**What §3.8 predicted, measured.** Every verify scenario on Postgres now appends one `CLOCK_SKEW_DETECTED` per +scratch store, because verify's clocks are anchored to the document. No scenario counted events, and all three +shipped examples still pass every applicable guarantee under `--store-url postgresql://…`. No existing Postgres +test asserted a complete event sequence against an injected clock, so none needed changing. SQLite runs of verify +report G13 `N/A`, which moves the counts T113 and T116 pin by one. + +**An injection is sized against the bound, never fixed (review of #136).** G13 and the conformance case both +inject a skew and ask whether it was reported, and both first used a fixed margin past the threshold. The bound is +half the round trip to the store, so on a link whose half exceeds that margin a *conforming* store reports nothing +and the fixed margin calls that silence a defect: verify would have graded the link and blamed the kernel. Both now +widen the injection from the bound the shifted store measured, until a store honest within its bound would have to +report it (`threshold + 2 * bound + alignment`, one definition in `state.py` so the two cannot drift apart), and +both retry a bounded number of times. A report on the clock they meant to align, where the aligning measurement's +own doubt could explain it, is met by aligning again rather than by a FAIL. A link that cannot be outrun is +**verify's internal error, exit 3** (`v0.4 §3.8`: a fact about the machine, never a verdict on the kernel) and, in +the suite, a failure whose reason names the link and says it is not a report the store failed to make. The test for +each injects real latency rather than simulating it. Whether the alignment's own doubt excuses a report is decided +by recomputing the rule from the measurement's fields rather than by reading `exceeded`, so a store whose +`exceeded` always answers true is still caught by the control. + +**A report the store cannot store changes nothing (review of #136).** `append_event` can fail, and it sat +unguarded, so a locked database would have raised out of `execute` before the action was decided, and out of the +`AmbiguousEffect` handler in place of the refusal the caller was owed: an observation deciding an outcome, which is +the one thing §3 says it never does. The append is now guarded like the read, logged once per store per kind, and +`_skew_reported` moves only after the store accepted the event, so a report that was lost is made by the next +action that can store it and a sink is handed only an event that was stored. + +**`v0.1 §6.2`'s list is not edited in place.** v0.2 and v0.3 added nine event types without touching it, and +§9.6 item 2 records this one where the others are recorded. `v0.6 §8` T141 is amended in place, as §8 T214 asks. + ### 12.2 Item 2: the transport classifier ### 12.3a Item 3a: attempt numbers never repeat diff --git a/src/ctrlrun/conformance/store/__init__.py b/src/ctrlrun/conformance/store/__init__.py index c0794ac..a7a8f35 100644 --- a/src/ctrlrun/conformance/store/__init__.py +++ b/src/ctrlrun/conformance/store/__init__.py @@ -18,9 +18,10 @@ **Not applicable is not a pass.** §2.4 allows exactly two N/As, each a property of the backend rather than of the harness: storage that cannot be opened from another process, and storage that does not outlive the object holding it. Both describe `InMemoryStateStore`, which says so in its -own docstring, and `falsely-declares-no-url` is what keeps the declaration honest. Any third N/A -is a failure, `report.ok` is `False` for a zero denominator, and there is no flag that folds one -into the count. +own docstring, and `falsely-declares-no-url` is what keeps the declaration honest. SPEC-v0.7 +§8 T214 adds exactly one more, also a property of the backend: a store that exposes no clock +measurement, because it reads only the application's clock. Any other N/A is a failure, +`report.ok` is `False` for a zero denominator, and there is no flag that folds one into the count. Nothing in the kernel imports this package, and `import ctrlrun` does not reach it (T140f). """ diff --git a/src/ctrlrun/conformance/store/fixtures.py b/src/ctrlrun/conformance/store/fixtures.py index 1393b05..5ba9d10 100644 --- a/src/ctrlrun/conformance/store/fixtures.py +++ b/src/ctrlrun/conformance/store/fixtures.py @@ -22,7 +22,7 @@ import tempfile from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, replace -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -30,7 +30,7 @@ from ...effect import DEFAULT_LEASE, EffectRecord, EffectState, Reservation from ...errors import NotExecuted from ...receipt import Event -from ...state import DelegationRecord, StateStore +from ...state import ClockSkew, DelegationRecord, StateStore from .backends import SQLiteBackend, StoreBackend @@ -364,6 +364,68 @@ def put_delegation(self, record: DelegationRecord) -> None: self._inner.put_delegation(record) +# --- clock (SPEC-v0.7 §8 T214) --------------------------------------------------------------- + + +@dataclass(frozen=True) +class _ClockSkewLookAlike: + """Every field `ClockSkew` has, and not `ClockSkew`. `Control` ignores it by design.""" + + skew: timedelta + bound: timedelta + threshold: timedelta + measured_at: datetime + trigger: str + + @property + def exceeded(self) -> bool: + return abs(self.skew) > self.threshold + self.bound + + +class _SkewLookAlike(_Wrapped): + """Exposes its measurement as a look-alike type rather than `ctrlrun.state.ClockSkew`.""" + + @property + def clock_skew(self) -> Any: + return _ClockSkewLookAlike( + timedelta(seconds=30), timedelta(0), timedelta(seconds=1), datetime.now(UTC), "open" + ) + + +class _SkewReadRaises(_Wrapped): + """Exposes the attribute, and reading it raises.""" + + @property + def clock_skew(self) -> ClockSkew | None: + raise RuntimeError("the measurement could not be read") + + +def _pinned(skew: timedelta) -> ClockSkew: + return ClockSkew( + skew=skew, + bound=timedelta(0), + threshold=timedelta(seconds=1), + measured_at=datetime.now(UTC), + trigger="open", + ) + + +class _SkewNeverReported(_Wrapped): + """A detector that never fires: every measurement says the clocks agree.""" + + @property + def clock_skew(self) -> ClockSkew | None: + return _pinned(timedelta(0)) + + +class _SkewAlwaysReported(_Wrapped): + """A detector that always fires: every measurement says the clocks are an hour apart.""" + + @property + def clock_skew(self) -> ClockSkew | None: + return _pinned(timedelta(hours=1)) + + # --- the declarations ------------------------------------------------------------------------ @@ -551,6 +613,30 @@ def raises_not_executed(root: Path) -> StoreBackend: _wrapping("upserts-a-delegation", _UpsertsADelegation), because="upserted on a duplicate id", ), + Fixture( + "skew-look-alike", + {"clock": "skew-measured"}, + _wrapping("skew-look-alike", _SkewLookAlike), + because="not a ctrlrun.state.ClockSkew", + ), + Fixture( + "skew-read-raises", + {"clock": "skew-measured"}, + _wrapping("skew-read-raises", _SkewReadRaises), + because="reading clock_skew raised", + ), + Fixture( + "skew-never-reported", + {"clock": "skew-measured"}, + _wrapping("skew-never-reported", _SkewNeverReported), + because="was not reported", + ), + Fixture( + "skew-always-reported", + {"clock": "skew-measured"}, + _wrapping("skew-always-reported", _SkewAlwaysReported), + because="aligned with the store's was reported", + ), Fixture( "falsely-declares-no-url", {"reservation": "e1-cross-process", "durability": "ambiguous-survives"}, diff --git a/src/ctrlrun/conformance/store/suites.py b/src/ctrlrun/conformance/store/suites.py index 68eb5b7..df5747e 100644 --- a/src/ctrlrun/conformance/store/suites.py +++ b/src/ctrlrun/conformance/store/suites.py @@ -15,6 +15,7 @@ from __future__ import annotations import contextlib +import inspect import json import os import subprocess @@ -39,7 +40,14 @@ ) from ...policy import Decision from ...receipt import Event, EventType, Receipt, ReceiptResult -from ...state import DelegationRecord, StateStore +from ...state import ( + ClockSkew, + DelegationRecord, + StateStore, + _decisive, + _explained_by_alignment, + _wider_margin, +) from ..report import CaseResult, SuiteStatus from .backends import StoreBackend, store_from_url @@ -1737,6 +1745,177 @@ def delegation_grant_json(backend: StoreBackend, processes: int = CONTENDERS) -> return passed("grant-json-round-trip", title) +# --- clock (SPEC-v0.7 §3, §8 T214) ----------------------------------------------------------- + +#: The one reason this case is `not_applicable`, and only where the attribute is **absent**. A +#: sentence true of every backend that reaches it, a third-party store with a clock it does not +#: expose included. +NO_CLOCK_MEASUREMENT = ( + "this backend exposes no clock measurement; SQLite and the in-memory store read only the " + "application's clock and have none to expose" +) + +#: How far past the threshold the case injects skew: §8 T209's five seconds. +SKEW_MARGIN = timedelta(seconds=5) + +_ABSENT = object() + + +def _exposes_clock_skew(store: StateStore) -> bool: + """Is the optional attribute there at all? Asked without calling it, so a property that + raises `AttributeError` is a read that raised and not an absent attribute; and asked through + `getattr` too, so a forwarding wrapper that does reach a real one counts as exposing it.""" + if inspect.getattr_static(store, "clock_skew", _ABSENT) is not _ABSENT: + return True + try: + getattr(store, "clock_skew") # noqa: B009 - not on the StateStore protocol + except AttributeError: + return False + except Exception: + return True + return True + + +def _clock_skew_of(store: StateStore) -> tuple[ClockSkew | None, str | None]: + """The store's measurement, or the reason it is unusable. `Control` would ignore anything + but a `ClockSkew` or `None` silently in production, so the suite is where its author finds + out.""" + try: + value = getattr(store, "clock_skew") # noqa: B009 - not on the StateStore protocol + except Exception as broke: + return None, ( + f"reading clock_skew raised {type(broke).__name__}: {broke}; Control ignores such a " + "store, so its skew would never be reported" + ) + if value is None or isinstance(value, ClockSkew): + return value, None + return None, ( + f"clock_skew is a {type(value).__name__}, not a ctrlrun.state.ClockSkew; Control ignores " + "it, so this store's skew would never be reported" + ) + + +#: How many times the case may align again, or widen an injection, before it says it could not +#: establish divergence on this link. Every loop in this suite is bounded. +SKEW_ATTEMPTS = 3 + + +def _host_clock_shifted(by: timedelta) -> Callable[[], datetime]: + """The host's clock, shifted. A function and not a lambda closing over a loop variable: the + stores here are opened inside loops, and a closure would hand the last shift to all of them. + """ + return lambda: datetime.now(UTC) + by + + +def _unestablished(what: str, measured: ClockSkew | None) -> str: + """The reason for a link the case could not outrun. It names the link, not the store: a + silence inside a conforming store's own bound is not a finding about it.""" + bound = "unknown" if measured is None else str(measured.bound) + return ( + f"could not establish {what} in {SKEW_ATTEMPTS} attempts: the store's measurements " + f"carry a bound of {bound}, half their round trip, and the case could not make its " + "injection decisive against it. This is a property of the link to the store, not a " + "report the store failed to make; grade it over a faster one" + ) + + +@case("skew-measured", "a store with its own clock names divergence from the application's") +def clock_skew_measured(backend: StoreBackend, processes: int = CONTENDERS) -> CaseResult: + """SPEC-v0.7 §3, §8 T214. An injected skew is reported and an aligned clock is not. + + **Aligned, not raw.** The case first measures the store against the host's real clock and + then aligns by the offset it found, so a CI runner whose clock has drifted grades the store + and not the runner. Both halves are asserted, because a detector that always fires passes + the first and one that never fires passes the second. + + **Sized against the bound, never fixed.** A conforming store reports only past + `threshold + bound`, so an injection is graded only once it is decisive against the bound + the shifted store measured and the aligning measurement's own doubt; until then the case + widens it and opens again. A report on the aligned clock that the aligning measurement's + doubt could explain is met by aligning again. A link it cannot outrun in `SKEW_ATTEMPTS` is + reported as that, by name, and never as a store that stayed silent. + + The suite's only seam is `open_with_clock`, and that is enough: the skew is injected on the + application's side, which is the direction an operator's hosts get wrong. + """ + case_id, title = "skew-measured", clock_skew_measured.title + probe = backend.open() + if not _exposes_clock_skew(probe): + return na(case_id, title, NO_CLOCK_MEASUREMENT) + + first: ClockSkew | None = None + aligned: ClockSkew | None = None + for _ in range(SKEW_ATTEMPTS): + first, problem = _clock_skew_of(probe) + if problem is not None: + return failed(case_id, title, problem) + if first is None: + return failed( + case_id, + title, + "clock_skew is None after open: the store retained no measurement, and a " + "detector that never ran cannot be graded", + ) + at = first.skew + aligned, problem = _clock_skew_of(backend.open_with_clock(_host_clock_shifted(-at))) + if problem is not None: + return failed(case_id, title, problem) + if aligned is None: + return failed(case_id, title, "an aligned store retained no measurement at open") + if not aligned.exceeded: + break + if not _explained_by_alignment(aligned, first.bound): + return failed( + case_id, + title, + f"an application clock aligned with the store's was reported as {aligned.skew} " + f"off (bound {aligned.bound}, threshold {aligned.threshold}, alignment within " + f"{first.bound}): a detector that fires on an aligned clock is one nobody keeps", + ) + probe = backend.open() + else: + return failed(case_id, title, _unestablished("an aligned clock", aligned)) + assert first is not None + offset, alignment, threshold = first.skew, first.bound, first.threshold + + for direction, sign in (("ahead of", 1), ("behind", -1)): + margin = SKEW_MARGIN + measured: ClockSkew | None = None + for _ in range(SKEW_ATTEMPTS): + injected = sign * (threshold + margin) + opened = backend.open_with_clock(_host_clock_shifted(injected - offset)) + measured, problem = _clock_skew_of(opened) + if problem is not None: + return failed(case_id, title, problem) + if measured is None: + return failed( + case_id, + title, + f"an application clock {direction} the store's by {abs(injected)} was not " + "reported (measured None: the store retained no measurement at open)", + ) + if _decisive(injected, measured, alignment): + break + margin = max(margin, _wider_margin(measured, alignment, SKEW_MARGIN)) + else: + return failed(case_id, title, _unestablished(f"a clock {direction} it", measured)) + if not measured.exceeded: + return failed( + case_id, + title, + f"an application clock {direction} the store's by {abs(injected)} was not " + f"reported (measured {measured.skew} within {measured.bound})", + ) + if (measured.skew > timedelta(0)) is not (sign > 0): + return failed( + case_id, + title, + f"an application clock {direction} the store's was measured as {measured.skew}; " + "a positive skew means the application is ahead", + ) + return passed(case_id, title) + + # --- clock plumbing ------------------------------------------------------------------------- @@ -1772,6 +1951,7 @@ def _clocked(backend: StoreBackend, clock: Callable[[], datetime]) -> StateStore continuation_extend_lease, ), "delegation": (delegation_insert, delegation_revoke, delegation_grant_json), + "clock": (clock_skew_measured,), } diff --git a/src/ctrlrun/control.py b/src/ctrlrun/control.py index 001ef5b..72007b6 100644 --- a/src/ctrlrun/control.py +++ b/src/ctrlrun/control.py @@ -11,6 +11,8 @@ import inspect import logging import os +import threading +import weakref from collections.abc import Callable, Iterator, Mapping, Sequence from contextlib import contextmanager from contextvars import ContextVar @@ -85,10 +87,22 @@ iso_timestamp, new_receipt_id, ) -from .state import SQLiteStateStore, StateStore +from .state import ClockSkew, SQLiteStateStore, StateStore _LOG = logging.getLogger(__name__) +#: SPEC-v0.7 §3.6. Which kinds of unusable `clock_skew` value each store has already been logged +#: for: once per store per kind, not once per action, which would flood a log exactly as the +#: event's own rate limit exists to avoid. Keyed weakly so a store's entry goes with the store; +#: a store that cannot be weakly referenced falls back to the reading `Control`'s own set. +_SKEW_WARNED: weakref.WeakKeyDictionary[Any, set[str]] = weakref.WeakKeyDictionary() +_SKEW_WARNED_LOCK = threading.Lock() +_SKEW_NOT_A_MEASUREMENT: Final = "not a ClockSkew" +_SKEW_READ_RAISED: Final = "read raised" +_SKEW_APPEND_FAILED: Final = "append failed" +#: The event's numbers are integer microseconds, exact, so a reader sees what the decision used. +_MICROSECOND: Final = timedelta(microseconds=1) + P = ParamSpec("P") R = TypeVar("R") @@ -310,6 +324,10 @@ def __init__( #: SPEC-v0.3 §3.2 — the provider-versus-context warning is emitted at most once per #: Control. A warning that repeats per call is a warning nobody reads. self._warned_about_principal = False + #: SPEC-v0.7 §3.6: the last clock measurement this Control reported, so the same one is + #: appended once however many actions read it. + self._skew_reported: ClockSkew | None = None + self._skew_warned: set[str] = set() @classmethod def from_file( @@ -569,6 +587,7 @@ def execute( is the only authority besides a human that may move a record out of `AMBIGUOUS` (SPEC-v0.2 §2.2). It runs at most once per call. """ + self._report_clock_skew() if effect_key is not None and not effect_key: raise InvalidArgument("effect_key must be a non-empty string or None") self._check_environment(action) @@ -828,6 +847,9 @@ def _observe_secure( try: approval, reservation = self._observe_take(action, approval_id, effect_key, lease) except (DuplicateEffect, AmbiguousEffect) as refused: + if isinstance(refused, AmbiguousEffect): + # SPEC-v0.7 §3.6, as in `_secure`: observe mode reserves, so it meets E3 too. + self._report_clock_skew(action, effect_key) observation.block(_blocked_by(refused)) self._append( EventType.EFFECT_RESERVATION_REFUSED, @@ -938,6 +960,7 @@ def resume(self, continuation: str, executor: Callable[[], Any]) -> Receipt: a resumption that decided outcomes differently would be a second answer to the only question this library exists to answer. """ + self._report_clock_skew() held = self._store.take_continuation(continuation) action = held.action started_at, approval = self._resumed_context(action, held.record.created_at) @@ -1333,6 +1356,10 @@ def _secure( approval, reservation = self._take(action, approval_id, effect_key, lease) break except AmbiguousEffect as refused: + # SPEC-v0.7 §3.6, before anything else: a store with its own clock re-measures + # when an expired lease is declared AMBIGUOUS, and the report belongs beside this + # refusal, naming this attempt. It changes nothing about the refusal. + self._report_clock_skew(action, effect_key) # SPEC-v0.2 §2.3 — the record is `AMBIGUOUS` now, whether it already was or # was just moved there by an expired lease (v0.1 §5.4). Either way it is in # the state reconciliation asks about, which is why the order is this way @@ -1762,6 +1789,98 @@ def _require_authority(self, what: str) -> Authority: ) return self._authority + # --- clock skew (SPEC-v0.7 §3.6) ---------------------------------------------------- + + def _report_clock_skew( + self, action: Action | None = None, effect_key: str | None = None + ) -> None: + """Append `CLOCK_SKEW_DETECTED` for a new, exceeded measurement the store retained. + + **It observes and decides nothing.** The store sits below `Control` and has no sink, and + `StateStore` is frozen, so a store with its own clock retains its latest measurement as + the optional `clock_skew` attribute and this pulls it. A store without the attribute + reports nothing, which is correct: only a store with a second clock has one to report. + + The value is used only if it is a `ctrlrun.state.ClockSkew`. Anything else, a read that + raises, and an append the store refuses are logged once per store per kind and never + raised: an observation must not be able to fail the action it observes, and a refusal it + sits beside must reach the caller as the refusal it was. With no `action` the report is + about the deployment and carries no `action_id`; beside an `AmbiguousEffect` it names the + attempt whose refusal it accompanies. It does not say skew caused that refusal. + + A report is marked as made only once the store has accepted it, so one the store could + not write is tried again by the next action, and sinks are handed only an event that was + stored, with the store's `event_id` (`v0.2 §4.1`). + """ + try: + value = getattr(self._store, "clock_skew", None) + if value is None: + return + if not isinstance(value, ClockSkew): + self._warn_clock_skew( + _SKEW_NOT_A_MEASUREMENT, + f"it is a {type(value).__name__}, not a ctrlrun.state.ClockSkew", + ) + return + if not value.exceeded or value == self._skew_reported: + return + data = { + "skew_us": value.skew // _MICROSECOND, + "bound_us": value.bound // _MICROSECOND, + "threshold_us": value.threshold // _MICROSECOND, + "direction": "ahead" if value.skew > timedelta(0) else "behind", + "trigger": value.trigger, + "measured_at": iso_timestamp(value.measured_at), + } + except Exception as broke: + self._warn_clock_skew( + _SKEW_READ_RAISED, f"reading it raised {type(broke).__name__}: {broke}" + ) + return + try: + stored = self._store.append_event( + Event( + type=EventType.CLOCK_SKEW_DETECTED, + action_id=None if action is None else action.action_id, + ts=self._clock(), + data=data, + effect_key=effect_key, + ) + ) + except Exception as broke: + # `Exception`, and the width is `_spend_unneeded_approval`'s argument: there is + # nothing to protect here. This is an observation; the action's own events, receipt + # and refusal are written by the paths that follow and raise as they always did. + self._warn_clock_skew( + _SKEW_APPEND_FAILED, + f"the store refused to append CLOCK_SKEW_DETECTED ({type(broke).__name__}: " + f"{broke}), so the report is retried by the next action", + ignored=False, + ) + return + self._skew_reported = value + self._fan_out("on_event", stored, str(stored.type)) + + def _warn_clock_skew(self, kind: str, detail: str, *, ignored: bool = True) -> None: + with _SKEW_WARNED_LOCK: + try: + seen = _SKEW_WARNED.setdefault(self._store, set()) + except TypeError: + seen = self._skew_warned + if kind in seen: + return + seen.add(kind) + _LOG.warning( + "%s: %s: %s. Nothing about any action changes (SPEC-v0.7 §3.6)", + type(self._store).__name__, + ( + "its clock_skew attribute was ignored, so this store's clock skew is never reported" + if ignored + else "a clock skew report was not recorded" + ), + detail, + ) + def _append_delegation(self, type_: EventType, data: Mapping[str, Any]) -> None: """Append one of §7's three action-less events and fan it out. diff --git a/src/ctrlrun/postgres.py b/src/ctrlrun/postgres.py index 157c2f3..8d51462 100644 --- a/src/ctrlrun/postgres.py +++ b/src/ctrlrun/postgres.py @@ -36,7 +36,7 @@ import threading from collections.abc import Callable from dataclasses import replace -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from typing import Any, Final from .action import Action @@ -69,6 +69,7 @@ from .migrations import migrate from .receipt import Event, EventType, Receipt from .state import ( + ClockSkew, DelegationRecord, HeldContinuation, _action_from_json, @@ -144,6 +145,59 @@ def _took(branch: str, effect_key: str) -> None: #: produces an error rather than a hung run. DROP_LOCK_TIMEOUT: Final = "10s" +#: SPEC-v0.7 §3.7. How far this host's clock may disagree with the store's, beyond the +#: measurement's own bound, before it is reported. A third of a percent of `DEFAULT_LEASE`: early +#: enough to name drift before it produces its first unexplained `AMBIGUOUS`, and far past what a +#: synchronized clock drifts by. The operator may set it, up to `DEFAULT_LEASE`; nothing turns +#: the measurement off. +DEFAULT_CLOCK_SKEW_THRESHOLD: Final = timedelta(seconds=1) + +#: SPEC-v0.7 §3.5. What caused a measurement (`ClockSkew.trigger`). +_OPENED: Final = "open" +_LEASE_EXPIRED: Final = "lease_expired" + + +def _checked_threshold(value: object) -> timedelta: + """SPEC-v0.7 §3.7: a positive `timedelta` up to `DEFAULT_LEASE`, or `InvalidArgument`. + + No value turns detection off, so there is no value to accept that would: zero, a negative, + `None` and a number are all refused rather than read as "never report". A threshold above + the default lease would stay silent while a default-lease reservation was declared + `AMBIGUOUS` by skew alone, which is the harm the measurement exists to name. + """ + if not isinstance(value, timedelta): + raise InvalidArgument( + f"clock_skew_threshold must be a timedelta, got {type(value).__name__} (SPEC-v0.7 §3.7)" + ) + if value <= timedelta(0) or value > DEFAULT_LEASE: + raise InvalidArgument( + f"clock_skew_threshold must be positive and at most DEFAULT_LEASE " + f"({DEFAULT_LEASE}), got {value}. No value turns the measurement off " + "(SPEC-v0.7 §3.7)" + ) + return value + + +def _measurement( + before: datetime, server: datetime, after: datetime, threshold: timedelta, trigger: str +) -> ClockSkew: + """SPEC-v0.7 §3.4's arithmetic, on one round trip. + + The server read its clock somewhere between `before` and `after`, so the best estimate of + the application's time at that instant is the midpoint, and the true offset lies within + half the round trip of it. A round trip the application clock measured as negative (an + injected or stepped clock) is taken by its size: the doubt is the same either way. + """ + half = abs(after - before) / 2 + midpoint = min(before, after) + half + return ClockSkew( + skew=midpoint - server, + bound=half, + threshold=threshold, + measured_at=midpoint, + trigger=trigger, + ) + def _psycopg() -> Any: """The driver, imported lazily so `import ctrlrun` never reaches it (T153).""" @@ -243,6 +297,7 @@ def __init__( *, clock: Callable[[], datetime] = _utc_now, schema: str = "public", + clock_skew_threshold: timedelta = DEFAULT_CLOCK_SKEW_THRESHOLD, ) -> None: if not url: raise InvalidArgument("a Postgres store needs a connection URL") @@ -251,12 +306,103 @@ def __init__( self._url = url self._schema = schema self._clock = clock + self._clock_skew_threshold = _checked_threshold(clock_skew_threshold) + self._clock_skew: ClockSkew | None = None + self._skew_lock = threading.Lock() + self._remeasured_at: datetime | None = None self._local = threading.local() self._open: set[Any] = set() self._open_lock = threading.Lock() connection = self._connection() self._refuse_without_ddl_rights(connection) migrate(connection, self._clock(), dialect="postgres") + self._measure_clock_skew(connection, _OPENED) + + # --- clock skew (SPEC-v0.7 §3) ------------------------------------------------------ + # + # Everything in this section observes and reports. No lease is evaluated against what it + # measures, no refusal depends on it, and a measurement that fails changes nothing (§3.2, + # §3.5). It exists because a lease written by one host is read by another, and v0.6 put the + # store on a third, so two clocks can disagree and nothing else would name it. + + @property + def clock_skew(self) -> ClockSkew | None: + """The most recent measurement of this store's clock against the application's. + + SPEC-v0.7 §3.6: an **optional store attribute**, not a `StateStore` method. `Control` + reads it at the start of every `execute` and `resume`, and after a reservation is + refused with `AmbiguousEffect`, and appends `CLOCK_SKEW_DETECTED` for a measurement + that is `exceeded` and new. Retained whether or not it exceeded the threshold; `None` + means no measurement has succeeded. Read-only. + """ + return self._clock_skew + + def _read_server_clock(self, connection: Any) -> datetime: + """The server's own clock, read once (§3.4). + + `clock_timestamp()` and not `now()`: `now()` is the transaction's start time, so a + reading taken inside a transaction would be off by however long it had run. + """ + with connection.cursor() as cursor: + cursor.execute("SELECT clock_timestamp()") + row = cursor.fetchone() + reading = None if row is None else row[0] + if not isinstance(reading, datetime) or reading.utcoffset() is None: + raise TypeError(f"clock_timestamp() returned {reading!r}, not an aware datetime") + return reading.astimezone(UTC) + + def _measure_clock_skew(self, connection: Any, trigger: str) -> None: + """Take one measurement and retain it (§3.4, §3.5). Never raises an `Exception`. + + A measurement that fails is logged and leaves the retained one as it was: it never + refuses an open and never alters a refusal, because an observation that could fail + the thing it observes would be a decision. + """ + try: + before = self._clock() + server = self._read_server_clock(connection) + after = self._clock() + measured = _measurement(before, server, after, self._clock_skew_threshold, trigger) + except Exception as broke: + _LOG.warning( + "could not measure this host's clock against the store's (%s): %s: %s. Nothing " + "is refused and no decision changes (SPEC-v0.7 §3.5)", + trigger, + type(broke).__name__, + broke, + extra={"trigger": trigger}, + ) + return + self._clock_skew = measured + if measured.exceeded: + _LOG.warning( + "this host's clock is %s the store's by %s (within %s; threshold %s; " + "measured on %s). Leases are still decided by this host's clock, so an expired " + "lease it declares AMBIGUOUS may be one its holder is still inside " + "(SPEC-v0.7 §3)", + "ahead of" if measured.skew > timedelta(0) else "behind", + abs(measured.skew), + measured.bound, + measured.threshold, + trigger, + extra={"trigger": trigger}, + ) + + def _remeasure_after_expiry(self, connection: Any, now: datetime) -> None: + """§3.5's second measurement: an expired lease was just declared `AMBIGUOUS`. + + That is the moment skew does its harm, so a measurement then puts a stated disagreement + beside a refusal that would otherwise have no cause on the record. At most once per + `DEFAULT_LEASE` per store, by the application clock: one skewed host must not flood a + sink with a report per expired lease. The attempt counts, not the success, so a failing + query is not retried on every refusal either. + """ + with self._skew_lock: + last = self._remeasured_at + if last is not None and last <= now < last + DEFAULT_LEASE: + return + self._remeasured_at = now + self._measure_clock_skew(connection, _LEASE_EXPIRED) @staticmethod def create_schema(url: str, schema: str) -> None: @@ -765,6 +911,10 @@ def _ambiguate(self, record: EffectRecord, now: datetime) -> None: record, ) self._commit(connection) + # SPEC-v0.7 §3.5: after the write is kept and before `_plan` raises the refusal, on + # this connection, which the commit has just left outside any transaction. It + # cannot raise, so the refusal that follows is the one 0.6.1 raised. + self._remeasure_after_expiry(connection, now) finally: with contextlib.suppress(Exception): connection.close() @@ -1475,7 +1625,11 @@ def events(self) -> tuple[Event, ...]: event_id=int(row[0]), ts=datetime.fromisoformat(str(row[1])), type=EventType(row[2]), - action_id=str(row[3]), + # NULL stays `None`. `str(row[3])` read it back as the string "None", so an + # event about no action (the three `DELEGATION_*` types, and SPEC-v0.7's + # at-open `CLOCK_SKEW_DETECTED`) named a proposal called "None" on this backend + # alone. T217 found it by comparing what a sink was handed with `events()`. + action_id=None if row[3] is None else str(row[3]), effect_key=row[4], approval_id=row[5], data=json.loads(str(row[6])), diff --git a/src/ctrlrun/receipt.py b/src/ctrlrun/receipt.py index 312d9f6..e1e8380 100644 --- a/src/ctrlrun/receipt.py +++ b/src/ctrlrun/receipt.py @@ -113,7 +113,8 @@ class ReceiptResult(StrEnum): class EventType(StrEnum): - """The closed set of event types in SPEC-v0.1 §6.2, extended by SPEC-v0.2 §2.5.""" + """The closed set of event types in SPEC-v0.1 §6.2, extended by SPEC-v0.2 §2.5, SPEC-v0.3 + §7 and SPEC-v0.7 §3.6.""" ACTION_PROPOSED = "ACTION_PROPOSED" POLICY_EVALUATED = "POLICY_EVALUATED" @@ -149,6 +150,12 @@ class EventType(StrEnum): DELEGATION_CREATED = "DELEGATION_CREATED" DELEGATION_REVOKED = "DELEGATION_REVOKED" DELEGATION_REJECTED = "DELEGATION_REJECTED" + #: SPEC-v0.7 §3.6: this host's clock and the store's disagree past the threshold, beyond + #: the measurement's own bound. Named for what happened, not for the store that noticed. + #: `action_id` is `None` on the report of a measurement taken at open, like the three + #: `DELEGATION_*` types; the report beside an expired lease names that attempt. It records + #: a fact beside a refusal and decides nothing: no lease is evaluated against it. + CLOCK_SKEW_DETECTED = "CLOCK_SKEW_DETECTED" @dataclass(frozen=True) @@ -160,7 +167,9 @@ class Event: `action_id` is `None` for the three `DELEGATION_*` types (SPEC-v0.3 §7): they are about an authority record, created and revoked outside any action's life, and they name the delegation in `data.delegation_id`. Inventing a synthetic `action_id` would put a value in - a field every reader takes to name a real proposal. + a field every reader takes to name a real proposal. The same holds for a + `CLOCK_SKEW_DETECTED` reporting a measurement taken when the store opened (SPEC-v0.7 §3.6), + which is about the deployment and not about an action. """ type: EventType diff --git a/src/ctrlrun/state.py b/src/ctrlrun/state.py index 9bbc639..a94cde4 100644 --- a/src/ctrlrun/state.py +++ b/src/ctrlrun/state.py @@ -399,6 +399,103 @@ def is_revoked(self) -> bool: return self.revoked_at is not None +#: SPEC-v0.7 §3.5. What caused a measurement: the store opening, or an expired lease being +#: declared `AMBIGUOUS` (`v0.1 §5.3 E3`). A closed set, like every vocabulary a reader parses. +_CLOCK_SKEW_TRIGGERS: Final = frozenset({"open", "lease_expired"}) + + +@dataclass(frozen=True) +class ClockSkew: + """One measurement of a store's clock against the application's (SPEC-v0.7 §3.4, §3.6). + + `skew` is the application's time at the midpoint of one round trip minus the store's reading, + so a positive value means the application clock is **ahead**. `bound` is half that round + trip: the store read its clock somewhere inside it, so the true offset lies within + `skew ± bound`. `measured_at` is the application's midpoint and `trigger` is `"open"` or + `"lease_expired"`. + + **It observes and decides nothing.** No lease is evaluated against it (§3.2): it is what a + store with its own clock retains, as the optional `clock_skew` attribute, for `Control` to + report as `CLOCK_SKEW_DETECTED`. `Control` reports only an instance of this class, so a store + that exposes one constructs it; a look-alike with the same fields is ignored. + + Fields are checked at construction, so a measurement that is malformed fails where it was + made rather than inside the `Control` that would report it. + """ + + skew: timedelta + bound: timedelta + threshold: timedelta + measured_at: datetime + trigger: str + + def __post_init__(self) -> None: + for name in ("skew", "bound", "threshold"): + if not isinstance(getattr(self, name), timedelta): + raise InvalidArgument(f"ClockSkew.{name} must be a timedelta") + if self.bound < timedelta(0): + raise InvalidArgument("ClockSkew.bound is half a round trip and cannot be negative") + if self.threshold <= timedelta(0): + raise InvalidArgument("ClockSkew.threshold must be positive") + if not isinstance(self.measured_at, datetime) or self.measured_at.utcoffset() is None: + raise InvalidArgument("ClockSkew.measured_at must be a timezone-aware datetime") + if self.trigger not in _CLOCK_SKEW_TRIGGERS: + raise InvalidArgument( + f"ClockSkew.trigger must be one of {sorted(_CLOCK_SKEW_TRIGGERS)}, " + f"got {self.trigger!r}" + ) + + @property + def exceeded(self) -> bool: + """Past the threshold by more than the measurement's own uncertainty (§3.4). + + A slow link widens `bound` and raises the bar exactly as far as the doubt it added, so + latency alone can never produce a report. + """ + return abs(self.skew) > self.threshold + self.bound + + +# --- grading a measurement: G13 and the store conformance suite's clock case ------------------ +# +# Both inject a skew and ask whether it was reported. A conforming store reports only past +# `threshold + bound`, so an injection sized without looking at the bound grades the link and +# not the store: a round trip slow enough that half of it exceeds the margin makes a correct +# store stay silent, and a fixed margin then reports that silence as a defect. One definition, +# so verify and the suite cannot come to disagree about when a silence is a finding. + + +def _decisive(injected: timedelta, measured: ClockSkew, alignment: timedelta) -> bool: + """Must a store honest within `measured.bound` report a skew of `injected`? + + The clock was aligned by a first measurement whose own doubt is `alignment`, so the true + skew is `injected` within `alignment`, and the store may read it anywhere within its bound + of that. It reports only past `threshold + bound`. So only an injection past + `threshold + 2 * bound + alignment` leaves a conforming store no room to stay silent. + """ + return abs(injected) > measured.threshold + 2 * measured.bound + alignment + + +def _wider_margin(measured: ClockSkew, alignment: timedelta, base: timedelta) -> timedelta: + """The margin past the threshold that would have been decisive against `measured`.""" + return 2 * measured.bound + alignment + base + + +def _explained_by_alignment(measured: ClockSkew, alignment: timedelta) -> bool: + """Could this report on a clock meant to be aligned be the aligning measurement's error? + + Only where the measurement is past the threshold by more than its own bound, and by no more + than the alignment's doubt beyond that. A report the store's own rule does not allow is a + detector firing when it must not, and one past both bounds contradicts the measurement the + alignment came from: both are findings, and neither is excused here. + + The rule is recomputed from the fields rather than read from `exceeded`, so a store whose + `exceeded` always answers true is caught by the first branch instead of being excused by + this one. + """ + past = abs(measured.skew) - measured.threshold + return measured.bound < past <= measured.bound + alignment + + class StateStore(ApprovalStore, Protocol): """Durable state behind a `Control` (SPEC-v0.1 §5.3): approvals, effects, evidence.""" diff --git a/src/ctrlrun/verify/guarantees.py b/src/ctrlrun/verify/guarantees.py index efc5fd6..7e495d4 100644 --- a/src/ctrlrun/verify/guarantees.py +++ b/src/ctrlrun/verify/guarantees.py @@ -19,7 +19,9 @@ #: SPEC-v0.6 §9.5 — `v2` adds G11 (§6.6). The version moves because the catalogue is a closed #: set a report is read against, and a reader that met an id it did not know would have no way #: to tell a new guarantee from a corrupted line. -CATALOGUE: Final = "ctrlrun.guarantees/v2" +#: SPEC-v0.7 §9.4: `v3` is G1 to G16. It moves once, with G13, and the other four join it as +#: their items land; nothing is released in between. +CATALOGUE: Final = "ctrlrun.guarantees/v3" @dataclass(frozen=True) @@ -48,6 +50,18 @@ class Guarantee: Guarantee("G9", "delegation cannot escalate", ("v0.3 §10 T76", "v0.3 §10 T81", "v0.3 §10 T75")), Guarantee("G10", "unknown exception is ambiguous", ("v0.1 §5.5", "v0.1 §7 T1", "v0.1 §7 T8")), Guarantee("G11", "an altered receipt is detected", ("v0.6 §6.5", "v0.6 §8 T164")), + Guarantee( + "G13", + "clock divergence is named", + ( + "v0.1 §5.3 E3", + "v0.7 §8 T209", + "v0.7 §8 T210", + "v0.7 §8 T211", + "v0.7 §8 T212", + "v0.7 §8 T213", + ), + ), ) #: By id, for `--only` and for the report. Insertion order is catalogue order. @@ -120,6 +134,13 @@ class Guarantee: "an injected clock" ) +#: G13's one N/A (SPEC-v0.7 §8.9). True of every run it appears on: SQLite has no clock of its +#: own, so there is nothing for the application's to diverge from. +STORE_READS_APPLICATION_CLOCK: Final = ( + "the store verify was given reads only the application's clock, so there is no second clock " + "to diverge from; pass --store-url postgresql://… to grade this" +) + #: `--only` (§4.6). NOT_SELECTED: Final = "not selected" @@ -147,6 +168,7 @@ class Guarantee: "NO_GRANT_MATCHES", "PER_CONNECTION_BACKEND", "PROCESSES", + "STORE_READS_APPLICATION_CLOCK", "SYNTHETIC_PREFIX", "Guarantee", ] diff --git a/src/ctrlrun/verify/scenarios.py b/src/ctrlrun/verify/scenarios.py index 6b57963..344aad7 100644 --- a/src/ctrlrun/verify/scenarios.py +++ b/src/ctrlrun/verify/scenarios.py @@ -75,7 +75,15 @@ new_receipt_id, verify_chain, ) -from ..state import InMemoryStateStore, SQLiteStateStore, StateStore +from ..state import ( + ClockSkew, + InMemoryStateStore, + SQLiteStateStore, + StateStore, + _decisive, + _explained_by_alignment, + _wider_margin, +) from . import guarantees as reg from .report import Counterexample, GuaranteeResult, Status from .worker import OUTCOME_COMMITTED @@ -97,6 +105,11 @@ WILDCARD: Final = "*" _ONE_HOUR: Final = timedelta(hours=1) +_ONE_SECOND: Final = timedelta(seconds=1) + +#: G13: how many times it may align again, or widen an injection, before it reports that it +#: could not establish divergence on this link. Every loop verify runs is bounded (§3.6). +_SKEW_ATTEMPTS: Final = 3 _ONE_MICROSECOND: Final = timedelta(microseconds=1) #: Every loop is bounded (§3.6). A child that wedges makes G4 fail red rather than hanging CI. @@ -172,6 +185,21 @@ def advance(self, delta: timedelta) -> None: self.now += delta +class _HostClock: + """This host's real clock, shifted by a fixed offset (G13, SPEC-v0.7 §8.9). + + The one scenario clock that is not anchored to the document, because G13 is about the host's + clock against the store's and an anchored clock would measure the anchor. It moves as a real + clock moves; nothing waits on it. + """ + + def __init__(self, offset: timedelta = timedelta(0)) -> None: + self.offset = offset + + def __call__(self) -> datetime: + return datetime.now(UTC) + self.offset + + class _Recorder: """The one sink a scenario registers (§3.5): in memory, so a counterexample has evidence. @@ -528,7 +556,7 @@ def __init__(self, loaded: _Loaded, scratch: Path, store_url: str | None = None) def _on_postgres(self) -> bool: return self._store_url.startswith(("postgresql://", "postgres://")) - def _store_for(self, gid: str, clock: _Clock) -> tuple[StateStore, str]: + def _store_for(self, gid: str, clock: Callable[[], datetime]) -> tuple[StateStore, str]: """A scratch store for one guarantee, and the address a subprocess can open it by. **Verify never opens, migrates or writes to the operator's store**, and on Postgres that @@ -2125,6 +2153,154 @@ def body(detail: dict[str, Any]) -> None: finally: store.close() + # --- G13: divergence between the store's clock and this host's is named ---------------- + + def g13(self) -> GuaranteeResult: + """SPEC-v0.7 §8.9. Graded against `--store-url postgresql://…`, `N/A` otherwise. + + **Aligned, not the raw clock.** The host running verify is often a CI runner and not the + operator's production host, so the control is a store whose application clock is aligned + with the server's by the offset a first measurement found. A verify that failed because + a runner's clock drifted would be grading the wrong machine. + + **Sized against the bound, never fixed.** A conforming store reports only past + `threshold + bound`, and the bound is half the round trip to the store, so an injection + of a fixed size grades the link on a slow one: a correct store stays silent and a fixed + margin calls that silence a defect. The injection is widened from the bound the shifted + store measured until a conforming store would have to report it, and a report on the + aligned clock that the aligning measurement's own doubt could explain is met by aligning + again rather than by a FAIL. A link verify cannot outrun in `_SKEW_ATTEMPTS` is + **verify's own internal error**, exit 3 (`v0.4 §3.8`): a fact about the machine and the + network, never a verdict on the kernel. + + The action is one no document names, so the policy denies it (`unknown_action`) and the + guarantee needs nothing from the document but the store: the report under test is taken + at the start of `execute`, before any decision, and a denial reaches it as surely as an + allow. Every store here is a scratch schema verify made and drops (SPEC-v0.6 §4.1). + + Both halves are asserted, `v0.4 §1.3`'s rule: a detector that always fires fails the + control, one that never fires fails the observable, and one that never runs fails the + control, because the measurement must be present. + """ + if not self._on_postgres: + return self.na("G13", reg.STORE_READS_APPLICATION_CLOCK) + recorder = _Recorder() + opened: list[StateStore] = [] + + def store_at(label: str, offset: timedelta) -> tuple[StateStore, Control]: + clock = _HostClock(offset) + store, _ = self._store_for(f"G13-{label}", clock) + opened.append(store) + control = Control( + self.policy, + store, + LocalApprovalProvider(store, clock=clock), + clock=clock, + sinks=[recorder], + environment=self._default_environment, + ) + return store, control + + def proposed(control: Control) -> list[Event]: + start = len(recorder.events) + action = Action( + name=f"{reg.SYNTHETIC_PREFIX}.clock-skew", + arguments={}, + principal=Principal(agent=APPROVER), + environment=self._default_environment, + ) + with suppress(ActionDenied): + control.execute(action, _Executor()) + return recorder.events[start:] + + def reported(events: Sequence[Event]) -> list[Event]: + return [event for event in events if event.type is EventType.CLOCK_SKEW_DETECTED] + + def measurement(store: StateStore, what: str) -> ClockSkew: + value = getattr(store, "clock_skew", None) + _expect_control( + isinstance(value, ClockSkew), + f"{what} measured its clock against this host's when it opened", + f"clock_skew is {value!r}", + ) + assert isinstance(value, ClockSkew) + return value + + # Opened before the body so a counterexample has a store to read, and used as the first + # attempt's probe rather than opening a schema nothing uses. + probe_zero, _ = store_at("probe-0", timedelta(0)) + + def body(detail: dict[str, Any]) -> None: + first: ClockSkew | None = None + for attempt in range(_SKEW_ATTEMPTS): + probe = ( + probe_zero if attempt == 0 else store_at(f"probe-{attempt}", timedelta(0))[0] + ) + first = measurement(probe, "the store") + store, aligned_control = store_at(f"aligned-{attempt}", -first.skew) + events = proposed(aligned_control) + aligned = measurement(store, "a store aligned with the server's clock") + if not reported(events): + break + _expect_control( + _explained_by_alignment(aligned, first.bound), + "a clock aligned with the store's is not reported", + f"CLOCK_SKEW_DETECTED was appended for a measurement of {aligned.skew} " + f"within {aligned.bound} of a {aligned.threshold} threshold, which the " + f"alignment's own doubt of {first.bound} does not explain", + ) + # The aligning measurement's doubt could explain it, so the alignment and not + # the store is what has not been established. Measure again. + else: + raise VerifyInternalError( + f"G13: could not establish an aligned clock against the store's in " + f"{_SKEW_ATTEMPTS} attempts; the round trip to the store carries a bound of " + f"{None if first is None else first.bound}, wider than the threshold it " + "would have to clear. That is a property of this link, not of the kernel " + "(SPEC-v0.4 §3.8)" + ) + assert first is not None + offset, alignment, threshold = -first.skew, first.bound, first.threshold + + for direction, sign in (("ahead", 1), ("behind", -1)): + margin, injected = _ONE_SECOND, timedelta(0) + control, measured = None, None + for attempt in range(_SKEW_ATTEMPTS): + injected = sign * (threshold + margin) + store, control = store_at(f"{direction}-{attempt}", offset + injected) + measured = measurement(store, f"a store {injected} from the server's clock") + if _decisive(injected, measured, alignment): + break + margin = max(margin, _wider_margin(measured, alignment, _ONE_SECOND)) + else: + raise VerifyInternalError( + f"G13: could not establish a clock {direction} of the store's past " + f"its own bound in {_SKEW_ATTEMPTS} attempts; the last bound was " + f"{None if measured is None else measured.bound}. That is a property of " + "this link, not of the kernel (SPEC-v0.4 §3.8)" + ) + assert control is not None + events = proposed(control) + head = events[0] if events else None + data = {} if head is None else dict(head.data) + _expect( + head is not None + and head.type is EventType.CLOCK_SKEW_DETECTED + and data.get("direction") == direction + and abs(int(data.get("skew_us", 0))) > int(data.get("threshold_us", 0)), + f"the first action on a clock {direction} of the store's by {abs(injected)}, " + f"which its own bound cannot explain, is preceded by " + f"CLOCK_SKEW_DETECTED(direction={direction!r})", + f"events were {[str(e.type) for e in events]}", + ) + detail["directions"] = ["ahead", "behind"] + + try: + return self.graded("G13", None, probe_zero, recorder, body) + finally: + for store in opened: + store.close() + @dataclass(frozen=True) class _AlteredChain: diff --git a/tests/test_clock_skew.py b/tests/test_clock_skew.py new file mode 100644 index 0000000..f8e8b46 --- /dev/null +++ b/tests/test_clock_skew.py @@ -0,0 +1,1287 @@ +"""Clock-skew detection. Build-list item 1 of v0.7; SPEC-v0.7 §3, §8.1 T209-T219. + +**Item 1 observes and reports. It changes no decision.** Every lease is decided against the +application clock exactly as at 0.6.1 (§3.2), and T213 is the test that says so with skew present. + +The injection point is the *application* clock, through `clock=`, and Postgres keeps its own. That +is the honest direction: it is the one an operator's hosts get wrong. The tests that need a server +skip without `CTRLRUN_TEST_POSTGRES`; the arithmetic, the threshold's refusals, `Control`'s handling +of the optional attribute, the conformance case on the two shipped backends and G13's `N/A` all +run without one. +""" + +from __future__ import annotations + +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +import pytest + +from ctrlrun import Action, Control, Policy, Principal, SQLiteStateStore +from ctrlrun.conformance.report import SuiteStatus +from ctrlrun.conformance.store import run as run_conformance +from ctrlrun.conformance.store.backends import InMemoryBackend, SQLiteBackend +from ctrlrun.effect import DEFAULT_LEASE, LEASE_EXPIRED, EffectState, plan_reservation +from ctrlrun.errors import AmbiguousEffect, DuplicateEffect, InvalidArgument +from ctrlrun.receipt import EventType +from ctrlrun.state import ClockSkew +from ctrlrun.verify import Status +from ctrlrun.verify import guarantees as reg +from ctrlrun.verify import run as run_verify + +URL = os.environ.get("CTRLRUN_TEST_POSTGRES") + +postgres = pytest.mark.skipif( + not URL, reason="CTRLRUN_TEST_POSTGRES is not set; no server to run against" +) + +ONE_SECOND = timedelta(seconds=1) +ALLOW = "schema: ctrlrun.policy/v1\nactions:\n refund:\n decision: allow\n" + +#: The conformance case's `not_applicable` reason, verbatim from §8 T214. +NO_CLOCK_REASON = ( + "this backend exposes no clock measurement; SQLite and the in-memory store read only the " + "application's clock and have none to expose" +) + +#: G13's `N/A` reason, verbatim from §8.9. +G13_NA_REASON = ( + "the store verify was given reads only the application's clock, so there is no second clock " + "to diverge from; pass --store-url postgresql://… to grade this" +) + + +def an_action(payment_id: str = "p1") -> Action: + return Action("refund", {"payment_id": payment_id, "amount": 1}, Principal("skew-agent")) + + +class Recording: + """A sink that keeps what it was handed, so a test can compare it with the store.""" + + def __init__(self) -> None: + self.events: list = [] + self.receipts: list = [] + + def on_event(self, event) -> None: + self.events.append(event) + + def on_receipt(self, receipt) -> None: + self.receipts.append(receipt) + + +def skew_events(events) -> list: + return [event for event in events if event.type is EventType.CLOCK_SKEW_DETECTED] + + +class Shifted: + """The host's real clock, shifted by a fixed offset. Moves, as a real clock does.""" + + def __init__(self, offset: timedelta = timedelta(0)) -> None: + self.offset = offset + + def __call__(self) -> datetime: + return datetime.now(UTC) + self.offset + + +class Frozen: + """A clock that moves only when a test moves it.""" + + def __init__(self, now: datetime) -> None: + self.now = now + + def __call__(self) -> datetime: + return self.now + + def advance(self, delta: timedelta) -> None: + self.now += delta + + +# --- the measurement's arithmetic (§3.4), no server needed ----------------------------------- + + +def _measured(before, server, after, threshold=ONE_SECOND, trigger="open"): + from ctrlrun.postgres import _measurement + + return _measurement(before, server, after, threshold, trigger) + + +T = datetime(2026, 9, 11, 12, 0, tzinfo=UTC) + + +def test_T209_the_arithmetic_names_an_application_clock_ahead(): + """§3.4: `skew = midpoint - s`, positive when the application is ahead.""" + measured = _measured(T + timedelta(seconds=6), T, T + timedelta(seconds=6, milliseconds=2)) + assert measured.skew == timedelta(seconds=6, milliseconds=1) + assert measured.bound == timedelta(milliseconds=1) + assert measured.threshold == ONE_SECOND + assert measured.measured_at == T + timedelta(seconds=6, milliseconds=1) + assert measured.trigger == "open" + assert measured.exceeded + + +def test_T210_the_arithmetic_names_an_application_clock_behind(): + measured = _measured(T - timedelta(seconds=6), T, T - timedelta(seconds=6)) + assert measured.skew == -timedelta(seconds=6) + assert measured.bound == timedelta(0) + assert measured.exceeded + + +def test_T212_latency_alone_is_never_reported_by_the_arithmetic(): + """§3.4's rule, stated as arithmetic. A symmetric round trip of twice the threshold that + agrees at the midpoint reports nothing, and neither does the worst case: the whole delay on + one side of the server's read, which puts the midpoint as far off as it can get.""" + symmetric = _measured(T - ONE_SECOND, T, T + ONE_SECOND) + assert symmetric.skew == timedelta(0) + assert symmetric.bound == ONE_SECOND + assert not symmetric.exceeded + + for before, after in ((T - 4 * ONE_SECOND, T), (T, T + 4 * ONE_SECOND)): + lopsided = _measured(before, T, after) + assert abs(lopsided.skew) == 2 * ONE_SECOND + assert lopsided.bound == 2 * ONE_SECOND + assert not lopsided.exceeded, (before, after) + + # And the precondition: the same round trip with skew added *is* reported, so the two + # assertions above are not passing because nothing is ever reported. + skewed = _measured(T + 2 * ONE_SECOND, T, T + 4 * ONE_SECOND) + assert skewed.skew == 3 * ONE_SECOND + assert skewed.bound == ONE_SECOND + assert skewed.exceeded + + +def test_exceeded_is_strictly_past_threshold_plus_bound(): + """The boundary, both sides: exactly `threshold + bound` is not reported.""" + at_edge = ClockSkew( + skew=timedelta(seconds=2), + bound=ONE_SECOND, + threshold=ONE_SECOND, + measured_at=T, + trigger="open", + ) + assert not at_edge.exceeded + past = ClockSkew( + skew=-timedelta(seconds=2, microseconds=1), + bound=ONE_SECOND, + threshold=ONE_SECOND, + measured_at=T, + trigger="lease_expired", + ) + assert past.exceeded + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("skew", 7), + ("bound", -timedelta(microseconds=1)), + ("threshold", timedelta(0)), + ("measured_at", datetime(2026, 9, 11, 12, 0)), + ("trigger", "periodic"), + ], + ids=["skew-not-a-timedelta", "negative-bound", "zero-threshold", "naive-time", "trigger"], +) +def test_a_malformed_measurement_fails_where_it_is_built(field, value): + """A third-party store that builds a bad `ClockSkew` finds out at construction, not inside + the `Control` that would report it.""" + fields = { + "skew": ONE_SECOND, + "bound": timedelta(0), + "threshold": ONE_SECOND, + "measured_at": T, + "trigger": "open", + } + ClockSkew(**fields) # the precondition: the unaltered fields are accepted + with pytest.raises(InvalidArgument): + ClockSkew(**{**fields, field: value}) + + +# --- T218: the threshold refuses what it must -------------------------------------------------- + + +@pytest.mark.parametrize( + "value", + [ + timedelta(0), + -ONE_SECOND, + DEFAULT_LEASE + timedelta(microseconds=1), + None, + True, + 1, + 1.0, + "1s", + ], + ids=["zero", "negative", "above-lease", "none", "bool", "int", "float", "string"], +) +def test_T218_the_threshold_refuses_what_it_must(value): + """Refused at construction, before any connection is attempted: the URL below names no + server, so a store that tried to connect first would raise something else.""" + from ctrlrun.postgres import PostgresStateStore + + with pytest.raises(InvalidArgument) as refused: + PostgresStateStore("postgresql://nobody@127.0.0.1:1/none", clock_skew_threshold=value) + assert "clock_skew_threshold" in str(refused.value) + + +def test_T218_the_default_threshold_is_one_second(): + from ctrlrun.postgres import DEFAULT_CLOCK_SKEW_THRESHOLD + + assert DEFAULT_CLOCK_SKEW_THRESHOLD == ONE_SECOND + + +# --- T214: the conformance case on the two shipped backends, and its fixtures ------------------ + + +def _skew_case(report): + suite = next(s for s in report.suites if s.name == "clock") + return next(c for c in suite.cases if c.id == "skew-measured") + + +@pytest.mark.parametrize("backend", [SQLiteBackend, InMemoryBackend], ids=["sqlite", "memory"]) +def test_T214_a_backend_with_no_clock_is_not_applicable_with_its_sentence(backend, tmp_path): + report = run_conformance(backend(tmp_path), only=("skew-measured",)) + case = _skew_case(report) + assert case.status is SuiteStatus.NOT_APPLICABLE, report.to_text() + assert case.reason == NO_CLOCK_REASON + + +@pytest.mark.parametrize( + ("fixture", "fragment"), + [ + ("skew-look-alike", "not a ctrlrun.state.ClockSkew"), + ("skew-read-raises", "reading clock_skew raised"), + ("skew-never-reported", "was not reported"), + ("skew-always-reported", "aligned with the store's was reported"), + ], +) +def test_T214_a_broken_measurement_fails_the_case_by_name(fixture, fragment, tmp_path): + """§8 T214: a present attribute that is not a `ClockSkew`, or whose read raises, fails, and + `not_applicable` is reserved for an absent one. The two grading halves have a fixture each, + so neither is a guard the other subsumes.""" + from ctrlrun.conformance.store.fixtures import FIXTURES + + chosen = next(f for f in FIXTURES if f.name == fixture) + report = run_conformance(chosen.backend(tmp_path), only=("skew-measured",)) + case = _skew_case(report) + assert case.status is SuiteStatus.FAIL, report.to_text() + assert fragment in (case.reason or ""), case.reason + + +# --- T216 and T217 from Control's side, no server needed --------------------------------------- + + +EXCEEDED = ClockSkew( + skew=timedelta(seconds=7, microseconds=5), + bound=timedelta(microseconds=250), + threshold=ONE_SECOND, + measured_at=T, + trigger="open", +) + + +class _ExposesSkew(SQLiteStateStore): + """A real SQLite store that exposes a measurement it was handed, as a store with a clock + would. Everything it decides is SQLite's own.""" + + measurement: object = None + + @property + def clock_skew(self): + return self.measurement + + +@dataclass(frozen=True) +class ClockSkewLookAlike: + skew: timedelta + bound: timedelta + threshold: timedelta + measured_at: datetime + trigger: str + + @property + def exceeded(self) -> bool: + return True + + +class _RaisesOnRead(SQLiteStateStore): + @property + def clock_skew(self): + raise RuntimeError("the measurement is unavailable") + + +def _run_ten(store, clock) -> tuple[list, list, list]: + """Ten actions through one Control: their outcomes, receipts' shapes and event types.""" + recording = Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=clock, sinks=[recording]) + outcomes = [] + for index in range(10): + receipt = control.execute(an_action(f"p{index}"), lambda: "ok", f"refund:p{index}") + outcomes.append((str(receipt.result), receipt.decision_reason, receipt.attempt)) + types = [str(event.type) for event in store.events()] + return outcomes, recording.receipts, types + + +@pytest.mark.parametrize( + "value", + [ + "7s ahead", + timedelta(seconds=7), + ClockSkewLookAlike(timedelta(seconds=7), timedelta(0), ONE_SECOND, T, "open"), + ], + ids=["string", "timedelta", "look-alike"], +) +def test_T216_a_value_that_is_not_a_clock_skew_changes_nothing(value, tmp_path, caplog): + clock = Frozen(T) + baseline_store = SQLiteStateStore(tmp_path / "baseline.db", clock=clock) + expected, expected_receipts, expected_types = _run_ten(baseline_store, clock) + + store = _ExposesSkew(tmp_path / "exposed.db", clock=clock) + store.measurement = value + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + got, receipts, types = _run_ten(store, clock) + + assert got == expected + assert types == expected_types + assert [r.result for r in receipts] == [r.result for r in expected_receipts] + assert EventType.CLOCK_SKEW_DETECTED.value not in types + warned = [r for r in caplog.records if "clock_skew" in r.getMessage()] + assert len(warned) == 1, [r.getMessage() for r in warned] + assert warned[0].levelno == logging.WARNING + assert "ClockSkew" in warned[0].getMessage() + + +def test_T216_a_clock_skew_read_that_raises_changes_nothing(tmp_path, caplog): + clock = Frozen(T) + expected, _, expected_types = _run_ten(SQLiteStateStore(tmp_path / "b.db", clock=clock), clock) + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + got, _, types = _run_ten(_RaisesOnRead(tmp_path / "r.db", clock=clock), clock) + assert got == expected + assert types == expected_types + warned = [r for r in caplog.records if "clock_skew" in r.getMessage()] + assert len(warned) == 1, [r.getMessage() for r in warned] + assert "RuntimeError" in warned[0].getMessage() + + +def test_T216_each_kind_of_error_is_logged_once_per_store(tmp_path, caplog): + """Two kinds on one store are two lines; the same kind again is none.""" + clock = Frozen(T) + store = _ExposesSkew(tmp_path / "kinds.db", clock=clock) + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + store.measurement = "not a measurement" + for index in range(3): + control.execute(an_action(f"a{index}"), lambda: "ok") + store.measurement = ClockSkewLookAlike( + timedelta(seconds=7), timedelta(0), ONE_SECOND, T, "" + ) + for index in range(3): + control.execute(an_action(f"b{index}"), lambda: "ok") + warned = [r for r in caplog.records if "clock_skew" in r.getMessage()] + assert len(warned) == 1, "a wrong type is one kind of error, whatever the type" + + raising = _RaisesOnRead(tmp_path / "raising.db", clock=clock) + other = Control(Policy.from_yaml(ALLOW), raising, clock=clock) + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + for index in range(3): + other.execute(an_action(f"c{index}"), lambda: "ok") + warned = [r for r in caplog.records if "clock_skew" in r.getMessage()] + assert len(warned) == 2 + + +def test_T217_the_event_reaches_every_sink_with_the_store_assigned_id(tmp_path): + clock = Frozen(T) + store = _ExposesSkew(tmp_path / "state.db", clock=clock) + store.measurement = EXCEEDED + first, second = Recording(), Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=clock, sinks=[first, second]) + + control.execute(an_action("p1"), lambda: "ok", "refund:p1") + + stored = skew_events(store.events()) + assert len(stored) == 1 + event = stored[0] + assert event.event_id is not None + for sink in (first, second): + handed = skew_events(sink.events) + assert handed == [event], "every sink gets the stored event, with the store's id" + assert event.action_id is None and event.effect_key is None + assert store.events()[0] == event, "the at-open report precedes the first action" + assert store.events()[1].type is EventType.ACTION_PROPOSED + assert event.data == { + "skew_us": 7_000_005, + "bound_us": 250, + "threshold_us": 1_000_000, + "direction": "ahead", + "trigger": "open", + "measured_at": "2026-09-11T12:00:00.000Z", + } + + control.execute(an_action("p2"), lambda: "ok", "refund:p2") + assert len(skew_events(store.events())) == 1, "the same measurement is reported once" + + # And a different measurement is reported again: the rule is "not the one it last reported", + # not "once per Control". + store.measurement = ClockSkew( + skew=-timedelta(seconds=3), + bound=timedelta(0), + threshold=ONE_SECOND, + measured_at=T + timedelta(minutes=1), + trigger="open", + ) + control.execute(an_action("p3"), lambda: "ok", "refund:p3") + reported = skew_events(store.events()) + assert len(reported) == 2 + assert reported[1].data["direction"] == "behind" + assert reported[1].data["skew_us"] == -3_000_000 + + +def test_T217_a_measurement_within_its_bound_is_never_appended(tmp_path): + """The precondition for T211's silence: `Control` appends only an `exceeded` measurement.""" + clock = Frozen(T) + store = _ExposesSkew(tmp_path / "state.db", clock=clock) + store.measurement = ClockSkew( + skew=timedelta(milliseconds=900), + bound=timedelta(0), + threshold=ONE_SECOND, + measured_at=T, + trigger="open", + ) + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + control.execute(an_action(), lambda: "ok") + assert skew_events(store.events()) == [] + + +def test_T217_a_store_without_the_attribute_reports_nothing(tmp_path, caplog): + clock = Frozen(T) + store = SQLiteStateStore(tmp_path / "state.db", clock=clock) + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + Control(Policy.from_yaml(ALLOW), store, clock=clock).execute(an_action(), lambda: "ok") + assert skew_events(store.events()) == [] + assert not [r for r in caplog.records if "clock_skew" in r.getMessage()] + + +def test_T217_resume_reports_at_its_start(tmp_path): + """§3.6: at the start of every `execute` **and** `resume`.""" + from ctrlrun import Suspended + + clock = Frozen(T) + store = _ExposesSkew(tmp_path / "state.db", clock=clock) + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + + def suspend(): + raise Suspended("skew-continuation") + + with pytest.raises(Suspended): + control.execute(an_action(), suspend, "refund:suspended") + assert skew_events(store.events()) == [] + store.measurement = EXCEEDED + control.resume("skew-continuation", lambda: "ok") + types = [event.type for event in store.events()] + assert EventType.CLOCK_SKEW_DETECTED in types + assert types.index(EventType.CLOCK_SKEW_DETECTED) < types.index(EventType.EXECUTION_RESUMED) + + +class _MeasuresOnRefusal(_ExposesSkew): + """A SQLite store that, like Postgres on `E3`'s path, retains a new measurement when a + reservation meets an expired lease. Only `Control`'s wiring is under test here; the + store-side re-measurement is T215's, against a real server.""" + + def reserve_effect(self, effect_key, action_id, lease=DEFAULT_LEASE): + try: + return super().reserve_effect(effect_key, action_id, lease) + except AmbiguousEffect: + self.measurement = ClockSkew( + skew=timedelta(seconds=9), + bound=timedelta(0), + threshold=ONE_SECOND, + measured_at=self._clock(), + trigger="lease_expired", + ) + raise + + +@pytest.mark.parametrize("mode", ["enforce", "observe"]) +def test_T215_the_report_beside_an_expired_lease_names_the_attempt(mode, tmp_path): + """§3.6: `Control` pulls again immediately after a reservation is refused with + `AmbiguousEffect`, in enforce mode and in observe mode, which reserves too.""" + clock = Frozen(T) + store = _MeasuresOnRefusal(tmp_path / "state.db", clock=clock) + document = ( + ALLOW + if mode == "enforce" + else "schema: ctrlrun.policy/v3\nmode: observe\nactions:\n refund:\n decision: allow\n" + ) + control = Control(Policy.from_yaml(document), store, clock=clock) + store.reserve_effect("refund:lapsed", "act_holder", timedelta(seconds=1)) + store.begin_execution("refund:lapsed", "act_holder") + clock.advance(timedelta(seconds=2)) + + attempt = an_action("late") + if mode == "enforce": + with pytest.raises(AmbiguousEffect): + control.execute(attempt, lambda: "ok", "refund:lapsed") + else: + control.execute(attempt, lambda: "ok", "refund:lapsed") + reported = skew_events(store.events()) + assert len(reported) == 1, [e.type for e in store.events()] + assert reported[0].action_id == attempt.action_id + assert reported[0].effect_key == "refund:lapsed" + assert reported[0].data["trigger"] == "lease_expired" + types = [e.type for e in store.events() if e.action_id == attempt.action_id] + assert types.index(EventType.CLOCK_SKEW_DETECTED) < types.index( + EventType.EFFECT_RESERVATION_REFUSED + ), "the report sits beside the refusal, before it" + + +class _HidesBehindAttributeError(SQLiteStateStore): + """Exposes the attribute, and reading it raises `AttributeError`, which `getattr` with a + default would take for an absent attribute.""" + + @property + def clock_skew(self): + raise AttributeError("the measurement was never set up") + + +class _HidingBackend(SQLiteBackend): + name = "hides-behind-attribute-error" + + def open(self): + store = _HidesBehindAttributeError(self._path) + self._open.append(store) + return store + + def open_with_clock(self, clock): + store = _HidesBehindAttributeError(self._path, clock=clock) + self._open.append(store) + return store + + +def test_T214_a_present_attribute_whose_read_raises_attribute_error_is_not_absent(tmp_path): + """`not_applicable` is reserved for an absent attribute. A property that raises + `AttributeError` is present, and `Control` would silently ignore it, so it fails by name.""" + report = run_conformance(_HidingBackend(tmp_path), only=("skew-measured",)) + case = _skew_case(report) + assert case.status is SuiteStatus.FAIL, report.to_text() + assert "reading clock_skew raised AttributeError" in (case.reason or "") + + +# --- T219: G13's N/A, no server needed ------------------------------------------------------- + + +def test_T219_G13_is_not_applicable_on_sqlite_with_its_sentence(tmp_path): + path = tmp_path / "ctrlrun.yaml" + path.write_text(ALLOW, encoding="utf-8") + report = run_verify(path, only=("G13",)) + result = next(r for r in report.guarantees if r.id == "G13") + assert result.status is Status.NOT_APPLICABLE + assert result.reason == G13_NA_REASON == reg.STORE_READS_APPLICATION_CLOCK + + +def test_T219_the_catalogue_is_v3_and_G13_is_in_it(): + assert reg.CATALOGUE == "ctrlrun.guarantees/v3" + assert "G13" in reg.BY_ID + assert "v0.1 §5.3 E3" in reg.BY_ID["G13"].descends_from + + +# ============================================================================================ +# Against a real server +# ============================================================================================ + + +@pytest.fixture +def schema(): + """A schema of this test's own, dropped afterwards. Stores opened in it are closed first.""" + from ctrlrun.postgres import PostgresStateStore + + assert URL + name = f"skew_{uuid.uuid4().hex[:12]}" + PostgresStateStore.create_schema(URL, name) + opened: list = [] + yield name, opened + for store in opened: + store.close() + PostgresStateStore.drop_schema(URL, name) + + +def _open(schema, clock, *, cls=None, **options): + from ctrlrun.postgres import PostgresStateStore + + name, opened = schema + store = (cls or PostgresStateStore)(URL, schema=name, clock=clock, **options) + opened.append(store) + return store + + +def _timed_open(schema, clock, **options): + started = time.monotonic() + store = _open(schema, clock, **options) + return store, timedelta(seconds=time.monotonic() - started) + + +#: The server runs on this host (T211's premise), so its clock and the host's agree to well +#: within this. It absorbs scheduling noise, never the injected offset. +HOST_TOLERANCE = timedelta(milliseconds=500) + + +@postgres +@pytest.mark.parametrize(("direction", "sign"), [("ahead", 1), ("behind", -1)]) +def test_T209_T210_an_injected_skew_is_named_with_its_bound(schema, direction, sign): + from ctrlrun.postgres import DEFAULT_CLOCK_SKEW_THRESHOLD + + injected = sign * (DEFAULT_CLOCK_SKEW_THRESHOLD + timedelta(seconds=5)) + clock = Shifted(injected) + store, round_trip = _timed_open(schema, clock) + + measured = store.clock_skew + assert isinstance(measured, ClockSkew) + assert measured.exceeded + assert (measured.skew > timedelta(0)) is (sign > 0) + assert measured.bound <= round_trip / 2, "the bound is at most half the round trip observed" + assert abs(measured.skew - injected) <= measured.bound + HOST_TOLERANCE + assert measured.trigger == "open" + assert measured.threshold == DEFAULT_CLOCK_SKEW_THRESHOLD + + recording = Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=clock, sinks=[recording]) + control.execute(an_action(), lambda: "ok", "refund:p1") + + events = store.events() + assert events[0].type is EventType.CLOCK_SKEW_DETECTED + assert events[1].type is EventType.ACTION_PROPOSED + data = events[0].data + assert data["direction"] == direction + assert data["trigger"] == "open" + assert events[0].action_id is None + for key in ("skew_us", "bound_us", "threshold_us"): + assert isinstance(data[key], int) and not isinstance(data[key], bool), key + assert data["skew_us"] == measured.skew // timedelta(microseconds=1) + assert data["bound_us"] == measured.bound // timedelta(microseconds=1) + assert data["threshold_us"] == 1_000_000 + assert (data["skew_us"] > 0) is (sign > 0) + assert abs(data["skew_us"]) > data["threshold_us"] + data["bound_us"] + assert skew_events(recording.events) == [events[0]] + + +@postgres +def test_T211_the_real_clock_is_measured_and_silent(schema): + """The positive control. Both halves, or the second proves nothing.""" + store = _open(schema, Shifted()) + measured = store.clock_skew + assert measured is not None, "the detector ran: a measurement is present" + assert not measured.exceeded, measured + recording = Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=Shifted(), sinks=[recording]) + control.execute(an_action(), lambda: "ok", "refund:p1") + assert skew_events(store.events()) == [] + assert skew_events(recording.events) == [] + + +def _latency_store(schema, before: timedelta, after: timedelta): + """A store whose application clock reads `before` off the host's clock until the server has + read its own, and `after` from then on: a round trip of `after - before` with no skew.""" + from ctrlrun.postgres import PostgresStateStore + + clock = Shifted(before) + + class _Latent(PostgresStateStore): + def _read_server_clock(self, connection): + read = super()._read_server_clock(connection) + clock.offset = after + return read + + return _open(schema, clock, cls=_Latent) + + +@postgres +def test_T212_latency_alone_is_never_reported(schema): + from ctrlrun.postgres import DEFAULT_CLOCK_SKEW_THRESHOLD as THRESHOLD + + store = _latency_store(schema, -THRESHOLD, THRESHOLD) + measured = store.clock_skew + assert measured is not None + assert not measured.exceeded, measured + assert THRESHOLD <= measured.bound <= THRESHOLD + HOST_TOLERANCE + + lopsided = _latency_store(schema, -4 * THRESHOLD, timedelta(0)) + assert lopsided.clock_skew is not None + assert not lopsided.clock_skew.exceeded, lopsided.clock_skew + + # The precondition: the same injected round trip with skew on top IS reported, so the + # silence above is not a detector that the injection switched off. + skewed = _latency_store(schema, 2 * THRESHOLD, 4 * THRESHOLD) + assert skewed.clock_skew is not None and skewed.clock_skew.exceeded + + +def _decide_on_both(store, sqlite_store, clock, scenario): + """Drive the same reservations against both stores and return what each decided.""" + decided = [] + for key, action_id, lease in scenario: + before = sqlite_store.get_effect(key) + expected = plan_reservation(before, key, action_id, lease, clock()) + outcomes = [] + for target in (store, sqlite_store): + try: + reservation = target.reserve_effect(key, action_id, lease) + except (DuplicateEffect, AmbiguousEffect) as refused: + outcomes.append((type(refused).__name__, getattr(refused, "state", None))) + else: + outcomes.append(("reserved", reservation.attempt, reservation.lease_expires_at)) + want = ( + ("reserved", expected.reservation.attempt, expected.reservation.lease_expires_at) + if expected.reservation is not None + else (type(expected.refusal).__name__, getattr(expected.refusal, "state", None)) + ) + decided.append((key, want, outcomes[0], outcomes[1])) + return decided + + +@postgres +def test_T213_lease_evaluation_is_byte_for_byte_unchanged(schema, tmp_path): + """With skew present and reported, every lease is decided by the application clock. + + The expected values come from `plan_reservation` with the application clock, and the records + afterwards are compared field by field with a SQLite store driven through the same steps, + which is 0.6.1's behaviour on a backend with no second clock. + """ + clock = Frozen(datetime(2025, 1, 1, 12, 0, tzinfo=UTC)) + store = _open(schema, clock) + sqlite_store = SQLiteStateStore(tmp_path / "oracle.db", clock=clock) + assert store.clock_skew is not None and store.clock_skew.exceeded, "skew is present" + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + control.execute(an_action("warmup"), lambda: "ok", "refund:warmup") + assert skew_events(store.events()), "and reported" + + for target in (store, sqlite_store): + target.reserve_effect("refund:live", "act_live", timedelta(minutes=10)) + target.begin_execution("refund:live", "act_live") + target.reserve_effect("refund:lapsing", "act_lapsing", timedelta(seconds=30)) + target.begin_execution("refund:lapsing", "act_lapsing") + target.reserve_effect("refund:done", "act_done", DEFAULT_LEASE) + target.begin_execution("refund:done", "act_done") + target.commit_effect("refund:done", "act_done", {"ok": True}) + target.reserve_effect("refund:failed", "act_failed", DEFAULT_LEASE) + target.begin_execution("refund:failed", "act_failed") + target.fail_effect("refund:failed", "act_failed", "declined") + clock.advance(timedelta(minutes=1)) + + decided = _decide_on_both( + store, + sqlite_store, + clock, + [ + ("refund:live", "act_2", DEFAULT_LEASE), + ("refund:lapsing", "act_3", DEFAULT_LEASE), + ("refund:lapsing", "act_4", DEFAULT_LEASE), + ("refund:done", "act_5", DEFAULT_LEASE), + ("refund:failed", "act_6", DEFAULT_LEASE), + ("refund:new", "act_7", DEFAULT_LEASE), + ], + ) + for key, want, on_postgres, on_sqlite in decided: + assert on_postgres == want, (key, on_postgres, want) + assert on_sqlite == want, (key, on_sqlite, want) + assert [d[1][0] for d in decided] == [ + "DuplicateEffect", + "AmbiguousEffect", + "AmbiguousEffect", + "DuplicateEffect", + "reserved", + "reserved", + ], "the scenario covers a live lease, an expired one, a record left ambiguous, a commit" + + for key in ("refund:live", "refund:lapsing", "refund:done", "refund:failed", "refund:new"): + mine, oracle = store.get_effect(key), sqlite_store.get_effect(key) + assert mine == oracle, (key, mine, oracle) + lapsed = store.get_effect("refund:lapsing") + assert lapsed is not None and lapsed.state is EffectState.AMBIGUOUS + assert lapsed.error == LEASE_EXPIRED + sqlite_store.close() + + +class _Counted: + """Counts the store's server-clock reads by wrapping the one method that makes them.""" + + def __init__(self, monkeypatch) -> None: + from ctrlrun.postgres import PostgresStateStore + + self.reads = 0 + original = PostgresStateStore._read_server_clock + + def counting(store, connection): + self.reads += 1 + return original(store, connection) + + monkeypatch.setattr(PostgresStateStore, "_read_server_clock", counting) + + +def _lapse(store, clock, key: str) -> None: + store.reserve_effect(key, f"act_{key}", timedelta(seconds=1)) + store.begin_execution(key, f"act_{key}") + clock.advance(timedelta(seconds=2)) + + +@postgres +def test_T215_when_it_measures_and_when_it_does_not(schema, monkeypatch): + counted = _Counted(monkeypatch) + clock = Frozen(datetime(2025, 1, 1, 12, 0, tzinfo=UTC)) + store = _open(schema, clock) + assert counted.reads == 1, "one at open" + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + + # Granted, then a live lease, then a committed record: none measures. + control.execute(an_action("g"), lambda: "ok", "refund:granted") + store.reserve_effect("refund:held", "act_held", timedelta(hours=1)) + with pytest.raises(DuplicateEffect): + control.execute(an_action("h"), lambda: "ok", "refund:held") + with pytest.raises(DuplicateEffect): + control.execute(an_action("g2"), lambda: "ok", "refund:granted") + assert counted.reads == 1 + + # A renewal over FAILED measures nothing either. + store.reserve_effect("refund:renewed", "act_r1", DEFAULT_LEASE) + store.begin_execution("refund:renewed", "act_r1") + store.fail_effect("refund:renewed", "act_r1", "declined") + control.execute(an_action("r"), lambda: "ok", "refund:renewed") + assert counted.reads == 1 + + # An expired lease declared AMBIGUOUS: one read, and the report names this attempt. + _lapse(store, clock, "refund:lapsed") + lapsed_action = an_action("l") + with pytest.raises(AmbiguousEffect): + control.execute(lapsed_action, lambda: "ok", "refund:lapsed") + assert counted.reads == 2 + reported = skew_events(store.events()) + assert [e.data["trigger"] for e in reported] == ["open", "lease_expired"] + assert reported[1].action_id == lapsed_action.action_id + assert reported[1].effect_key == "refund:lapsed" + + # An ambiguous record refuses without measuring. + with pytest.raises(AmbiguousEffect): + control.execute(an_action("l2"), lambda: "ok", "refund:lapsed") + assert counted.reads == 2 + + # A second expired lease within DEFAULT_LEASE of the first re-measures nothing. + _lapse(store, clock, "refund:lapsed-again") + with pytest.raises(AmbiguousEffect): + control.execute(an_action("l3"), lambda: "ok", "refund:lapsed-again") + assert counted.reads == 2 + assert len(skew_events(store.events())) == 2 + + # And the interval is a window, not a one-shot: past it, an expired lease measures again. + clock.advance(DEFAULT_LEASE) + _lapse(store, clock, "refund:lapsed-later") + with pytest.raises(AmbiguousEffect): + control.execute(an_action("l4"), lambda: "ok", "refund:lapsed-later") + assert counted.reads == 3 + + +@postgres +def test_T216_a_failed_measurement_at_open_changes_nothing(schema, monkeypatch, caplog): + from ctrlrun.postgres import PostgresStateStore + + def broken(store, connection): + raise RuntimeError("clock_timestamp() is unavailable") + + monkeypatch.setattr(PostgresStateStore, "_read_server_clock", broken) + with caplog.at_level(logging.WARNING, logger="ctrlrun.postgres"): + store = _open(schema, Shifted(timedelta(hours=1))) + assert store.clock_skew is None + named = [r for r in caplog.records if "clock_timestamp() is unavailable" in r.getMessage()] + assert named and named[0].name == "ctrlrun.postgres" + # The store works: it opened, and it reserves. + assert store.reserve_effect("refund:after", "act_after").attempt == 1 + + +@postgres +def test_T216_a_failed_measurement_on_the_E3_path_changes_nothing( + schema, monkeypatch, caplog, tmp_path +): + from ctrlrun.postgres import PostgresStateStore + + clock = Frozen(datetime(2025, 1, 1, 12, 0, tzinfo=UTC)) + store = _open(schema, clock) + oracle = SQLiteStateStore(tmp_path / "oracle.db", clock=clock) + retained = store.clock_skew + assert retained is not None + + def broken(self, connection): + raise RuntimeError("clock_timestamp() is unavailable") + + monkeypatch.setattr(PostgresStateStore, "_read_server_clock", broken) + for target in (store, oracle): + target.reserve_effect("refund:lapsed", "act_a", timedelta(seconds=1)) + target.begin_execution("refund:lapsed", "act_a") + clock.advance(timedelta(seconds=2)) + + raised = [] + with caplog.at_level(logging.WARNING, logger="ctrlrun.postgres"): + for target in (store, oracle): + with pytest.raises(AmbiguousEffect) as refused: + target.reserve_effect("refund:lapsed", "act_b") + raised.append((type(refused.value), str(refused.value), refused.value.action_id)) + assert raised[0] == raised[1] + assert store.get_effect("refund:lapsed") == oracle.get_effect("refund:lapsed") + assert store.clock_skew is retained, "the retained measurement is left as it was" + assert [r for r in caplog.records if "clock_timestamp() is unavailable" in r.getMessage()] + oracle.close() + + +@postgres +def test_T217_through_control_against_postgres(schema): + clock = Shifted(timedelta(seconds=30)) + store = _open(schema, clock) + recording = Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=clock, sinks=[recording]) + control.execute(an_action("p1"), lambda: "ok", "refund:p1") + control.execute(an_action("p2"), lambda: "ok", "refund:p2") + stored = skew_events(store.events()) + assert len(stored) == 1, "a second execute against the same measurement appends nothing" + assert stored[0].event_id is not None + assert skew_events(recording.events) == stored + + +@postgres +def test_T217_an_event_about_no_action_reads_back_as_none_on_postgres(schema): + """The store half of T217: `events()` holds the event the sink was handed. On Postgres a + NULL `action_id` read back as the string "None" until this item; the same event on SQLite + and in memory reads back as `None`, which is what `Event` documents.""" + from ctrlrun.receipt import Event + + store = _open(schema, Shifted()) + written = store.append_event( + Event(type=EventType.DELEGATION_REVOKED, action_id=None, ts=T, data={"delegation_id": "d"}) + ) + (read,) = store.events() + assert read.action_id is None + assert read == written + + +@postgres +@pytest.mark.parametrize( + "threshold", [timedelta(microseconds=1), DEFAULT_LEASE], ids=["smallest", "largest"] +) +def test_T218_no_accepted_value_stops_the_measurement(schema, threshold): + store = _open(schema, Shifted(), clock_skew_threshold=threshold) + assert store.clock_skew is not None + assert store.clock_skew.threshold == threshold + + +@postgres +def test_T214_postgres_passes_the_skew_case(tmp_path): + from ctrlrun.conformance.store.backends import PostgresBackend + + backend = PostgresBackend(URL) + try: + report = run_conformance(backend, only=("skew-measured",)) + finally: + backend.reset() + case = _skew_case(report) + assert case.status is SuiteStatus.PASS, report.to_text() + + +# --- T219: G13 graded against Postgres, and its control both ways ------------------------------ + + +def _g13(tmp_path): + path = tmp_path / "ctrlrun.yaml" + path.write_text(ALLOW, encoding="utf-8") + report = run_verify(path, only=("G13",), store_url=URL) + return next(r for r in report.guarantees if r.id == "G13") + + +@postgres +def test_T219_G13_passes_against_postgres(tmp_path): + result = _g13(tmp_path) + assert result.status is Status.PASS, (result.reason, result.counterexample) + + +@postgres +def test_T219_a_detector_that_always_fires_fails_G13(tmp_path, monkeypatch): + monkeypatch.setattr(ClockSkew, "exceeded", property(lambda self: True)) + result = _g13(tmp_path) + assert result.status is Status.FAIL + assert result.reason == reg.CONTROL_FAILED + + +@postgres +def test_T219_a_detector_that_never_fires_fails_G13(tmp_path, monkeypatch): + monkeypatch.setattr(ClockSkew, "exceeded", property(lambda self: False)) + result = _g13(tmp_path) + assert result.status is Status.FAIL + assert result.reason != reg.CONTROL_FAILED, "the observable fails, not the control" + + +@postgres +def test_T219_a_detector_that_never_runs_fails_G13s_control(tmp_path, monkeypatch): + from ctrlrun.postgres import PostgresStateStore + + def broken(store, connection): + raise RuntimeError("no clock") + + monkeypatch.setattr(PostgresStateStore, "_read_server_clock", broken) + result = _g13(tmp_path) + assert result.status is Status.FAIL + assert result.reason == reg.CONTROL_FAILED + + +# ============================================================================================ +# Review of #136: an observation that cannot be stored, and links slower than the margin +# ============================================================================================ + + +class _RefusesSkewEvents(_MeasuresOnRefusal): + """A store whose `append_event` fails for `CLOCK_SKEW_DETECTED` alone, as a store whose + database is briefly locked might. Every other event is stored.""" + + refusing = True + + def append_event(self, event): + if self.refusing and event.type is EventType.CLOCK_SKEW_DETECTED: + raise RuntimeError("the events table is locked") + return super().append_event(event) + + +def _drive(store, clock, document: str = ALLOW): + """One allow, one duplicate refusal and one expired lease, through one Control. + + Returns what each call came to, the receipts' shapes and the event types, with the skew + events left out, so a run on a store that could not store its report is compared with one + that had nothing to report.""" + control = Control(Policy.from_yaml(document), store, clock=clock) + outcomes = [] + for key in ("refund:one", "refund:one", "refund:lapsed"): + if key == "refund:lapsed": + store.reserve_effect(key, "act_holder", timedelta(seconds=1)) + store.begin_execution(key, "act_holder") + clock.advance(timedelta(seconds=2)) + action = Action("refund", {"key": key, "n": len(outcomes)}, Principal("skew-agent")) + try: + receipt = control.execute(action, lambda: "ok", key) + except Exception as refused: + outcomes.append( + (type(refused).__name__, re.sub(r"act_[0-9a-f]{32}", "act_*", str(refused))) + ) + else: + outcomes.append((str(receipt.result), receipt.decision_reason, receipt.attempt)) + receipts = [ + (str(r.result), r.decision_reason, r.effect_key, r.attempt, r.error is None) + for r in store.receipts() + ] + types = [ + str(event.type) + for event in store.events() + if event.type is not EventType.CLOCK_SKEW_DETECTED + ] + return outcomes, receipts, types + + +OBSERVE = "schema: ctrlrun.policy/v3\nmode: observe\nactions:\n refund:\n decision: allow\n" + + +@pytest.mark.parametrize("document", [ALLOW, OBSERVE], ids=["enforce", "observe"]) +def test_T216_a_report_the_store_cannot_append_changes_no_outcome(document, tmp_path, caplog): + """Review of #136, finding 3. The report is an observation, so a store that cannot write it + must leave every outcome, receipt and refusal exactly as a run with no skew leaves them: the + allow, the duplicate refusal, and the `AmbiguousEffect` beside which the second report would + have gone.""" + plain_clock = Frozen(T) + plain = SQLiteStateStore(tmp_path / "plain.db", clock=plain_clock) + expected = _drive(plain, plain_clock, document) + assert [o[0] for o in expected[0]] == ( + ["committed", "DuplicateEffect", "AmbiguousEffect"] + if document == ALLOW + else ["observed", "observed", "observed"] + ), "the precondition: the run reaches an allow, a duplicate and an expired lease" + + clock = Frozen(T) + store = _RefusesSkewEvents(tmp_path / "refusing.db", clock=clock) + store.measurement = EXCEEDED + with caplog.at_level(logging.WARNING, logger="ctrlrun"): + got = _drive(store, clock, document) + + assert got == expected + assert skew_events(store.events()) == [] + warned = [r for r in caplog.records if "CLOCK_SKEW_DETECTED" in r.getMessage()] + assert len(warned) == 1, "once per store per kind, not once per action" + assert "the events table is locked" in warned[0].getMessage() + + +def test_T216_a_report_that_could_not_be_appended_is_tried_again(tmp_path): + """`_skew_reported` moves only after the store accepted the event, so a measurement whose + report was lost is reported by the next action that can store it.""" + clock = Frozen(T) + store = _RefusesSkewEvents(tmp_path / "state.db", clock=clock) + store.measurement = EXCEEDED + sink = Recording() + control = Control(Policy.from_yaml(ALLOW), store, clock=clock, sinks=[sink]) + control.execute(an_action("p1"), lambda: "ok", "refund:p1") + assert skew_events(store.events()) == [] and skew_events(sink.events) == [] + + store.refusing = False + control.execute(an_action("p2"), lambda: "ok", "refund:p2") + stored = skew_events(store.events()) + assert len(stored) == 1 and stored[0].data["skew_us"] == 7_000_005 + assert skew_events(sink.events) == stored, "a sink is handed only an event that was stored" + + +def test_T216_resume_survives_a_report_the_store_cannot_append(tmp_path): + from ctrlrun import Suspended + + clock = Frozen(T) + store = _RefusesSkewEvents(tmp_path / "state.db", clock=clock) + control = Control(Policy.from_yaml(ALLOW), store, clock=clock) + + def suspend(): + raise Suspended("refused-report") + + with pytest.raises(Suspended): + control.execute(an_action(), suspend, "refund:suspended") + store.measurement = EXCEEDED + receipt = control.resume("refused-report", lambda: "ok") + assert str(receipt.result) == "committed" + assert skew_events(store.events()) == [] + + +# --- the conformance case on a link slower than its margin (review of #136, finding 2) ------- + + +class _SlowLinkStore(SQLiteStateStore): + """A store with a clock of its own, **honest within its bound**, behind a slow link. + + Its clock is this host's, so the true skew is exactly the injected clock minus now. It + reports that plus an error no larger than the bound it states, and each open is handed its + `(error, bound)` from a script, so a test can give the probe a slow round trip and the + later opens a fast one, or the reverse. Everything it decides is SQLite's own.""" + + def __init__(self, path, *, clock, error: timedelta, bound: timedelta) -> None: + super().__init__(path, clock=clock) + midpoint = datetime.now(UTC) + self._measured = ClockSkew( + skew=(clock() - midpoint) + error, + bound=bound, + threshold=ONE_SECOND, + measured_at=midpoint, + trigger="open", + ) + + @property + def clock_skew(self): + return self._measured + + +class _SlowLinkBackend(SQLiteBackend): + name = "slow-link" + + def __init__(self, root, script) -> None: + super().__init__(root) + self._script = list(script) + self.opens = 0 + + def _next(self): + step = self._script[min(self.opens, len(self._script) - 1)] + self.opens += 1 + return step + + def open(self): + return self.open_with_clock(lambda: datetime.now(UTC)) + + def open_with_clock(self, clock): + error, bound = self._next() + store = _SlowLinkStore(self._path, clock=clock, error=error, bound=bound) + self._open.append(store) + return store + + +def _seconds(error: float, bound: float) -> tuple[timedelta, timedelta]: + return timedelta(seconds=error), timedelta(seconds=bound) + + +def test_T214_a_conforming_store_behind_a_slow_link_passes(tmp_path): + """A round trip whose half is six seconds: a skew of the threshold plus five is inside the + store's own doubt, so a conforming store reports nothing, and a fixed margin would fail it. + The case must widen the injection until it is decisive against the bound it measured.""" + backend = _SlowLinkBackend(tmp_path, [_seconds(0, 6)]) + case = _skew_case(run_conformance(backend, only=("skew-measured",))) + assert case.status is SuiteStatus.PASS, case.reason + + +def test_T214_a_probe_slower_than_the_store_is_measured_again(tmp_path): + """The probe's own doubt is the alignment's. Here it is off by five seconds, inside its six + second bound, and the aligned store, on a fast link, honestly reports that five. That is the + probe's error, not a detector firing on an aligned clock, so the case aligns again.""" + backend = _SlowLinkBackend(tmp_path, [_seconds(5, 6), _seconds(0, 0.001), _seconds(0, 0.001)]) + case = _skew_case(run_conformance(backend, only=("skew-measured",))) + assert case.status is SuiteStatus.PASS, case.reason + + +def test_T214_a_link_it_can_never_outrun_says_so_and_blames_no_store(tmp_path): + """Every open's bound ten times the last: no injection is ever decisive. Bounded, and the + reason names the link rather than claiming the store failed to report.""" + backend = _SlowLinkBackend( + tmp_path, [_seconds(0, 0.001), _seconds(0, 0.001)] + [_seconds(0, 10**k) for k in range(8)] + ) + case = _skew_case(run_conformance(backend, only=("skew-measured",))) + assert case.status is SuiteStatus.FAIL + assert "could not establish" in (case.reason or ""), case.reason + assert "was not reported" not in (case.reason or "") + assert backend.opens <= 12, "every retry loop is bounded" + + +# --- G13 on a slow link (review of #136, finding 1) ------------------------------------------ + + +def _slow_after(monkeypatch, *, first: tuple[float, float], rest: tuple[float, float]): + """Real latency on the server-clock read: `(before, after)` seconds around it for the first + measurement, and `rest` for every later one.""" + from ctrlrun.postgres import PostgresStateStore + + original = PostgresStateStore._read_server_clock + calls = [0] + + def slow(store, connection): + before, after = first if calls[0] == 0 else rest + calls[0] += 1 + time.sleep(before) + read = original(store, connection) + time.sleep(after) + return read + + monkeypatch.setattr(PostgresStateStore, "_read_server_clock", slow) + + +@postgres +def test_T219_G13_on_a_link_slower_than_its_margin_passes(tmp_path, monkeypatch): + """A 2.1 s round trip after the probe. The threshold plus one second is then inside a + conforming store's bound, so it reports nothing; G13 must widen the injection from the bound + the shifted store measured, not report a FAIL the kernel did not earn.""" + _slow_after(monkeypatch, first=(0, 0), rest=(1.05, 1.05)) + result = _g13(tmp_path) + assert result.status is Status.PASS, (result.reason, result.counterexample) + + +@postgres +def test_T219_G13_says_when_it_cannot_establish_divergence(tmp_path, monkeypatch): + """Given one attempt on that link, G13 cannot put a skew past a conforming store's bound. It + says so as verify's internal error, exit 3, and never as a FAIL.""" + from ctrlrun.verify import scenarios + from ctrlrun.verify.scenarios import VerifyInternalError + + monkeypatch.setattr(scenarios, "_SKEW_ATTEMPTS", 1) + _slow_after(monkeypatch, first=(0, 0), rest=(1.05, 1.05)) + with pytest.raises(VerifyInternalError) as raised: + _g13(tmp_path) + assert "could not establish" in str(raised.value) + + +@postgres +def test_T219_G13_aligns_again_when_the_probe_was_slow(tmp_path, monkeypatch): + """A probe whose 2.4 s round trip fell entirely before the server read is off by 1.2 s, + inside its bound. The aligned store, on a fast link, honestly reports that 1.2 s: the + probe's error, not a detector firing on an aligned clock, so G13 aligns again.""" + _slow_after(monkeypatch, first=(2.4, 0), rest=(0, 0)) + result = _g13(tmp_path) + assert result.status is Status.PASS, (result.reason, result.counterexample) diff --git a/tests/test_store_conformance.py b/tests/test_store_conformance.py index 400ccfc..c5cb3cd 100644 --- a/tests/test_store_conformance.py +++ b/tests/test_store_conformance.py @@ -104,14 +104,30 @@ def test_T141_the_shipped_backends_pass(backend, tmp_path): assert report.ok, report.to_text() -def test_T141_sqlite_reports_no_not_applicable(tmp_path): - """SQLite has durable, shareable storage, so nothing about it is inapplicable.""" +#: SPEC-v0.7 §8 T214. The one N/A T141 admits beyond §2.4's two, amended in the same commit as +#: the case that produces it (§9.6 item 7): SQLite reads only the application's clock. +NO_CLOCK = ( + "this backend exposes no clock measurement; SQLite and the in-memory store read only the " + "application's clock and have none to expose" +) + + +def test_T141_sqlite_reports_only_the_clock_not_applicable(tmp_path): + """SQLite has durable, shareable storage, so nothing about its storage is inapplicable. The + one N/A is SPEC-v0.7's skew case, because SQLite has no clock of its own to measure.""" report = run(SQLiteBackend(tmp_path)) - assert report.not_applicable_cases == (), report.to_text() + na = [ + (suite.name, case.id, case.reason) + for suite in report.suites + for case in suite.cases + if case.status is SuiteStatus.NOT_APPLICABLE + ] + assert na == [("clock", "skew-measured", NO_CLOCK)], report.to_text() -def test_T141_in_memory_reports_only_the_two_honest_reasons(tmp_path): - """§2.4's table, by **reason**. A third reason from this backend is a failure, not a property. +def test_T141_in_memory_reports_only_the_honest_reasons(tmp_path): + """§2.4's table, by **reason**, plus SPEC-v0.7 T214's clock case. Any other reason from this + backend is a failure, not a property. Counted by reason rather than by case: the cross-process family grew when contended cases were added for `consume_approval_and_reserve`, `take_continuation` and `grant`/`deny`, and @@ -136,6 +152,7 @@ def test_T141_in_memory_reports_only_the_two_honest_reasons(tmp_path): assert reasons == { "this backend's storage cannot be opened from another process", "this backend's storage does not outlive the object that holds it", + NO_CLOCK, }, reasons for suite in report.suites: for case in suite.cases: diff --git a/tests/test_verify.py b/tests/test_verify.py index 6f21af4..86fff22 100644 --- a/tests/test_verify.py +++ b/tests/test_verify.py @@ -169,12 +169,13 @@ def test_T101_a_policy_with_no_approve_rule_makes_G1_and_G2_not_applicable(tmp_p assert report.applicable == report.passed + report.failed # Ten in the catalogue; G1 and G2 for the missing approve band, G8 and G9 for the # missing authority section. Six applicable, and the count is over those six. + # And G13, which is N/A on every SQLite run: SQLite has no clock of its own (SPEC-v0.7 §8.9). assert report.applicable == 7 - assert report.not_applicable == 4 + assert report.not_applicable == 5 text = report.to_text() assert "8/8" not in text assert f"{report.passed}/{report.applicable} declared guarantees pass." in text - assert "4 not applicable: G1, G2, G8, G9." in text + assert "5 not applicable: G1, G2, G8, G9, G13." in text def test_T101b_zero_applicable_guarantees_is_not_a_pass(tmp_path): @@ -785,8 +786,14 @@ def test_G11_is_applicable_even_where_every_action_is_denied(tmp_path): def test_the_catalogue_is_closed_and_ordered(): - assert reg.CATALOGUE == "ctrlrun.guarantees/v2" - assert [guarantee.id for guarantee in reg.GUARANTEES] == [f"G{n}" for n in range(1, 12)] + """SPEC-v0.7 §9.4: `v3` is G1 to G16, and each id lands with its item. Ordered by number, + so an id that arrives before a lower one still sits where a reader looks for it.""" + assert reg.CATALOGUE == "ctrlrun.guarantees/v3" + ids = [guarantee.id for guarantee in reg.GUARANTEES] + assert ids[:11] == [f"G{n}" for n in range(1, 12)] + assert "G13" in ids + assert ids == sorted(ids, key=lambda gid: int(gid[1:])), ids + assert len(ids) == len(set(ids)) for guarantee in reg.GUARANTEES: assert guarantee.descends_from, f"{guarantee.id} names no acceptance test" @@ -854,15 +861,16 @@ def test_observe_mode_is_refused_before_any_scenario_runs(tmp_path): assert "observe" in str(refused.value) -def test_the_v1_payments_template_reports_five_over_five_with_five_not_applicable(): +def test_the_v1_payments_template_reports_six_over_six_with_six_not_applicable(): """The definition of done, dogfooded rather than described (SPEC-v0.4 §4.1).""" report = run(V1_PAYMENTS) assert report.exit_code == 0 - assert (report.passed, report.applicable, report.not_applicable) == (6, 6, 5) + assert (report.passed, report.applicable, report.not_applicable) == (6, 6, 6) text = report.to_text() assert "6/6 declared guarantees pass." in text - assert "5 not applicable: G3, G4, G5, G8, G9." in text + # G13 is N/A on SQLite, which has no clock of its own (SPEC-v0.7 §8.9). + assert "6 not applicable: G3, G4, G5, G8, G9, G13." in text assert "10/10" not in text diff --git a/tests/test_verify_action.py b/tests/test_verify_action.py index abf97c0..ba8fa80 100644 --- a/tests/test_verify_action.py +++ b/tests/test_verify_action.py @@ -139,7 +139,8 @@ def test_T118_ci_asserts_the_two_shapes_the_specification_names(): assert 'test "$AUTHORITY" = "verified 11/11"' in script assert 'test "$TEMPLATES" = "verified 6/6"' in script - assert 'test "$TEMPLATES_NA" = "5"' in script + assert 'test "$AUTHORITY_NA" = "1"' in script + assert 'test "$TEMPLATES_NA" = "6"' in script @pytest.mark.authority @@ -152,10 +153,11 @@ def test_T118_the_two_configurations_really_do_report_those_shapes(): assert authority.badge is not None assert authority.badge["message"] == "verified 11/11" - assert authority.not_applicable == 0 + # G13 only: SQLite has no clock of its own to diverge from (SPEC-v0.7 §8.9). + assert authority.not_applicable == 1 assert templates.badge is not None assert templates.badge["message"] == "verified 6/6" - assert templates.not_applicable == 5 + assert templates.not_applicable == 6 def test_T118_the_action_uploads_the_report_and_writes_a_job_summary(): @@ -198,7 +200,7 @@ def test_T119_the_denominator_is_applicable_and_never_the_catalogue_size(): assert badge is not None assert badge["message"] == f"verified {report.passed}/{report.applicable}" assert report.applicable == 6 - assert len(reg.GUARANTEES) == 11 + assert report.applicable < len(reg.GUARANTEES) assert "/10" not in badge["message"] @@ -208,7 +210,7 @@ def test_T119_the_colour_is_about_failures_and_has_no_amber_for_not_applicable( from ctrlrun.verify import scenarios passing = run(V1_PAYMENTS) - assert passing.not_applicable == 5 + assert passing.not_applicable == 6 assert passing.badge is not None assert passing.badge["color"] == BADGE_PASS_COLOR diff --git a/tests/test_verify_report.py b/tests/test_verify_report.py index 0be9634..f7727d6 100644 --- a/tests/test_verify_report.py +++ b/tests/test_verify_report.py @@ -106,9 +106,10 @@ def test_T113_the_summary_is_the_last_line_and_names_the_not_applicable_ids(tmp_ assert last == report.summary_line() assert last.startswith(f"{report.passed}/{report.applicable} declared guarantees pass.") - assert "5 not applicable: G3, G4, G5, G8, G9." in last - # The fraction is passes over applicable. A report with five N/As does not say 10/10. - assert "10/10" not in text + # G13 is N/A on every SQLite run (SPEC-v0.7 §8.9): SQLite has no clock of its own. + assert "6 not applicable: G3, G4, G5, G8, G9, G13." in last + # The fraction is passes over applicable. A report with six N/As does not say 12/12. + assert "12/12" not in text def test_T113_a_failing_report_names_the_subject_and_prints_the_counterexample( @@ -131,9 +132,9 @@ def test_T113_a_failing_report_names_the_subject_and_prints_the_counterexample( @pytest.mark.parametrize( ("document", "expected"), [ - (ALL_APPLICABLE, "9/9 declared guarantees pass. 2 not applicable"), - (WITH_NOT_APPLICABLE, "6/6 declared guarantees pass. 5 not applicable"), - (EMPTY, "0/0 declared guarantees pass. 11 not applicable"), + (ALL_APPLICABLE, "9/9 declared guarantees pass. 3 not applicable"), + (WITH_NOT_APPLICABLE, "6/6 declared guarantees pass. 6 not applicable"), + (EMPTY, "0/0 declared guarantees pass. 12 not applicable"), ], ids=["passing", "some-na", "all-na"], ) @@ -183,7 +184,7 @@ def test_T114_the_document_matches_the_schema_field_for_field(tmp_path): assert set(document) == TOP_LEVEL assert document["schema"] == REPORT_SCHEMA == "ctrlrun.verify/v1" - assert document["catalogue"] == reg.CATALOGUE == "ctrlrun.guarantees/v2" + assert document["catalogue"] == reg.CATALOGUE == "ctrlrun.guarantees/v3" assert set(document["policy"]) == {"path", "sha256", "schema", "mode", "actions"} assert document["authority"] is None assert document["store"] == {"backend": "sqlite", "scratch": True} @@ -437,14 +438,14 @@ def broken(self, name, vector, decision, expected_reason): assert "internal error" in result.stderr -def test_T116_a_run_with_five_not_applicable_still_exits_0(tmp_path, monkeypatch): +def test_T116_a_run_with_six_not_applicable_still_exits_0(tmp_path, monkeypatch): """N/A never changes the exit code by itself.""" monkeypatch.chdir(tmp_path) result = _cli(tmp_path, WITH_NOT_APPLICABLE) assert result.exit_code == 0 - assert "5 not applicable" in result.stdout + assert "6 not applicable" in result.stdout def test_T116_json_and_junit_can_be_combined(tmp_path, monkeypatch):