diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..1479a5a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,35 @@ +# Changelog + +All notable changes to this project are documented here. The format is based on +[Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres +to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Versions are per +package; this entry covers the `cockroachdb`, `cockroachdb-core`, and +`cockroachdb-sqlalchemy` packages. + +## [Unreleased] + +Follower reads can now target a specific staleness, and a `@follower_read` / +SQLAlchemy session helper that raised `AttributeError` at runtime has been fixed. + +### Added +- `hours_ago`, `minutes_ago` and `seconds_ago` keyword-only parameters on the + `follower_read()` decorator, `FollowerReadContext.scoped()`, and the SQLAlchemy + session `follower_read()` helper. They add up and compose to + `AS OF SYSTEM TIME '-Ns'`; with nothing set the read still uses + `follower_read_timestamp()`. +- `cockroachdb.staleness.build_staleness()` — the single source of truth for + composing + validating a staleness expression. Exported from `cockroachdb`. + +### Fixed +- **SQLAlchemy `session.follower_read()` raised `AttributeError` at runtime.** It + called `FollowerReadContext.set(...)` / `reset(...)` (methods that do not exist) + and passed a `bool` into a `str | None` contextvar. It now uses the real + `FollowerReadContext.scoped()` contextmanager. +- A negative `hours_ago`/`minutes_ago`/`seconds_ago` now raises `ValueError` at + call time instead of producing an invalid clause. + +### Known limitation +- `MetricsCollector.record_follower_read()` is still not wired to any execution + path (no call sites library-wide). Recording follower-read metrics requires + instrumenting the SQLAlchemy event / psycopg execution layer — tracked + separately. diff --git a/README.md b/README.md index 2828388..c721e08 100644 --- a/README.md +++ b/README.md @@ -145,11 +145,13 @@ with CockroachDBSession(engine) as session: ### 4. Follower Reads +CockroachDB's [follower reads](https://www.cockroachlabs.com/docs/stable/follower-reads.html) let any replica serve reads at a slightly stale timestamp (about 4.2s old), avoiding the round-trip to the leaseholder. Mark a read with `@follower_read` and it runs `AS OF SYSTEM TIME` — always a follower read, never a silent fallback to current data: + ```python from cockroachdb import follower_read -@follower_read(staleness="follower_read_timestamp()") +@follower_read() def get_dashboard_data(conn): """Read from the nearest follower replica — no cross-region latency.""" with conn.cursor() as cur: @@ -157,6 +159,58 @@ def get_dashboard_data(conn): return cur.fetchall() ``` +#### Choosing how far back to read + +By default the read runs at `follower_read_timestamp()`. Use the keyword-only `hours_ago`, `minutes_ago` and `seconds_ago` params to read further back — they add up and compose to `AS OF SYSTEM TIME '-Ns'`: + +```python +@follower_read() # follower_read_timestamp() +@follower_read(seconds_ago=20) # '-20s' +@follower_read(minutes_ago=5) # '-300s' +@follower_read(hours_ago=1) # '-3600s' +@follower_read(hours_ago=1, minutes_ago=30) # '-5400s' (1h30m ago) +``` + +When no additive part is set, the `staleness` string is used — pass a built-in like `follower_read_timestamp()` (default) or a raw interval. A time more recent than `follower_read_timestamp()` still reads historical data, but the leaseholder serves it rather than the nearest replica; a time older than the table's GC window fails in CockroachDB. + +#### Context manager and SQLAlchemy + +The same additive params work on the `FollowerReadContext.scoped()` context manager (and the template/SQLAlchemy-session helpers that delegate to it): + +```python +from cockroachdb.core.follower_read import FollowerReadContext + +with FollowerReadContext.scoped(minutes_ago=2, seconds_ago=30): # '-150s' + ... # every read in this block runs AS OF SYSTEM TIME + +# SQLAlchemy session helper — reads go to a follower for the callback's duration +session.follower_read(lambda s: s.query(Product).all(), seconds_ago=20) +``` + +The SQLAlchemy session `follower_read()` helper routes through the same mechanism and is fully functional (an earlier version raised `AttributeError` at runtime; fixed). + +#### Misuse fails instead of reading current data + +Staleness composition lives in one place — `build_staleness()` (exported from `cockroachdb`) — which never silently falls back to a current read. A **negative** `hours_ago`/`minutes_ago`/`seconds_ago` raises `ValueError` at call time (at decoration time for `@follower_read`, before any session opens for the context manager and SQLAlchemy helper): + +```python +from cockroachdb import build_staleness + +build_staleness(minutes_ago=2, seconds_ago=30) # "'-150s'" +build_staleness() # "follower_read_timestamp()" +build_staleness(seconds_ago=-1) # raises ValueError +``` + +#### When to use follower reads + +| Use case | Follower read? | +|----------|---------------| +| Product browsing, search results | Yes — slight staleness is imperceptible | +| Dashboards, aggregate metrics | Yes — a few seconds lag is fine | +| Checkout, payment | No — must see real-time inventory | +| Authentication, permissions | No — must see latest credentials | + + ### 5. Multi-Region with Context Manager ```python diff --git a/packages/cockroachdb-core/src/cockroachdb/core/follower_read.py b/packages/cockroachdb-core/src/cockroachdb/core/follower_read.py index 9fb21a1..76ac9f1 100644 --- a/packages/cockroachdb-core/src/cockroachdb/core/follower_read.py +++ b/packages/cockroachdb-core/src/cockroachdb/core/follower_read.py @@ -6,6 +6,8 @@ from contextlib import contextmanager from typing import TYPE_CHECKING +from cockroachdb.staleness import build_staleness + if TYPE_CHECKING: from collections.abc import Iterator @@ -31,10 +33,31 @@ def current_staleness(cls) -> str | None: @classmethod @contextmanager - def scoped(cls, staleness: str = "follower_read_timestamp()") -> Iterator[str]: - """Context manager that activates follower reads for the block duration.""" - token = _follower_read_var.set(staleness) + def scoped( + cls, + staleness: str = "follower_read_timestamp()", + *, + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, + ) -> Iterator[str]: + """Context manager that activates follower reads for the block duration. + + Staleness is additive: ``hours_ago``/``minutes_ago``/``seconds_ago`` add + up and, when positive, compose to ``'-Ns'`` taking precedence over + *staleness*. With no parts set, *staleness* is used. + + Raises: + ValueError: if any additive part is negative (fail fast). + """ + resolved = build_staleness( + staleness, + hours_ago=hours_ago, + minutes_ago=minutes_ago, + seconds_ago=seconds_ago, + ) + token = _follower_read_var.set(resolved) try: - yield staleness + yield resolved finally: _follower_read_var.reset(token) diff --git a/packages/cockroachdb-core/src/cockroachdb/core/template.py b/packages/cockroachdb-core/src/cockroachdb/core/template.py index 52c3188..574e38b 100644 --- a/packages/cockroachdb-core/src/cockroachdb/core/template.py +++ b/packages/cockroachdb-core/src/cockroachdb/core/template.py @@ -157,10 +157,24 @@ def in_region(self, region: str) -> RegionalTemplate: @contextmanager def follower_read( - self, staleness: str = "follower_read_timestamp()" + self, + staleness: str = "follower_read_timestamp()", + *, + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, ) -> Iterator[CockroachDBTemplate]: - """Context manager that activates follower reads.""" - with FollowerReadContext.scoped(staleness): + """Context manager that activates follower reads. + + Staleness is additive; see :meth:`FollowerReadContext.scoped`. A + negative additive part raises ``ValueError``. + """ + with FollowerReadContext.scoped( + staleness, + hours_ago=hours_ago, + minutes_ago=minutes_ago, + seconds_ago=seconds_ago, + ): yield self @@ -249,8 +263,22 @@ def in_region(self, region: str) -> AsyncRegionalTemplate: @contextmanager def follower_read( - self, staleness: str = "follower_read_timestamp()" + self, + staleness: str = "follower_read_timestamp()", + *, + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, ) -> Iterator[AsyncCockroachDBTemplate]: - """Context manager that activates follower reads.""" - with FollowerReadContext.scoped(staleness): + """Context manager that activates follower reads. + + Staleness is additive; see :meth:`FollowerReadContext.scoped`. A + negative additive part raises ``ValueError``. + """ + with FollowerReadContext.scoped( + staleness, + hours_ago=hours_ago, + minutes_ago=minutes_ago, + seconds_ago=seconds_ago, + ): yield self diff --git a/packages/cockroachdb-sqlalchemy/src/cockroachdb/sqla/session.py b/packages/cockroachdb-sqlalchemy/src/cockroachdb/sqla/session.py index a3e7eac..f1d9a77 100644 --- a/packages/cockroachdb-sqlalchemy/src/cockroachdb/sqla/session.py +++ b/packages/cockroachdb-sqlalchemy/src/cockroachdb/sqla/session.py @@ -165,30 +165,52 @@ def execute_with_retry(self, fn: Callable[[Session], T]) -> T: # Follower reads # ------------------------------------------------------------------ - def follower_read(self, fn: Callable[[Session], T]) -> T: + def follower_read( + self, + fn: Callable[[Session], T], + *, + staleness: str = "follower_read_timestamp()", + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, + ) -> T: """Execute *fn* as a CockroachDB follower read. - Sets the :data:`FollowerReadContext` contextvar so the event - listener injects ``SET TRANSACTION AS OF SYSTEM TIME - follower_read_timestamp()`` before the query. + Activates the :class:`FollowerReadContext` for the block so the + event listener injects ``SET TRANSACTION AS OF SYSTEM TIME + {staleness}`` before the query. + + Staleness is additive: ``hours_ago``/``minutes_ago``/``seconds_ago`` + add up and, when positive, compose to ``'-Ns'`` taking precedence + over *staleness*. With no parts set, *staleness* is used. Args: fn: A callable that receives a ``Session`` and returns a result. Should perform **read-only** operations. + staleness: AS OF SYSTEM TIME expression used when no additive + parts are set. + hours_ago: Whole hours of staleness to add. Must be >= 0. + minutes_ago: Whole minutes of staleness to add. Must be >= 0. + seconds_ago: Whole seconds of staleness to add. Must be >= 0. Returns: The return value of *fn*. + + Raises: + ValueError: if any additive part is negative (fail fast — raised + before any session is opened). """ - token = FollowerReadContext.set(True) - try: + with FollowerReadContext.scoped( + staleness, + hours_ago=hours_ago, + minutes_ago=minutes_ago, + seconds_ago=seconds_ago, + ): session = self._session_factory() try: - result = fn(session) - return result + return fn(session) finally: session.close() - finally: - FollowerReadContext.reset(token) # ------------------------------------------------------------------ # Region-scoped operations diff --git a/packages/cockroachdb/src/cockroachdb/__init__.py b/packages/cockroachdb/src/cockroachdb/__init__.py index bb8eed2..d90b9b3 100644 --- a/packages/cockroachdb/src/cockroachdb/__init__.py +++ b/packages/cockroachdb/src/cockroachdb/__init__.py @@ -26,6 +26,7 @@ CockroachDBRetryExhausted, RegionValidationError, ) +from cockroachdb.staleness import build_staleness from cockroachdb.types import ( DistanceMetric, IsolationLevel, @@ -48,6 +49,7 @@ "TransactionOutcome", # Meta "__version__", + "build_staleness", "changefeed_handler", "crdb_region", # Decorators diff --git a/packages/cockroachdb/src/cockroachdb/decorators.py b/packages/cockroachdb/src/cockroachdb/decorators.py index e2f126d..fd81382 100644 --- a/packages/cockroachdb/src/cockroachdb/decorators.py +++ b/packages/cockroachdb/src/cockroachdb/decorators.py @@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, ParamSpec, TypeVar from cockroachdb.exceptions import CockroachDBRetryExhausted, is_retryable_error +from cockroachdb.staleness import build_staleness from cockroachdb.types import IsolationLevel if TYPE_CHECKING: @@ -103,18 +104,39 @@ def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> T: def follower_read( staleness: str = "follower_read_timestamp()", + *, + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, ) -> Callable[[Callable[P, T]], Callable[P, T]]: """Execute read queries AS OF SYSTEM TIME for low-latency replica reads. - The staleness parameter controls freshness — either a built-in function - like follower_read_timestamp() or an interval like '-5s'. + Always performs a follower read — there is no fallback to a current read. + + Staleness is additive: ``hours_ago``/``minutes_ago``/``seconds_ago`` add up + and, when their total is positive, compose to ``'-Ns'`` which takes + precedence over *staleness*. With no parts set, *staleness* is used (a + built-in like ``follower_read_timestamp()`` or an interval like ``'-5s'``). Args: - staleness: AS OF SYSTEM TIME expression. + staleness: AS OF SYSTEM TIME expression used when no additive parts set. + hours_ago: Whole hours of staleness to add. Must be >= 0. + minutes_ago: Whole minutes of staleness to add. Must be >= 0. + seconds_ago: Whole seconds of staleness to add. Must be >= 0. + + Raises: + ValueError: if any additive part is negative (raised at decoration + time — fail fast rather than silently read current data). """ + resolved = build_staleness( + staleness, + hours_ago=hours_ago, + minutes_ago=minutes_ago, + seconds_ago=seconds_ago, + ) def decorator(fn: Callable[P, T]) -> Callable[P, T]: - fn.__crdb_follower_read__ = staleness # type: ignore[attr-defined] + fn.__crdb_follower_read__ = resolved # type: ignore[attr-defined] return fn return decorator diff --git a/packages/cockroachdb/src/cockroachdb/staleness.py b/packages/cockroachdb/src/cockroachdb/staleness.py new file mode 100644 index 0000000..5ca925b --- /dev/null +++ b/packages/cockroachdb/src/cockroachdb/staleness.py @@ -0,0 +1,52 @@ +"""Shared helper for building ``AS OF SYSTEM TIME`` staleness expressions. + +Lives in the base ``cockroachdb`` package (not ``cockroachdb-core``) because +both the ``@follower_read`` decorator here and ``FollowerReadContext.scoped`` +in core must reuse it, and core depends on base — not the other way around. +""" + +from __future__ import annotations + + +def build_staleness( + staleness: str = "follower_read_timestamp()", + *, + hours_ago: int = 0, + minutes_ago: int = 0, + seconds_ago: int = 0, +) -> str: + """Build an ``AS OF SYSTEM TIME`` expression. + + Additive staleness: ``hours_ago``/``minutes_ago``/``seconds_ago`` add up + and, when their total is positive, compose to ``'-Ns'`` which takes + precedence over *staleness*. When no parts are set, *staleness* is returned + unchanged (default ``follower_read_timestamp()``). + + This is the single source of truth for the fail-fast rule: a negative part + raises immediately rather than silently falling back to a current read. + + Args: + staleness: Fallback AS OF SYSTEM TIME expression used when no + additive parts are set. + hours_ago: Whole hours of staleness to add. Must be >= 0. + minutes_ago: Whole minutes of staleness to add. Must be >= 0. + seconds_ago: Whole seconds of staleness to add. Must be >= 0. + + Returns: + Either ``'-Ns'`` (composed from the additive parts) or *staleness*. + + Raises: + ValueError: if any additive part is negative. + """ + for name, value in ( + ("hours_ago", hours_ago), + ("minutes_ago", minutes_ago), + ("seconds_ago", seconds_ago), + ): + if value < 0: + raise ValueError(f"{name} must be >= 0, got {value}") + + total_seconds = hours_ago * 3600 + minutes_ago * 60 + seconds_ago + if total_seconds > 0: + return f"'-{total_seconds}s'" + return staleness diff --git a/tests/test_follower_read.py b/tests/test_follower_read.py index a5a3be5..821dd1a 100644 --- a/tests/test_follower_read.py +++ b/tests/test_follower_read.py @@ -2,7 +2,11 @@ from __future__ import annotations +import pytest + from cockroachdb.core.follower_read import FollowerReadContext +from cockroachdb.decorators import follower_read +from cockroachdb.staleness import build_staleness class TestFollowerReadContextIsActive: @@ -62,3 +66,85 @@ def test_multiple_sequential_scopes(self) -> None: with FollowerReadContext.scoped("-10s"): assert FollowerReadContext.is_active() is True assert FollowerReadContext.is_active() is False + + +class TestBuildStaleness: + """Tests for the shared build_staleness helper (CONTRACT v2).""" + + def test_default_returns_follower_read_timestamp(self) -> None: + # Nothing set -> current default. + assert build_staleness() == "follower_read_timestamp()" + + def test_custom_staleness_passthrough(self) -> None: + assert build_staleness("-5s") == "-5s" + + def test_additive_seconds(self) -> None: + assert build_staleness(seconds_ago=30) == "'-30s'" + + def test_additive_composition_adds_up(self) -> None: + # 1h + 2m + 3s = 3723s + assert build_staleness(hours_ago=1, minutes_ago=2, seconds_ago=3) == "'-3723s'" + + def test_additive_wins_over_staleness(self) -> None: + # Any positive part takes precedence over the staleness fallback. + assert build_staleness("follower_read_timestamp()", minutes_ago=5) == "'-300s'" + + def test_zero_parts_fall_back_to_staleness(self) -> None: + assert build_staleness("-5s", hours_ago=0, minutes_ago=0, seconds_ago=0) == "-5s" + + @pytest.mark.parametrize( + ("kwargs", "bad"), + [ + ({"hours_ago": -1}, "hours_ago"), + ({"minutes_ago": -1}, "minutes_ago"), + ({"seconds_ago": -1}, "seconds_ago"), + ], + ) + def test_negative_part_raises(self, kwargs: dict[str, int], bad: str) -> None: + with pytest.raises(ValueError, match=bad): + build_staleness(**kwargs) + + +class TestScopedAdditive: + """Additive staleness on FollowerReadContext.scoped (CONTRACT v2).""" + + def test_additive_composition(self) -> None: + with FollowerReadContext.scoped(minutes_ago=2, seconds_ago=30) as s: + assert s == "'-150s'" + assert FollowerReadContext.current_staleness() == "'-150s'" + + def test_default_unchanged(self) -> None: + with FollowerReadContext.scoped() as s: + assert s == "follower_read_timestamp()" + + def test_negative_raises(self) -> None: + with pytest.raises(ValueError, match="seconds_ago"): + with FollowerReadContext.scoped(seconds_ago=-1): + pass + # Fail fast: context must not have been activated. + assert FollowerReadContext.is_active() is False + + +class TestFollowerReadDecoratorAdditive: + """Additive staleness on the @follower_read decorator (CONTRACT v2).""" + + def test_default(self) -> None: + @follower_read() + def query() -> None: + pass + + assert query.__crdb_follower_read__ == "follower_read_timestamp()" + + def test_additive_composition(self) -> None: + @follower_read(hours_ago=1, seconds_ago=5) + def query() -> None: + pass + + assert query.__crdb_follower_read__ == "'-3605s'" + + def test_negative_raises_at_decoration(self) -> None: + with pytest.raises(ValueError, match="minutes_ago"): + + @follower_read(minutes_ago=-5) + def query() -> None: + pass diff --git a/tests/test_sqla_session.py b/tests/test_sqla_session.py index ddcba86..ec5eb79 100644 --- a/tests/test_sqla_session.py +++ b/tests/test_sqla_session.py @@ -10,6 +10,7 @@ from cockroachdb.exceptions import CockroachDBRetryExhausted from cockroachdb.sqla.session import ( CockroachDBSession, + FollowerReadContext, RegionalSession, _calc_delay, _is_serialization_error, @@ -129,45 +130,50 @@ def fn(session): class TestFollowerRead: - """Tests for CockroachDBSession.follower_read.""" + """Tests for CockroachDBSession.follower_read (real FollowerReadContext).""" - @patch("cockroachdb.sqla.session.FollowerReadContext") - def test_sets_and_resets_context(self, mock_frc): - engine = MagicMock() - crdb = CockroachDBSession(engine) - mock_session = MagicMock() - crdb._session_factory = MagicMock(return_value=mock_session) - mock_token = MagicMock() - mock_frc.set.return_value = mock_token - - crdb.follower_read(lambda s: "result") - - mock_frc.set.assert_called_once_with(True) - mock_frc.reset.assert_called_once_with(mock_token) - - @patch("cockroachdb.sqla.session.FollowerReadContext") - def test_returns_fn_result(self, mock_frc): - engine = MagicMock() - crdb = CockroachDBSession(engine) + def _crdb_with_fake_session(self): + # MagicMock engine is fine: sessionmaker(bind=...) binds lazily and we + # override _session_factory so no real connection is ever opened. + crdb = CockroachDBSession(MagicMock()) mock_session = MagicMock() crdb._session_factory = MagicMock(return_value=mock_session) - mock_frc.set.return_value = MagicMock() - - result = crdb.follower_read(lambda s: 42) - assert result == 42 - - @patch("cockroachdb.sqla.session.FollowerReadContext") - def test_closes_session_on_exception(self, mock_frc): - engine = MagicMock() - crdb = CockroachDBSession(engine) - mock_session = MagicMock() - crdb._session_factory = MagicMock(return_value=mock_session) - mock_frc.set.return_value = MagicMock() - + return crdb, mock_session + + def test_default_staleness_activated_and_reset(self): + crdb, _ = self._crdb_with_fake_session() + seen = {} + crdb.follower_read(lambda s: seen.setdefault("v", FollowerReadContext.current_staleness())) + assert seen["v"] == "follower_read_timestamp()" + assert FollowerReadContext.is_active() is False # reset after block + + def test_additive_staleness_composes(self): + crdb, _ = self._crdb_with_fake_session() + seen = {} + crdb.follower_read( + lambda s: seen.setdefault("v", FollowerReadContext.current_staleness()), + minutes_ago=2, + seconds_ago=30, + ) + assert seen["v"] == "'-150s'" + + def test_negative_raises_before_session_opened(self): + crdb, _ = self._crdb_with_fake_session() + with pytest.raises(ValueError, match="minutes_ago"): + crdb.follower_read(lambda s: None, minutes_ago=-1) + crdb._session_factory.assert_not_called() # fail fast + assert FollowerReadContext.is_active() is False + + def test_returns_fn_result(self): + crdb, _ = self._crdb_with_fake_session() + assert crdb.follower_read(lambda s: 42) == 42 + + def test_closes_session_on_exception(self): + crdb, mock_session = self._crdb_with_fake_session() with pytest.raises(RuntimeError): crdb.follower_read(lambda s: (_ for _ in ()).throw(RuntimeError("boom"))) - mock_session.close.assert_called_once() + assert FollowerReadContext.is_active() is False # reset even on error class TestInRegion: