From 2d15a97d0e9126ae0a3ba1d43c349bcf31839ca3 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Tue, 3 Nov 2026 01:50:41 +0000 Subject: [PATCH] feat(reporting-storage): order the account change feed by xid and seq Account feeds now order by (xid, seq) behind xid < pg_snapshot_xmin, as the deployment feed does, replacing the page truncation from #1483. A lower seq that commits late is returned on a later page, never skipped, and no feed depends on the account lock. Account cursors carry xid; a seq-only cursor is mapped to a position that skips nothing and still expires exactly through pruned_through_seq. Consumer holds, change pruning and period retirement compare (xid, seq), prune victims are taken in that order, and the migration adds the horizon xid columns, the consumer check constraint and the account_id, xid, seq index idempotently. Mirrors adcp-client#3289. Closes #1825. Co-Authored-By: Claude Sonnet 5.5 --- docs/reporting-release-notes.md | 17 + docs/reporting-row-storage.md | 63 +- .../ledger/change_feed.py | 338 +++++++--- .../ledger/period_retirement.py | 8 +- .../src/adcp_reporting_storage/ledger/pg.py | 5 +- .../ledger/reporting_changes.sql | 65 +- ...test_reporting_change_feed_consumers_pg.py | 102 ++- .../test_reporting_change_feed_horizon_pg.py | 7 +- .../test_reporting_change_feed_pg.py | 598 ++++++++++++++++-- 9 files changed, 1016 insertions(+), 187 deletions(-) diff --git a/docs/reporting-release-notes.md b/docs/reporting-release-notes.md index 575ad019e3..20ea8c9a0f 100644 --- a/docs/reporting-release-notes.md +++ b/docs/reporting-release-notes.md @@ -18,6 +18,23 @@ restart across the upgrade does not turn a replay into a conflict. This is part the shared layout in adcontextprotocol/adcp#7996 (Stage 2, tracked in [#1457](https://github.com/adcontextprotocol/adcp-client-python/issues/1457)). +## Account change feed orders by `(xid, seq)` + +The account change feed (`changes_after(account_id=...)`) now orders by `(xid, seq)` behind +`xid < pg_snapshot_xmin(pg_current_snapshot())`, as the deployment-wide feed already did, matching the JavaScript SDK +([adcp-client#3289](https://github.com/adcontextprotocol/adcp-client/pull/3289)). It replaces the page truncation at +the first in-flight change from #1483, and no longer depends on the account lock for its order: a lower `seq` that +commits late is returned on a later page, never skipped. Account cursors now carry `xid`. A seq-only account cursor +saved by an earlier release is still accepted and mapped to a position that skips nothing, at the cost of redelivering +a few changes, and it still expires exactly (`pruned_through_seq` is kept). `reporting_changes.sql` is re-applied by +`create_schema` and is idempotent: it adds the account horizon's xid columns, replaces the consumer check constraint, +and creates `adcp_reporting_changes_account_xid`. Create that index with `CREATE INDEX CONCURRENTLY` first on a large +changes table, stop older hosts' change maintenance before the new reader runs, and do not roll back after a consumer +saved a new-format account cursor. Not affected: deployment-wide cursors, the buyer-facing `changes_after` projection and +anyone who never enabled `change_feed`. See +[host change feed and consumers](reporting-row-storage.md#host-change-feed-and-consumers) +([#1825](https://github.com/adcontextprotocol/adcp-client-python/issues/1825)). + ## Feed consumer checkpoints are forward-only `save_feed_consumer_cursor` used to replace the saved position unconditionally. It now ignores a cursor behind the diff --git a/docs/reporting-row-storage.md b/docs/reporting-row-storage.md index 7c8d3ec48e..3942bfb588 100644 --- a/docs/reporting-row-storage.md +++ b/docs/reporting-row-storage.md @@ -386,7 +386,7 @@ missing. The JavaScript SDK has no replay of `python-written.json`; the live tes `PgReportingLedgerStore(pool=pool, change_feed=True)` also appends one row to the shared `adcp_reporting_changes` table (PostgreSQL 13 or later, `xid8`) for every obligation, revision and adjustment, in the transaction that writes -the record and after the account lock. The JavaScript SDK writes the same table. It is separate from +the record. The JavaScript SDK writes the same table. It is separate from `reporting_ledger_changes`, the buyer-facing feed behind `changes_checkpoint` and `changes_after`, until the shared Core layout. @@ -398,18 +398,44 @@ await store.save_feed_consumer_cursor("warehouse-sink", page.cursor) # holds ba await store.changes_after(account_id="acct_a", cursor=page.cursor) ``` -- **Account feeds** order by `seq` and truncate each page before the first change whose transaction is not below - every transaction still in flight (`xid >= pg_snapshot_xmin(pg_current_snapshot())`): a page is the committed prefix - in `seq` order and stops there, and its cursor never moves past a held-back change. `seq` is taken under the account - lock, but ordering does not rely on that lock. Truncation is deliberate: a plain `xid < xmin` *filter* would be - wrong, because `seq` order and `xid` order can differ, and it would hand out a later change while skipping an earlier - one it excluded. `xmin` is cluster-wide, so **any older open transaction anywhere in the database holds account pages - back until it ends, and an empty page can mean "caught up" or "held back"**; poll again rather than concluding - nothing is new. The row scan is still served by the `(account_id, seq)` index. This narrows the window for a second - writer but does not close it for every interleaving: one writer SDK per database remains the contract. -- **The deployment-wide feed** (no `account_id`) orders by `(xid, seq)` and returns only changes whose transaction is - older than every transaction still in flight, so a slow commit is never skipped. A long-running transaction delays - this feed; it never makes it lose a record. +- **Every feed orders by `(xid, seq)`**, account-scoped or deployment-wide (no `account_id`), and returns only changes + whose transaction is older than every transaction still in flight (`xid < pg_snapshot_xmin(pg_current_snapshot())`), + so a slow commit is never skipped. The order does not depend on the account lock, on the order a writer drew `seq` + in, or on which lock key a writer took: a change with a lower `seq` and a higher `xid` that commits late is returned + on a later page, after the cursor. The cost is that `xmin` is cluster-wide, so **any older open transaction anywhere + in the database holds every feed back until it ends, and an empty page can mean "caught up" or "held back"**; poll + again rather than concluding nothing is new, and keep transactions short. The account scan is served by the + `(account_id, xid, seq)` index that `create_schema` creates. One writer SDK per database remains the contract: a + second SDK can write rows the first SDK's lifecycle, retention and idempotency logic does not know about. +- **Account cursors carry `xid`** (`{"v":1,"account":...,"seq":...,"xid":...}`), the same bytes the JavaScript SDK + writes. Store the cursor string opaquely. Cursor `seq` and `xid` are range-checked, so a value the database would + reject raises `ValueError`, never a driver error. +- **Upgrading from account cursors that carry only `seq`.** Account feeds used to order by `seq` and truncate each + page before the first in-flight change. A cursor saved then has no `xid`. `changes_after` accepts it and maps it to + the position `(X, S)`, where `X` is the smaller of the oldest `xid` among the account's changes after `S` and the + reading snapshot's `xmin`. Every change after `S` has an `xid` at or above both (a change that has not committed yet + belongs to a transaction at or above `xmin`), so nothing is skipped. The cost is redelivery: changes the consumer + already read can come back (those with a lower `seq` and a higher `xid`, and, if a long transaction was open at that + moment, every change committed since it began), so treat redelivery as normal, as for any at-least-once consumer. + The returned cursor is in the new format even on an empty page, so a sink saves it with its next batch. A seq-only + cursor raises `CURSOR_EXPIRED` when **any** change with a higher `seq` was pruned: `pruned_through_seq` still records + the highest `seq` pruned, whatever its `xid`, for exactly this check, because a tuple compare alone would miss a + pruned change with a higher `seq` and a lower `xid`. A consumer registered before the upgrade has a NULL `cursor_xid` + and holds retention by `seq` until its next `save_feed_consumer_cursor`; saving a seq-only cursor registers the + mapped position. +- **Rolling out the `(xid, seq)` account feed.** `create_schema` applies `reporting_changes.sql`, which is idempotent: + it adds `pruned_through_xid` and `pruned_through_xid_seq` to `adcp_reporting_change_account_horizon`, replaces the + `adcp_reporting_feed_consumers` check constraint with `adcp_reporting_feed_consumers_cursor_scope` (an account + consumer may now carry an `xid`; a pre-upgrade account row keeps a NULL one), and creates + `adcp_reporting_changes_account_xid`. On a large `adcp_reporting_changes` table create the index first with + `CREATE INDEX CONCURRENTLY IF NOT EXISTS adcp_reporting_changes_account_xid ON adcp_reporting_changes + (account_id, xid, seq)`, because a plain `CREATE INDEX` blocks appends while it builds; the migration then skips it. + The old `(account_id, seq)` index stays, because it serves the seq-only mapping; drop it once no older host runs. + **Do not run an older host's change maintenance (`prune_changes`, `retire_expired_periods`) alongside the new + reader.** The older code holds an account consumer's changes by `seq` alone and records only `pruned_through_seq`, + so against a consumer holding new-format cursors it can prune a change the consumer has not read without advancing + the horizon the new reader checks. Stop the older hosts' maintenance, or the hosts, first. Rolling back after a + consumer saved a new-format account cursor is not supported (the older SDK rejects it): reset those consumers. - A cursor older than the retained history raises `ReportingChangeCursorExpiredError` (`code == "CURSOR_EXPIRED"`). Resynchronize from `list_current_revisions` and resume from the cursor returned with that read. - **A read with no cursor can raise `CURSOR_EXPIRED` too.** `changes_after` reads the prune horizon, reads the page, @@ -418,12 +444,15 @@ await store.changes_after(account_id="acct_a", cursor=page.cursor) The page is rejected instead. Retrying the read starts from the new, later horizon, which is the oldest retained change; a consumer that had been reading from a cursor resynchronizes from `list_current_revisions` as usual. Handle `CURSOR_EXPIRED` on every `changes_after` call, including a first read, by matching `error.code`. -- A registered consumer holds back `prune_changes` until it has passed those changes in its own feed order, up to - `max_feed_hold_days` (default 7) after its last update, so a dead consumer cannot block retention. Schedule - `prune_changes(change_retention_days=30)` next to the other maintenance. Cursors are the same bytes in both SDKs. +- A registered consumer holds back `prune_changes` and `retire_expired_periods` until it has passed those changes in + `(xid, seq)` order (an account consumer only within its account), up to `max_feed_hold_days` (default 7) after its + last update, so a dead consumer cannot block retention. `prune_changes` takes its victims in `(xid, seq)` order and + records a per-account `(xid, seq)` horizon, so a batch cut by its limit never leaves a retained change below the + horizon. Schedule `prune_changes(change_retention_days=30)` next to the other maintenance. Cursors are the same bytes + in both SDKs. - **Checkpoints are forward-only within a feed scope.** `save_feed_consumer_cursor` ignores a cursor behind the saved one, so a delayed or repeated acknowledgment cannot pull the consumer, or its retention hold, backward and make it - reprocess changes. Account consumers compare `seq`; deployment consumers compare `(xid, seq)`. The call returns + reprocess changes. Every consumer compares `(xid, seq)`. The call returns `True` when it wrote the checkpoint and `False` when it ignored it (behind the saved position, or equal with `refresh_unchanged=False`). An equal cursor by default rewrites the checkpoint and renews the hold, which is the heartbeat for an idle consumer. A change of scope (account to deployment, deployment to account, or between diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/change_feed.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/change_feed.py index ac1e7b6ca0..5df003e57d 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/change_feed.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/change_feed.py @@ -2,13 +2,17 @@ ``PgReportingLedgerStore(change_feed=True)`` appends one row to ``adcp_reporting_changes`` per committed obligation, revision, adjustment and retirement, in the transaction that -writes the record and after the account lock, so per-account sequence order is commit -order. The JavaScript SDK writes the same table, so a host reads one feed. - -* **Account feeds** order by ``seq``. -* **The deployment-wide feed** orders by ``(xid, seq)`` and returns only changes whose - transaction precedes every transaction still in flight, so a slow commit is never - skipped. A long-running transaction delays this feed; it never makes it lose a record. +writes the record. The JavaScript SDK writes the same table, so a host reads one feed. + +* **Every feed**, account-scoped or deployment-wide, orders by ``(xid, seq)`` and returns + only changes whose transaction precedes every transaction still in flight + (``xid < pg_snapshot_xmin(pg_current_snapshot())``), so a slow commit is never skipped. + The order does not depend on the account lock, on the order a writer drew ``seq`` in, or + on which lock key a writer took: a change with a lower ``seq`` and a higher ``xid`` that + commits late is returned on a later page, after the cursor. A long-running transaction + delays every feed; it never makes one lose a record. +* **Account cursors carry ``xid``.** A cursor from before that, which carries only ``seq``, + is accepted and mapped to a position that skips nothing (see :func:`changes_after`). * Cursors are opaque, versioned and stable across restarts, processes and SDKs. A cursor older than the retained history raises :class:`ReportingChangeCursorExpiredError`; the consumer then resynchronizes from ``list_current_revisions`` and resumes from the @@ -16,7 +20,8 @@ commits mid-read raises the same error (for a cursorless read too) instead of returning a page that skipped the pruned changes. * A named consumer registered with ``save_feed_consumer_cursor`` holds back change pruning - (and period retirement) until it has passed those changes, up to ``max_feed_hold_days`` + (and period retirement) until it has passed those changes in ``(xid, seq)`` order, up to + ``max_feed_hold_days`` after its last update, so a dead consumer cannot block retention indefinitely. Its checkpoint is forward-only within a feed scope; ``rewind=True`` replays deliberately and ``remove_feed_consumer`` unregisters it. @@ -35,7 +40,7 @@ from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal +from typing import TYPE_CHECKING, Any, Literal, NamedTuple if TYPE_CHECKING: from adcp_reporting_storage.ledger.pg import PgReportingLedgerStore @@ -49,6 +54,7 @@ "changes_after", "decode_reporting_change_cursor_v1", "encode_reporting_change_cursor_v1", + "is_legacy_account_change_cursor", "prune_changes", "remove_feed_consumer", "save_feed_consumer_cursor", @@ -93,11 +99,34 @@ def __init__(self) -> None: ) +_INT64_MAX = 2**63 - 1 +_UINT64_MAX = 2**64 - 1 + + +def _fits_unsigned(value: object, maximum: int) -> bool: + """A decimal string PostgreSQL's ``bigint`` / ``xid8`` cast would accept. + + A value it would reject must fail as ``ValueError`` here, not as a raw database error + mid-query. ASCII digits only: ``\\d`` also matches other scripts' digits, which ``int`` + accepts and the database cast does not. + """ + return ( + isinstance(value, str) + and re.fullmatch(r"[0-9]{1,20}", value) is not None + and int(value) <= maximum + ) + + def encode_reporting_change_cursor_v1( *, account: str | None, seq: str, xid: str | None = None ) -> str: + """Encode a cursor. Every cursor carries ``xid``; only a seq-only account cursor omits it. + + Omitting ``xid`` for an account is how a cursor from before the account feed ordered by + ``(xid, seq)`` is spelled; :func:`changes_after` maps one to a position that skips nothing. + """ cursor: dict[str, Any] = {"v": 1, "account": account, "seq": seq} - if account is None: + if account is None or xid is not None: cursor["xid"] = xid raw = json.dumps(cursor, separators=(",", ":"), ensure_ascii=False).encode("utf-8") return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") @@ -114,18 +143,26 @@ def decode_reporting_change_cursor_v1(value: str, account: str | None) -> dict[s not isinstance(parsed, dict) or parsed.get("v") != 1 or parsed.get("account") != account - or not isinstance(parsed.get("seq"), str) - or not re.fullmatch(r"\d{1,19}", parsed["seq"]) + or not _fits_unsigned(parsed.get("seq"), _INT64_MAX) or ( - not isinstance(parsed.get("xid"), str) or not re.fullmatch(r"\d{1,20}", parsed["xid"]) - if account is None - else "xid" in parsed + account is None + if "xid" not in parsed + else not _fits_unsigned(parsed["xid"], _UINT64_MAX) ) ): raise ValueError("change cursor does not belong to this feed") return parsed +def is_legacy_account_change_cursor(cursor: dict[str, Any]) -> bool: + """True for an account cursor issued before account feeds ordered by ``(xid, seq)``. + + It carries a ``seq`` and no ``xid``. :func:`changes_after` accepts it and maps it to a + position that cannot skip a change. + """ + return cursor.get("account") is not None and "xid" not in cursor + + def _positive(value: int, name: str, maximum: int) -> None: if type(value) is not int or not 1 <= value <= maximum: raise ValueError(f"{name} must be between 1 and {maximum}") @@ -136,8 +173,31 @@ def _require_feed(store: PgReportingLedgerStore) -> None: raise ValueError("this operation requires the store change_feed option") +def feed_consumer_behind_change(consumer: str, change: str) -> str: + """SQL condition: the registered consumer row has not yet passed the change row. + + In the consumer's own feed order: every consumer reads ``(xid, seq)`` order, an account + consumer only within its account. A row saved before account feeds carried an ``xid`` has + a NULL ``cursor_xid`` and holds by ``seq`` until the consumer saves its next cursor. + Shared by change pruning, period retirement and anything else that honours consumer holds. + """ + return ( + f"(({consumer}.account_id IS NULL OR {consumer}.account_id = {change}.account_id)" + f" AND CASE WHEN {consumer}.cursor_xid IS NULL" + f" THEN {change}.seq > {consumer}.cursor_seq" + f" ELSE ({change}.xid, {change}.seq) > ({consumer}.cursor_xid, {consumer}.cursor_seq)" + " END)" + ) + + def _page_query(*, account: bool) -> str: - """The change page query for an account feed (``seq`` order) or the deployment feed. + """The change page query for an account feed or the deployment feed. + + Both feeds order by ``(xid, seq)`` and stop below the reading snapshot's ``xmin``. Every + change with a higher ``xid`` than a returned one either committed before it or is not + returned yet, so a cursor never passes a change that commits late, with no assumption about + the lock a writer took or the order it drew ``seq`` in. The account scan is served by the + ``(account_id, xid, seq)`` index, in order. Every column is qualified with the table alias, in the projection and in ORDER BY. ``seq::text`` is *named* ``seq``, and PostgreSQL resolves a bare ORDER BY name to the output @@ -150,57 +210,84 @@ def _page_query(*, account: bool) -> str: " to_char(change.recorded_at AT TIME ZONE 'UTC'," ' \'YYYY-MM-DD"T"HH24:MI:SS.US"Z"\')' ) - if account: - return ( - f"SELECT {columns}," # noqa: S608 # nosec B608 - " change.xid >= pg_snapshot_xmin(pg_current_snapshot()) AS in_flight" - " FROM adcp_reporting_changes change" - " WHERE change.account_id = %s AND change.seq > %s::bigint" - " AND (%s::text[] IS NULL OR change.record_kind = ANY(%s::text[]))" - " ORDER BY change.seq LIMIT %s" - ) + scope = " change.account_id = %s AND" if account else "" return ( f"SELECT {columns} FROM adcp_reporting_changes change" # noqa: S608 # nosec B608 - " WHERE (change.xid, change.seq) > (%s::text::xid8, %s::bigint)" + f" WHERE{scope} (change.xid, change.seq) > (%s::text::xid8, %s::bigint)" " AND change.xid < pg_snapshot_xmin(pg_current_snapshot())" " AND (%s::text[] IS NULL OR change.record_kind = ANY(%s::text[]))" " ORDER BY change.xid, change.seq LIMIT %s" ) -async def _read_horizon(connection: Any, account_id: str | None) -> tuple[str, str]: - """The pruned horizon ``(xid, seq)`` in this feed's order; ``xid`` is ``"0"`` for accounts.""" +class _Horizon(NamedTuple): + xid: str + #: The ``seq`` of the highest ``(xid, seq)`` pruned. + seq: str + #: The highest ``seq`` pruned whatever its ``xid``: the exact check for a seq-only cursor. + max_seq: str + + +async def _read_horizon(connection: Any, account_id: str | None) -> _Horizon: + """The pruned horizon in this feed's own order.""" if account_id is not None: row = await ( await connection.execute( - "SELECT pruned_through_seq::text FROM adcp_reporting_change_account_horizon" + "SELECT pruned_through_xid::text, pruned_through_xid_seq::text," + " pruned_through_seq::text FROM adcp_reporting_change_account_horizon" " WHERE account_id = %s", (account_id,), ) ).fetchone() - return "0", (row[0] if row else "0") - row = await ( - await connection.execute( - "SELECT pruned_through_xid::text, pruned_through_seq::text" - " FROM adcp_reporting_change_horizon WHERE singleton" - ) - ).fetchone() - return (row[0], row[1]) if row else ("0", "0") + else: + row = await ( + await connection.execute( + "SELECT pruned_through_xid::text, pruned_through_seq::text," + " pruned_through_seq::text FROM adcp_reporting_change_horizon WHERE singleton" + ) + ).fetchone() + return _Horizon(*row) if row else _Horizon("0", "0", "0") + +def _assert_retained(position: tuple[str | None, str], horizon: _Horizon) -> None: + """Raise unless ``position`` ``(xid, seq)`` is at or past the pruned ``horizon``. -def _assert_retained( - account_id: str | None, position: tuple[str, str], horizon: tuple[str, str] -) -> None: - """Raise unless ``position`` ``(xid, seq)`` is at or past the pruned ``horizon``.""" + A position without an ``xid`` is a seq-only account cursor, checked against the highest + ``seq`` ever pruned: a tuple compare would miss a pruned change with a higher ``seq`` and + a lower ``xid``. + """ + xid, seq = position behind = ( - int(horizon[1]) > int(position[1]) - if account_id is not None - else (int(horizon[0]), int(horizon[1])) > (int(position[0]), int(position[1])) + int(horizon.max_seq) > int(seq) + if xid is None + else (int(horizon.xid), int(horizon.seq)) > (int(xid), int(seq)) ) if behind: raise ReportingChangeCursorExpiredError() +async def _legacy_account_cursor_xid(connection: Any, account_id: str, seq: str) -> str: + """Map a seq-only account cursor to a transaction id that cannot skip a change. + + Every change after the cursor's ``seq`` has an ``xid`` at or above both the oldest such + change that is visible now and the reading snapshot's ``xmin`` (a change that is not + visible yet belongs to a transaction that was in flight, or started after the snapshot, + whose id is at or above ``xmin``), so the position ``(xid, seq)`` below the smaller of the + two hands them all out. + """ + row = await ( + await connection.execute( + "SELECT LEAST(" + " pg_snapshot_xmin(pg_current_snapshot())," + " COALESCE((SELECT change.xid FROM adcp_reporting_changes change" + " WHERE change.account_id = %s AND change.seq > %s::bigint" + " ORDER BY change.xid LIMIT 1), pg_snapshot_xmin(pg_current_snapshot())))::text", + (account_id, seq), + ) + ).fetchone() + return str(row[0]) + + async def changes_after( store: PgReportingLedgerStore, *, @@ -211,6 +298,19 @@ async def changes_after( ) -> ReportingChangesPageV1: """Read the host change feed. Omit ``cursor`` to start at the oldest retained change. + Every feed, account-scoped or deployment-wide, orders by ``(xid, seq)`` and returns only + changes whose transaction precedes every transaction still in flight, so no commit is ever + skipped and the order does not depend on the account lock. An empty page can therefore + mean caught up or held back by an older open transaction (``xmin`` is cluster-wide). + + An account cursor issued before account feeds carried an ``xid`` holds only a ``seq``. It + is accepted and mapped to a position that cannot skip a change: the lowest transaction id + any change after its ``seq`` could have (the smaller of the oldest such change's ``xid`` + and the reading snapshot's ``xmin``). That can hand out a few changes the consumer already + read, those with a lower ``seq`` and a higher ``xid``, so consumers must treat redelivery + as normal. The page returns a cursor in the new format, even when it is empty. A seq-only + cursor raises ``CURSOR_EXPIRED`` when any change with a higher ``seq`` was pruned. + Raises :class:`ReportingChangeCursorExpiredError` (``code == "CURSOR_EXPIRED"``) when the starting position is older than the retained history. A read **without** a ``cursor`` can raise it too: the pruned horizon is read again after the page, so a prune that commits @@ -222,23 +322,31 @@ async def changes_after( if kinds is not None and any(kind not in _KINDS for kind in kinds): raise ValueError("unknown change kind") decoded = decode_reporting_change_cursor_v1(cursor, account_id) if cursor is not None else None + legacy = decoded is not None and is_legacy_account_change_cursor(decoded) selected = list(kinds) if kinds else None async with store._pool.connection() as connection: # The pruned horizon in this feed's own order. A supplied cursor below it may have # missed a pruned change and fails closed; a consumer with no cursor begins at it. - horizon_xid, horizon_seq = await _read_horizon(connection, account_id) + horizon = await _read_horizon(connection, account_id) + checked: tuple[str | None, str] | None = None if decoded is not None: - _assert_retained( - account_id, (decoded.get("xid", "0"), decoded["seq"]), (horizon_xid, horizon_seq) + checked = (decoded.get("xid"), decoded["seq"]) + _assert_retained(checked, horizon) + position_seq = decoded["seq"] + position_xid = ( + await _legacy_account_cursor_xid(connection, account_id, position_seq) + if legacy and account_id is not None + else decoded["xid"] ) - position_seq = decoded["seq"] if decoded else horizon_seq - position_xid = decoded.get("xid", "0") if decoded else horizon_xid - query = _page_query(account=account_id is not None) - if account_id is not None: - params: tuple[Any, ...] = (account_id, position_seq, selected, selected, limit) else: - params = (position_xid, position_seq, selected, selected, limit) - rows = await (await connection.execute(query, params)).fetchall() + position_xid, position_seq = horizon.xid, horizon.seq + query = _page_query(account=account_id is not None) + scope: tuple[Any, ...] = (account_id,) if account_id is not None else () + rows = await ( + await connection.execute( + query, (*scope, position_xid, position_seq, selected, selected, limit) + ) + ).fetchall() # Pruning can commit between the horizon read and the page read, and a prune deletes rows # and advances the horizon in one transaction. Read the horizon again, as a new statement # after the page, and compare it with the position this page started from: a page read @@ -246,31 +354,21 @@ async def changes_after( # skip them silently. This covers a read with no cursor too. Retrying starts from the # new, later horizon. A prune that commits after this read cannot have affected the page. _assert_retained( - account_id, (position_xid, position_seq), await _read_horizon(connection, account_id) + checked or (position_xid, position_seq), await _read_horizon(connection, account_id) ) - if account_id is not None: - # The account feed orders by ``seq``, which equals commit order only while every writer - # takes the account lock before ``nextval``. Defense in depth for a writer that does not: - # stop the page at the first change whose transaction is not below every in-flight - # transaction (``xid >= xmin``). A plain ``xid < xmin`` filter would not do, because seq - # order and xid order can differ, and it would hand out a later change while skipping an - # earlier one it excluded. Truncating keeps the page a prefix of seq order. This narrows - # the window for a second writer but does not close it for every interleaving: one - # writer SDK per database remains the contract. ``xmin`` is cluster-wide, so any older - # open transaction holds account pages back until it ends; an empty page can mean - # "caught up" or "held back". The row scan is still served by the (account_id, seq) index. - blocked = next((index for index, row in enumerate(rows) if row[7]), None) - if blocked is not None: - rows = rows[:blocked] records = tuple(ReportingChangeRecordV1(row[3], row[4], row[5], row[2], row[6]) for row in rows) if rows: last = rows[-1] next_cursor = encode_reporting_change_cursor_v1( - account=account_id, seq=last[0], xid=last[1] if account_id is None else None + account=account_id, seq=last[0], xid=last[1] ) + elif cursor is not None and not legacy: + next_cursor = cursor else: - next_cursor = cursor or encode_reporting_change_cursor_v1( - account=account_id, seq=position_seq, xid=position_xid if account_id is None else None + # A seq-only cursor is replaced by its mapped position even on an empty page, so the + # consumer saves a cursor in the new format. + next_cursor = encode_reporting_change_cursor_v1( + account=account_id, seq=position_seq, xid=position_xid ) return ReportingChangesPageV1(records, next_cursor) @@ -290,8 +388,8 @@ async def save_feed_consumer_cursor( Checkpoints are forward-only within a feed scope: a cursor behind the saved one is ignored, so a delayed or repeated acknowledgment cannot pull the retention hold back or - make the consumer reprocess changes. Account consumers compare ``seq``; deployment - consumers compare ``(xid, seq)``. Returns ``True`` when the checkpoint was written and + make the consumer reprocess changes. Every consumer compares ``(xid, seq)``. Returns + ``True`` when the checkpoint was written and ``False`` when it was ignored (behind the saved position, or equal with ``refresh_unchanged=False``). @@ -306,6 +404,12 @@ async def save_feed_consumer_cursor( ``refresh_unchanged=False`` recovers an acknowledgment without renewing the retention hold when its position has not advanced. + + A seq-only account cursor (from before account feeds carried an ``xid``) is mapped to the + position :func:`changes_after` would read it from, so the hold sits below every unread + change. A consumer row saved before that upgrade has a NULL ``cursor_xid`` and holds by + ``seq`` until its next save; a cursor in the new format replaces it unless its ``seq`` is + behind. """ _require_feed(store) if not isinstance(name, str) or not _CONSUMER_NAME.match(name): @@ -320,8 +424,16 @@ async def save_feed_consumer_cursor( decoded = decode_reporting_change_cursor_v1(cursor, account) # One atomic upsert: the WHERE of DO UPDATE is re-evaluated against the latest committed # row version under the row lock, so concurrent writers cannot interleave a stale position - # over a newer one. No row is returned when the update is skipped. + # over a newer one. No row is returned when the update is skipped. Forward-only in + # (xid, seq) order for every scope. A row saved before account feeds carried an xid (NULL + # cursor_xid) has only a seq to compare: a cursor in the new format replaces it unless its + # seq is behind. async with store._pool.connection() as connection: + xid = ( + await _legacy_account_cursor_xid(connection, account, decoded["seq"]) + if account is not None and is_legacy_account_change_cursor(decoded) + else decoded["xid"] + ) result = await connection.execute( "INSERT INTO adcp_reporting_feed_consumers" " (consumer_name, account_id, cursor_xid, cursor_seq)" @@ -331,22 +443,20 @@ async def save_feed_consumer_cursor( " cursor_seq = EXCLUDED.cursor_seq, updated_at = clock_timestamp()" " WHERE %(rewind)s::boolean" " OR adcp_reporting_feed_consumers.account_id IS DISTINCT FROM EXCLUDED.account_id" - " OR CASE WHEN EXCLUDED.account_id IS NULL" - " THEN (adcp_reporting_feed_consumers.cursor_xid," - " adcp_reporting_feed_consumers.cursor_seq)" - " < (EXCLUDED.cursor_xid, EXCLUDED.cursor_seq)" - " OR (%(refresh)s::boolean AND" - " (adcp_reporting_feed_consumers.cursor_xid," - " adcp_reporting_feed_consumers.cursor_seq)" - " = (EXCLUDED.cursor_xid, EXCLUDED.cursor_seq))" - " ELSE adcp_reporting_feed_consumers.cursor_seq < EXCLUDED.cursor_seq" - " OR (%(refresh)s::boolean" - " AND adcp_reporting_feed_consumers.cursor_seq = EXCLUDED.cursor_seq) END" + " OR (adcp_reporting_feed_consumers.cursor_xid IS NULL" + " AND adcp_reporting_feed_consumers.cursor_seq <= EXCLUDED.cursor_seq)" + " OR (adcp_reporting_feed_consumers.cursor_xid," + " adcp_reporting_feed_consumers.cursor_seq)" + " < (EXCLUDED.cursor_xid, EXCLUDED.cursor_seq)" + " OR (%(refresh)s::boolean AND" + " (adcp_reporting_feed_consumers.cursor_xid," + " adcp_reporting_feed_consumers.cursor_seq)" + " = (EXCLUDED.cursor_xid, EXCLUDED.cursor_seq))" " RETURNING consumer_name", { "name": name, "account": account, - "xid": decoded.get("xid") if account is None else None, + "xid": xid, "seq": decoded["seq"], "refresh": refresh_unchanged, "rewind": rewind, @@ -393,18 +503,16 @@ async def prune_changes( async with store._pool.connection() as connection, connection.transaction(): victims = await ( await connection.execute( - "WITH victims AS (" + "WITH victims AS (" # noqa: S608 # nosec B608 " SELECT change.seq FROM adcp_reporting_changes change" " WHERE change.recorded_at < clock_timestamp() - (%s::integer * INTERVAL '1 day')" " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_feed_consumers consumer" " WHERE consumer.updated_at >" " clock_timestamp() - (%s::integer * INTERVAL '1 day')" - " AND ((consumer.account_id IS NULL" - " AND (change.xid, change.seq) >" - " (consumer.cursor_xid, consumer.cursor_seq))" - " OR (consumer.account_id = change.account_id" - " AND change.seq > consumer.cursor_seq)))" - " ORDER BY change.seq LIMIT %s)" + f" AND {feed_consumer_behind_change('consumer', 'change')})" + # Victims in (xid, seq) order, so a batch cut by the limit leaves nothing + # retained below the per-account (xid, seq) horizon recorded below. + " ORDER BY change.xid, change.seq LIMIT %s)" " DELETE FROM adcp_reporting_changes change USING victims" " WHERE change.seq = victims.seq" " RETURNING change.seq::text, change.xid::text, change.account_id", @@ -421,16 +529,38 @@ async def prune_changes( " (pruned_through_xid, pruned_through_seq)", (highest[1], highest[0], highest[1], highest[0]), ) - per_account: dict[str, int] = {} - for seq, _, account in victims: - per_account[account] = max(per_account.get(account, 0), int(seq)) - for account, seq in per_account.items(): + # Per account: the highest (xid, seq) pruned, which a cursor is compared with, and the + # highest seq pruned whatever its xid, which is the exact check for a seq-only cursor. + per_account: dict[str, tuple[int, int, int]] = {} + for seq_text, xid_text, account in victims: + seq, xid = int(seq_text), int(xid_text) + top_xid, top_seq, max_seq = per_account.get(account, (xid, seq, seq)) + if (xid, seq) > (top_xid, top_seq): + top_xid, top_seq = xid, seq + per_account[account] = (top_xid, top_seq, max(max_seq, seq)) + for account, (top_xid, top_seq, max_seq) in per_account.items(): await connection.execute( - "INSERT INTO adcp_reporting_change_account_horizon (account_id, pruned_through_seq)" - " VALUES (%s, %s::bigint) ON CONFLICT (account_id) DO UPDATE" - " SET pruned_through_seq = GREATEST(" - " adcp_reporting_change_account_horizon.pruned_through_seq," - " EXCLUDED.pruned_through_seq)", - (account, seq), + "INSERT INTO adcp_reporting_change_account_horizon AS horizon" + " (account_id, pruned_through_seq, pruned_through_xid, pruned_through_xid_seq)" + " VALUES (%(account)s, %(max_seq)s::bigint, %(xid)s::text::xid8," + " %(xid_seq)s::bigint)" + " ON CONFLICT (account_id) DO UPDATE" + " SET pruned_through_seq = GREATEST(horizon.pruned_through_seq," + " EXCLUDED.pruned_through_seq)," + " pruned_through_xid = CASE" + " WHEN (EXCLUDED.pruned_through_xid, EXCLUDED.pruned_through_xid_seq)" + " > (horizon.pruned_through_xid, horizon.pruned_through_xid_seq)" + " THEN EXCLUDED.pruned_through_xid ELSE horizon.pruned_through_xid END," + " pruned_through_xid_seq = CASE" + " WHEN (EXCLUDED.pruned_through_xid, EXCLUDED.pruned_through_xid_seq)" + " > (horizon.pruned_through_xid, horizon.pruned_through_xid_seq)" + " THEN EXCLUDED.pruned_through_xid_seq" + " ELSE horizon.pruned_through_xid_seq END", + { + "account": account, + "max_seq": str(max_seq), + "xid": str(top_xid), + "xid_seq": str(top_seq), + }, ) return len(victims) diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/period_retirement.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/period_retirement.py index 20354f3ace..21707be5ce 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/period_retirement.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/period_retirement.py @@ -44,6 +44,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any +from adcp_reporting_storage.ledger.change_feed import feed_consumer_behind_change from adcp_reporting_storage.ledger.row_retention import NoLongerEligibleError from adcp_reporting_storage.ledger.row_storage import ReportingRowStoreError @@ -122,15 +123,12 @@ async def retire_expired_periods( connection, "adcp_reporting_feed_consumers" ): holds.append( - " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_changes change" + " AND NOT EXISTS (SELECT 1 FROM adcp_reporting_changes change" # noqa: S608 # nosec B608 " JOIN adcp_reporting_feed_consumers consumer" " ON consumer.updated_at > clock_timestamp()" " - (%(hold_days)s::integer * INTERVAL '1 day')" " WHERE change.obligation_id = o.reporting_obligation_id" - " AND ((consumer.account_id IS NULL AND (change.xid, change.seq) >" - " (consumer.cursor_xid, consumer.cursor_seq))" - " OR (consumer.account_id = change.account_id" - " AND change.seq > consumer.cursor_seq)))" + f" AND {feed_consumer_behind_change('consumer', 'change')})" ) parameters = { "account": account_id, diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py index 8ab1f7ee27..fc5c442d40 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/pg.py @@ -572,8 +572,9 @@ async def changes_after( ) -> ReportingChangesPageV1: """Read the shared host change feed; see :mod:`adcp_reporting_storage.ledger.change_feed`. - Account feeds order by ``seq``; the deployment-wide feed orders by ``(xid, seq)`` - below the reading snapshot's ``xmin``. A cursor older than the retained history + Every feed, account-scoped or deployment-wide, orders by ``(xid, seq)`` below the + reading snapshot's ``xmin``; an account cursor from before that carries only ``seq`` + and is mapped to a position that skips nothing. A cursor older than the retained history raises ``ReportingChangeCursorExpiredError``, and so does a read with no cursor when a prune commits while the page is being read (the horizon is read again after the page). Requires ``change_feed=True``. diff --git a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_changes.sql b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_changes.sql index e114507b98..a3ae811163 100644 --- a/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_changes.sql +++ b/packages/adcp-reporting-storage/src/adcp_reporting_storage/ledger/reporting_changes.sql @@ -8,9 +8,16 @@ -- PgReportingLedgerStore(change_feed=True) applies it from create_schema. -- -- A row is written in the same transaction as each obligation, revision, adjustment --- and retirement, taking nextval only after the account lock, so per-account sequence --- order is commit order. Deployment-wide readers order by (xid, seq) and stop below --- the reading snapshot's xmin, so a slow commit is never skipped. +-- and retirement. Every reader, account-scoped or deployment-wide, orders by (xid, seq) +-- and stops below the reading snapshot's xmin, so a slow commit is never skipped and no +-- feed depends on the account lock or on the order a writer drew seq in. Account cursors +-- carry xid; a seq-only cursor from before that is mapped to a position that skips +-- nothing (adcp-client#3289, adcp-client-python#1825). +-- +-- Re-applying this file upgrades an install from before the account feed ordered by +-- (xid, seq): it adds the account horizon's xid columns, replaces the consumer table's +-- check constraint and creates the (account_id, xid, seq) index. On a large changes table +-- create that index first with CREATE INDEX CONCURRENTLY; this file then skips it. -- -- This table is separate from reporting_ledger_changes, the buyer-facing feed behind -- the protocol's changes_checkpoint and changes_after. They converge with the shared @@ -34,8 +41,13 @@ CREATE TABLE IF NOT EXISTS adcp_reporting_changes ( obligation_id TEXT NOT NULL, recorded_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() ); +-- Superseded by adcp_reporting_changes_account_xid for the account feed; kept so +-- re-applying this file never drops an index, and because it serves the mapping of a +-- seq-only account cursor. Drop it once no older host runs and no such cursor remains. CREATE INDEX IF NOT EXISTS adcp_reporting_changes_account ON adcp_reporting_changes (account_id, seq); +CREATE INDEX IF NOT EXISTS adcp_reporting_changes_account_xid + ON adcp_reporting_changes (account_id, xid, seq); CREATE INDEX IF NOT EXISTS adcp_reporting_changes_deployment ON adcp_reporting_changes (xid, seq); CREATE INDEX IF NOT EXISTS adcp_reporting_changes_recorded @@ -50,19 +62,56 @@ CREATE TABLE IF NOT EXISTS adcp_reporting_change_horizon ( ); INSERT INTO adcp_reporting_change_horizon (singleton) VALUES (TRUE) ON CONFLICT DO NOTHING; --- Highest seq ever pruned per account: an account cursor below it fails closed. +-- Highest (xid, seq) ever pruned per account: an account cursor below it fails closed. +-- pruned_through_seq is the highest seq pruned whatever its xid; account feeds no longer +-- compare it, it exists so a seq-only cursor from before the (xid, seq) order can still be +-- checked exactly (a tuple compare alone would miss a pruned change with a higher seq and +-- a lower xid). CREATE TABLE IF NOT EXISTS adcp_reporting_change_account_horizon ( account_id TEXT PRIMARY KEY, - pruned_through_seq BIGINT NOT NULL + pruned_through_seq BIGINT NOT NULL, + pruned_through_xid xid8 NOT NULL DEFAULT '0', + pruned_through_xid_seq BIGINT NOT NULL DEFAULT 0 ); +ALTER TABLE adcp_reporting_change_account_horizon + ADD COLUMN IF NOT EXISTS pruned_through_xid xid8 NOT NULL DEFAULT '0'; +ALTER TABLE adcp_reporting_change_account_horizon + ADD COLUMN IF NOT EXISTS pruned_through_xid_seq BIGINT NOT NULL DEFAULT 0; --- Registered consumers hold back pruning in their own feed order: account --- consumers by seq within their account, deployment consumers by (xid, seq). +-- Registered consumers hold back pruning in (xid, seq) order, per account for an account +-- consumer. A row saved before the account feed carried an xid has a NULL cursor_xid and holds +-- by seq until the consumer saves its next cursor. CREATE TABLE IF NOT EXISTS adcp_reporting_feed_consumers ( consumer_name TEXT PRIMARY KEY CHECK (consumer_name ~ '^[A-Za-z0-9_.:-]{1,128}$'), account_id TEXT, cursor_xid xid8, cursor_seq BIGINT NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), - CHECK ((account_id IS NULL) = (cursor_xid IS NOT NULL)) + CONSTRAINT adcp_reporting_feed_consumers_cursor_scope + CHECK (account_id IS NOT NULL OR cursor_xid IS NOT NULL) ); +-- An install from before the account feed carried an xid has the anonymous check +-- (account_id IS NULL) = (cursor_xid IS NOT NULL), which rejects an account consumer that has +-- an xid. Drop it and add the named one; both steps are guarded, so re-applying is a no-op. +DO $adcp_feed_consumers$ +DECLARE stale TEXT; +BEGIN + FOR stale IN + SELECT conname FROM pg_constraint + WHERE conrelid = 'adcp_reporting_feed_consumers'::regclass AND contype = 'c' + AND conname <> 'adcp_reporting_feed_consumers_cursor_scope' + AND strpos(pg_get_constraintdef(oid), 'account_id IS NULL') > 0 + LOOP + EXECUTE format('ALTER TABLE adcp_reporting_feed_consumers DROP CONSTRAINT %I', stale); + END LOOP; + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'adcp_reporting_feed_consumers'::regclass + AND conname = 'adcp_reporting_feed_consumers_cursor_scope' + ) THEN + ALTER TABLE adcp_reporting_feed_consumers + ADD CONSTRAINT adcp_reporting_feed_consumers_cursor_scope + CHECK (account_id IS NOT NULL OR cursor_xid IS NOT NULL); + END IF; +END +$adcp_feed_consumers$; diff --git a/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py b/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py index b7f7195944..ef1fabbb1a 100644 --- a/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py +++ b/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py @@ -1,7 +1,8 @@ """Feed consumer checkpoints are forward-only within a scope (adcp-client-python #1509). Ported from the JavaScript SDK (adcp-client #3197): a delayed or repeated acknowledgment -must not move a consumer, or its retention hold, backward. +must not move a consumer, or its retention hold, backward. Account consumers compare +``(xid, seq)`` like deployment consumers (adcp-client #3289, adcp-client-python #1825). """ from __future__ import annotations @@ -24,7 +25,15 @@ async def feed_store(pool: Any) -> PgReportingLedgerStore: return store -def account_cursor(seq: int, account: str = "acct_a") -> str: +def account_cursor(seq: int, account: str = "acct_a", xid: int | None = None) -> str: + """A new-format account cursor; ``xid`` defaults to ``seq`` so the two orders agree.""" + return encode_reporting_change_cursor_v1( + account=account, seq=str(seq), xid=str(seq if xid is None else xid) + ) + + +def legacy_account_cursor(seq: int, account: str = "acct_a") -> str: + """The cursor an older SDK saved: a ``seq`` and no ``xid``.""" return encode_reporting_change_cursor_v1(account=account, seq=str(seq)) @@ -66,29 +75,29 @@ async def test_an_account_consumer_never_moves_backward_unless_asked_to() -> Non store = await feed_store(pool) assert await store.save_feed_consumer_cursor("sink", account_cursor(10)) is True assert await store.save_feed_consumer_cursor("sink", account_cursor(20)) is True - assert await saved(pool, "sink") == ("acct_a", None, 20) + assert await saved(pool, "sink") == ("acct_a", "20", 20) # A delayed acknowledgment is ignored and says so. applied = await store.save_feed_consumer_cursor("sink", account_cursor(15)) - assert await saved(pool, "sink") == ("acct_a", None, 20) + assert await saved(pool, "sink") == ("acct_a", "20", 20) assert applied is False # A deliberate replay rewinds, with or without the refresh option. assert await store.save_feed_consumer_cursor("sink", account_cursor(5), rewind=True) is True - assert await saved(pool, "sink") == ("acct_a", None, 5) + assert await saved(pool, "sink") == ("acct_a", "5", 5) assert ( await store.save_feed_consumer_cursor( "sink", account_cursor(3), rewind=True, refresh_unchanged=False ) is True ) - assert await saved(pool, "sink") == ("acct_a", None, 3) + assert await saved(pool, "sink") == ("acct_a", "3", 3) # Compared numerically, not as text: 100 is ahead of 95. await store.save_feed_consumer_cursor("sink", account_cursor(95)) assert await store.save_feed_consumer_cursor("sink", account_cursor(100)) is True assert await store.save_feed_consumer_cursor("sink", account_cursor(95)) is False - assert await saved(pool, "sink") == ("acct_a", None, 100) + assert await saved(pool, "sink") == ("acct_a", "100", 100) async def test_a_repeated_acknowledgment_renews_the_hold_unless_told_not_to() -> None: @@ -151,11 +160,11 @@ async def test_a_change_of_feed_scope_replaces_the_position() -> None: await store.save_feed_consumer_cursor("sink", account_cursor(500, "acct_a")) # Positions in different feeds are not comparable, so a scope change always applies. assert await store.save_feed_consumer_cursor("sink", account_cursor(1, "acct_b")) is True - assert await saved(pool, "sink") == ("acct_b", None, 1) + assert await saved(pool, "sink") == ("acct_b", "1", 1) assert await store.save_feed_consumer_cursor("sink", deployment_cursor(5, 1)) is True assert await saved(pool, "sink") == (None, "5", 1) assert await store.save_feed_consumer_cursor("sink", account_cursor(900, "acct_a")) is True - assert await saved(pool, "sink") == ("acct_a", None, 900) + assert await saved(pool, "sink") == ("acct_a", "900", 900) async def test_concurrent_writers_cannot_interleave_a_stale_position_over_a_newer_one( @@ -216,7 +225,7 @@ async def connection(*args: Any, **kwargs: Any): task.cancel() await asyncio.gather(*tasks, return_exceptions=True) - assert await saved(pool, "sink") == ("acct_a", None, 9), "a stale write landed last" + assert await saved(pool, "sink") == ("acct_a", "9", 9), "a stale write landed last" assert results == (True, False) @@ -243,6 +252,77 @@ async def test_remove_feed_consumer_releases_the_hold() -> None: assert await store.remove_feed_consumer("sink") is False # Saving under the same name registers it again, at the position given. assert await store.save_feed_consumer_cursor("sink", account_cursor(2)) is True - assert await saved(pool, "sink") == ("acct_a", None, 2) + assert await saved(pool, "sink") == ("acct_a", "2", 2) with pytest.raises(ValueError): await store.remove_feed_consumer("bad name") + + +async def test_an_account_consumer_compares_xid_then_seq() -> None: + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + assert await store.save_feed_consumer_cursor("sink", account_cursor(9, xid=50)) is True + # A later transaction with a lower seq is ahead in (xid, seq) order; an earlier + # transaction with a higher seq is behind. + assert await store.save_feed_consumer_cursor("sink", account_cursor(2, xid=60)) is True + assert await saved(pool, "sink") == ("acct_a", "60", 2) + assert await store.save_feed_consumer_cursor("sink", account_cursor(99, xid=55)) is False + assert await saved(pool, "sink") == ("acct_a", "60", 2) + # Same transaction: seq decides. + assert await store.save_feed_consumer_cursor("sink", account_cursor(3, xid=60)) is True + assert await store.save_feed_consumer_cursor("sink", account_cursor(1, xid=60)) is False + assert await saved(pool, "sink") == ("acct_a", "60", 3) + + +async def test_a_null_xid_consumer_row_is_replaced_by_a_new_format_cursor_unless_its_seq_is_behind() -> ( + None +): + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_feed_consumers" + " (consumer_name, account_id, cursor_xid, cursor_seq)" + " VALUES ('old', 'acct_a', NULL, 9000)" + ) + # A consumer row saved before the upgrade holds by seq; a cursor behind it is ignored. + assert await store.save_feed_consumer_cursor("old", account_cursor(5, xid=7)) is False + assert await saved(pool, "old") == ("acct_a", None, 9000) + assert await store.save_feed_consumer_cursor("old", account_cursor(9000, xid=7)) is True + assert await saved(pool, "old") == ("acct_a", "7", 9000) + assert await store.save_feed_consumer_cursor("old", account_cursor(5, xid=7)) is False + assert ( + await store.save_feed_consumer_cursor("old", account_cursor(5, xid=7), rewind=True) + is True + ) + + +async def test_saving_a_seq_only_account_cursor_registers_the_position_changes_after_reads() -> ( + None +): + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_changes (xid, account_id, record_kind, record_id," + " obligation_id) VALUES ('10'::xid8, 'acct_a', 'revision', 'r1', 'o')," + " ('20'::xid8, 'acct_a', 'revision', 'r2', 'o')," + " ('30'::xid8, 'acct_a', 'revision', 'r3', 'o')" + ) + first_seq = (await fetch_all(pool, "SELECT seq FROM adcp_reporting_changes ORDER BY seq"))[ + 0 + ][0] + # The hold sits below every unread change: the oldest one after the cursor, xid 20. + assert await store.save_feed_consumer_cursor("sink", legacy_account_cursor(first_seq)) + assert await saved(pool, "sink") == ("acct_a", "20", first_seq) + page = await store.changes_after( + account_id="acct_a", cursor=legacy_account_cursor(first_seq) + ) + assert [r.record_id for r in page.records] == ["r2", "r3"] + # A new-format cursor replaces it, forward only. + assert await store.save_feed_consumer_cursor("sink", page.cursor) is True + row = await saved(pool, "sink") + assert row is not None and row[1] == "30" + # A mapped seq-only cursor is forward-only against a new-format registration. + assert ( + await store.save_feed_consumer_cursor("sink", legacy_account_cursor(first_seq)) is False + ) diff --git a/tests/conformance/reporting/test_reporting_change_feed_horizon_pg.py b/tests/conformance/reporting/test_reporting_change_feed_horizon_pg.py index 9948666ec6..b1c76c34b9 100644 --- a/tests/conformance/reporting/test_reporting_change_feed_horizon_pg.py +++ b/tests/conformance/reporting/test_reporting_change_feed_horizon_pg.py @@ -89,6 +89,7 @@ def __getattr__(self, name: str) -> Any: def feed_cases() -> list[Any]: return [ pytest.param(ACCOUNT, "with-cursor", id="account-with-cursor"), + pytest.param(ACCOUNT, "legacy-cursor", id="account-seq-only-cursor"), pytest.param(ACCOUNT, "no-cursor", id="account-without-cursor"), pytest.param(None, "with-cursor", id="deployment-with-cursor"), pytest.param(None, "no-cursor", id="deployment-without-cursor"), @@ -99,7 +100,9 @@ def starting_cursor(account: str | None, mode: str) -> str | None: if mode == "no-cursor": return None if account is not None: - return encode_reporting_change_cursor_v1(account=account, seq="0") + # A seq-only cursor is the form an older SDK saved: the horizon check is exact for it. + xid = None if mode == "legacy-cursor" else "0" + return encode_reporting_change_cursor_v1(account=account, seq="0", xid=xid) return encode_reporting_change_cursor_v1(account=None, seq="0", xid="0") @@ -150,7 +153,7 @@ async def test_a_prune_that_commits_before_the_read_starts_is_not_a_race( await seed_changes(pool) assert await store.prune_changes() == 2 - if mode == "with-cursor": + if mode != "no-cursor": with pytest.raises(ReportingChangeCursorExpiredError): await store.changes_after(account_id=account, cursor=starting_cursor(account, mode)) else: diff --git a/tests/conformance/reporting/test_reporting_change_feed_pg.py b/tests/conformance/reporting/test_reporting_change_feed_pg.py index fb3bcbe4a0..9d41968d37 100644 --- a/tests/conformance/reporting/test_reporting_change_feed_pg.py +++ b/tests/conformance/reporting/test_reporting_change_feed_pg.py @@ -2,7 +2,9 @@ from __future__ import annotations +import asyncio import base64 +import json from typing import Any import pytest @@ -11,12 +13,17 @@ ReportingChangeCursorExpiredError, decode_reporting_change_cursor_v1, encode_reporting_change_cursor_v1, + is_legacy_account_change_cursor, ) from ._generation_support import configuration, isolated_reporting_pool, obligation_for from .test_reporting_row_storage_pg import fetch_all, many_rows, typed_revision +def b64(body: dict[str, Any]) -> str: + return base64.urlsafe_b64encode(json.dumps(body).encode()).decode() + + async def feed_store(pool: Any) -> PgReportingLedgerStore: store = PgReportingLedgerStore(pool=pool, change_feed=True) await store.create_schema() @@ -42,10 +49,22 @@ def test_cursors_use_the_format_the_javascript_sdk_reads_and_writes() -> None: "seq": "7", "xid": "42", } + # A seq-only account cursor is what an older SDK saved; it is still accepted. js_account = ( base64.urlsafe_b64encode(b'{"v":1,"account":"acct_a","seq":"9"}').rstrip(b"=").decode() ) assert encode_reporting_change_cursor_v1(account="acct_a", seq="9") == js_account + assert is_legacy_account_change_cursor(decode_reporting_change_cursor_v1(js_account, "acct_a")) + # The current account cursor carries the xid of the last change, as the JavaScript SDK writes it. + js_account_xid = ( + base64.urlsafe_b64encode(b'{"v":1,"account":"acct_a","seq":"9","xid":"42"}') + .rstrip(b"=") + .decode() + ) + assert encode_reporting_change_cursor_v1(account="acct_a", seq="9", xid="42") == js_account_xid + decoded_account = decode_reporting_change_cursor_v1(js_account_xid, "acct_a") + assert decoded_account["xid"] == "42" + assert not is_legacy_account_change_cursor(decoded_account) for bad, account in ( ("not base64!", None), (js_account, "acct_b"), @@ -53,6 +72,15 @@ def test_cursors_use_the_format_the_javascript_sdk_reads_and_writes() -> None: (js_deployment, "acct_a"), (base64.urlsafe_b64encode(b'{"v":2,"account":null,"seq":"1","xid":"1"}').decode(), None), (base64.urlsafe_b64encode(b'{"v":1,"account":null,"seq":"x","xid":"1"}').decode(), None), + # A deployment cursor always needs an xid; values the database cast rejects fail here. + (base64.urlsafe_b64encode(b'{"v":1,"account":null,"seq":"1"}').decode(), None), + (b64({"v": 1, "account": "a", "seq": "9223372036854775808"}), "a"), + (b64({"v": 1, "account": "a", "seq": "1", "xid": "18446744073709551616"}), "a"), + (b64({"v": 1, "account": None, "seq": "1", "xid": "99999999999999999999"}), None), + (b64({"v": 1, "account": "a", "seq": ""}), "a"), + (b64({"v": 1, "account": "a", "seq": "1", "xid": "x"}), "a"), + (b64({"v": 1, "account": "a", "seq": "1", "xid": None}), "a"), + (b64({"v": 1, "account": "a", "seq": "\u0663"}), "a"), # a non-ASCII digit ): with pytest.raises(ValueError): decode_reporting_change_cursor_v1(bad, account) @@ -244,6 +272,8 @@ async def test_pages_cross_99_to_100_in_numeric_order_on_both_feeds() -> None: (xid,) = (await fetch_all(pool, "SELECT xid::text FROM adcp_reporting_changes LIMIT 1"))[0] for account, start in ( + ("acct_a", encode_reporting_change_cursor_v1(account="acct_a", seq="94", xid=xid)), + # A seq-only account cursor maps to a position below every later change. ("acct_a", encode_reporting_change_cursor_v1(account="acct_a", seq="94")), (None, encode_reporting_change_cursor_v1(account=None, seq="94", xid=xid)), ): @@ -256,11 +286,12 @@ async def test_pages_cross_99_to_100_in_numeric_order_on_both_feeds() -> None: assert seen == list(range(95, 131)), (account, seen) -# -- account feed ordering, ported from the JavaScript SDK (adcp-client #3228) --------------- +# -- account feed ordering by (xid, seq), ported from the JavaScript SDK (adcp-client #3289) ---- +ACCOUNT = "acct_hol_feed" INSERT_CHANGE = ( "INSERT INTO adcp_reporting_changes (account_id, record_kind, record_id, obligation_id)" - " VALUES ('acct_hol_feed', 'revision', %s, 'robl_hol_feed')" + " VALUES (%s, 'revision', %s, 'robl_hol_feed')" ) @@ -268,16 +299,50 @@ def ids(page: Any) -> list[str]: return [record.record_id for record in page.records] -async def committed(pool: Any, record_id: str) -> None: +async def committed(pool: Any, record_id: str, account: str = ACCOUNT) -> None: async with pool.connection() as connection: - await connection.execute(INSERT_CHANGE, (record_id,)) + await connection.execute(INSERT_CHANGE, (account, record_id)) + + +async def insert_at( + pool: Any, account: str, record_id: str, xid: int, *, age_days: int = 0 +) -> None: + """A committed change with a chosen (historical, so always below the snapshot's xmin) + transaction id, and optionally an old ``recorded_at``, for ordering and pruning cases that + need exact values.""" + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_changes (xid, account_id, record_kind, record_id," + " obligation_id, recorded_at) VALUES (%s::text::xid8, %s, 'revision', %s," + " 'robl_hol_feed', clock_timestamp() - (%s::integer * INTERVAL '1 day'))", + (str(xid), account, record_id, age_days), + ) + + +async def seq_of(pool: Any, record_id: str) -> str: + rows = await fetch_all( + pool, "SELECT seq::text FROM adcp_reporting_changes WHERE record_id = %s", (record_id,) + ) + return str(rows[0][0]) + + +async def remaining(pool: Any, account: str) -> list[str]: + rows = await fetch_all( + pool, + "SELECT record_id FROM adcp_reporting_changes WHERE account_id = %s ORDER BY seq", + (account,), + ) + return [row[0] for row in rows] + + +def legacy_cursor(account: str, seq: str) -> str: + """The cursor an older SDK saved: a ``seq`` and no ``xid``.""" + return encode_reporting_change_cursor_v1(account=account, seq=seq) async def until(read: Any, expected: list[str]) -> Any: """Other tests share the cluster, so an unrelated older transaction can briefly hold a page back: positive reads poll. The assertions that a page is held back do not need it.""" - import asyncio - for _ in range(200): page = await read() if ids(page) == expected: @@ -289,30 +354,30 @@ async def until(read: Any, expected: list[str]) -> Any: async def test_a_lower_seq_that_commits_after_a_higher_one_is_never_passed_over() -> None: async with isolated_reporting_pool() as pool: store = await feed_store(pool) - start = await store.changes_after(account_id="acct_hol_feed") + start = await store.changes_after(account_id=ACCOUNT) slow = await pool.getconn() try: - await slow.execute(INSERT_CHANGE, ("rrev_low",)) # the lower seq, commits last + await slow.execute(INSERT_CHANGE, (ACCOUNT, "rrev_low")) # the lower seq, commits last await committed(pool, "rrev_high") # a higher seq, committed already - during = await store.changes_after(account_id="acct_hol_feed", cursor=start.cursor) + during = await store.changes_after(account_id=ACCOUNT, cursor=start.cursor) assert ids(during) == [], "the page stops before the unfinished transaction" assert during.cursor == start.cursor, "the cursor does not move past it" await slow.commit() finally: await pool.putconn(slow) after = await until( - lambda: store.changes_after(account_id="acct_hol_feed", cursor=start.cursor), + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), ["rrev_low", "rrev_high"], ) assert ids(after) == ["rrev_low", "rrev_high"] -async def test_a_held_back_change_is_never_skipped_when_a_later_one_passes_a_plain_xid_filter() -> ( +async def test_a_lower_xid_with_a_higher_seq_is_returned_first_and_the_held_back_one_after_it() -> ( None ): async with isolated_reporting_pool() as pool: store = await feed_store(pool) - start = await store.changes_after(account_id="acct_hol_feed") + start = await store.changes_after(account_id=ACCOUNT) early = await pool.getconn() blocker = await pool.getconn() try: @@ -320,23 +385,88 @@ async def test_a_held_back_change_is_never_skipped_when_a_later_one_passes_a_pla await early.execute("SELECT pg_current_xact_id()") await blocker.execute("SELECT pg_current_xact_id()") await committed(pool, "rrev_first") # lower seq, but a transaction after `blocker` - await early.execute(INSERT_CHANGE, ("rrev_second",)) # higher seq, xid before `blocker` + await early.execute( + INSERT_CHANGE, (ACCOUNT, "rrev_second") + ) # higher seq, xid < blocker await early.commit() - # `blocker` is the oldest transaction in flight, so rrev_first is held back while - # rrev_second sits below it. Filtering on xid alone would return rrev_second and a - # cursor past rrev_first, which would then never be seen. - during = await store.changes_after(account_id="acct_hol_feed", cursor=start.cursor) - assert ids(during) == [] - assert during.cursor == start.cursor, "the cursor does not move past the held change" + # `blocker` is the oldest transaction in flight, so rrev_first (a later xid) is held + # back while rrev_second sits below it. The feed orders by (xid, seq), so rrev_second + # comes first and the cursor stops at it, before rrev_first. + during = await until( + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), + ["rrev_second"], + ) + assert ids(during) == ["rrev_second"] + assert during.cursor != start.cursor await blocker.commit() + rest = await until( + lambda: store.changes_after(account_id=ACCOUNT, cursor=during.cursor), + ["rrev_first"], + ) + assert ids(rest) == ["rrev_first"], "the held-back change is delivered once, after it" finally: await pool.putconn(early) await pool.putconn(blocker) + everything = await until( + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), + ["rrev_second", "rrev_first"], + ) + assert ids(everything) == ["rrev_second", "rrev_first"], "a fresh reader sees the order" + + +async def test_a_lower_seq_with_a_higher_xid_that_commits_late_is_delivered_after_the_cursor() -> ( + None +): + """No account lock is taken anywhere here: two writers under different lock keys.""" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + start = await store.changes_after(account_id=ACCOUNT) + quick = await pool.getconn() + slow = await pool.getconn() + try: + await quick.execute("SELECT pg_current_xact_id()") # xid Q + await slow.execute("SELECT pg_current_xact_id()") # xid S > Q + await slow.execute(INSERT_CHANGE, (ACCOUNT, "rrev_late")) # lower seq, higher xid + await quick.execute(INSERT_CHANGE, (ACCOUNT, "rrev_quick")) # higher seq, lower xid + await quick.commit() + during = await until( + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), + ["rrev_quick"], + ) + assert ids(during) == ["rrev_quick"], "the late transaction is still open" + await slow.commit() + rest = await until( + lambda: store.changes_after(account_id=ACCOUNT, cursor=during.cursor), + ["rrev_late"], + ) + assert ids(rest) == ["rrev_late"] + assert ( + ids(await store.changes_after(account_id=ACCOUNT, cursor=rest.cursor)) == [] + ), "and nothing is delivered twice" + finally: + await pool.putconn(quick) + await pool.putconn(slow) + + +async def test_an_open_transaction_on_another_account_holds_a_later_change_back() -> None: + other = "acct_hol_feed_other" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + start = await store.changes_after(account_id=ACCOUNT) + opened = await pool.getconn() + try: + await opened.execute(INSERT_CHANGE, (other, "rrev_other_inflight")) # in flight + await committed(pool, "rrev_after_other") + # xmin is cluster-wide, so a later xid is held back whichever account the open one wrote. + assert ids(await store.changes_after(account_id=ACCOUNT, cursor=start.cursor)) == [] + await opened.commit() + finally: + await pool.putconn(opened) after = await until( - lambda: store.changes_after(account_id="acct_hol_feed", cursor=start.cursor), - ["rrev_first", "rrev_second"], + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), + ["rrev_after_other"], ) - assert ids(after) == ["rrev_first", "rrev_second"] + assert ids(after) == ["rrev_after_other"] async def test_a_committed_prefix_is_returned_and_the_cursor_stops_before_the_first_in_flight() -> ( @@ -344,20 +474,20 @@ async def test_a_committed_prefix_is_returned_and_the_cursor_stops_before_the_fi ): async with isolated_reporting_pool() as pool: store = await feed_store(pool) - start = await store.changes_after(account_id="acct_hol_feed") + start = await store.changes_after(account_id=ACCOUNT) await committed(pool, "rrev_prefix_a") opened = await pool.getconn() try: - await opened.execute(INSERT_CHANGE, ("rrev_prefix_b",)) # in flight + await opened.execute(INSERT_CHANGE, (ACCOUNT, "rrev_prefix_b")) # in flight await committed(pool, "rrev_prefix_c") # committed after it during = await until( - lambda: store.changes_after(account_id="acct_hol_feed", cursor=start.cursor), + lambda: store.changes_after(account_id=ACCOUNT, cursor=start.cursor), ["rrev_prefix_a"], ) - assert ids(during) == ["rrev_prefix_a"], "truncated, not filtered" + assert ids(during) == ["rrev_prefix_a"], "c is not handed out past b" await opened.commit() rest = await until( - lambda: store.changes_after(account_id="acct_hol_feed", cursor=during.cursor), + lambda: store.changes_after(account_id=ACCOUNT, cursor=during.cursor), ["rrev_prefix_b", "rrev_prefix_c"], ) assert ids(rest) == ["rrev_prefix_b", "rrev_prefix_c"] @@ -365,23 +495,415 @@ async def test_a_committed_prefix_is_returned_and_the_cursor_stops_before_the_fi await pool.putconn(opened) -async def test_the_account_feed_query_is_served_by_the_account_seq_index() -> None: - from adcp_reporting_storage.ledger.change_feed import _page_query +async def test_the_account_cursor_carries_the_xid_of_the_last_change_and_is_bound_to_its_account() -> ( + None +): + scoped = "acct_hol_feed_cursor" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "rrev_cursor_a", 20) + page = await store.changes_after(account_id=scoped) + decoded = decode_reporting_change_cursor_v1(page.cursor, scoped) + assert decoded["xid"] == "20" + assert not is_legacy_account_change_cursor(decoded) + with pytest.raises(ValueError): + decode_reporting_change_cursor_v1(page.cursor, "acct_hol_feed_elsewhere") + with pytest.raises(ValueError, match="does not belong to this feed"): + await store.changes_after(cursor=page.cursor) + + +async def test_cursor_values_the_database_would_reject_fail_as_value_error() -> None: + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + for body, account in ( + ({"v": 1, "account": "a", "seq": "9223372036854775808"}, "a"), + ({"v": 1, "account": "a", "seq": "1", "xid": "18446744073709551616"}, "a"), + ({"v": 1, "account": None, "seq": "1", "xid": "99999999999999999999"}, None), + ({"v": 1, "account": None, "seq": "1"}, None), + ({"v": 1, "account": "a", "seq": ""}, "a"), + ({"v": 1, "account": "a", "seq": "1", "xid": "x"}, "a"), + ): + with pytest.raises(ValueError): + await store.changes_after(account_id=account, cursor=b64(body)) + decode_reporting_change_cursor_v1( + b64( + { + "v": 1, + "account": "a", + "seq": "9223372036854775807", + "xid": "18446744073709551615", + } + ), + "a", + ) + + +# -- account cursors issued before the (xid, seq) order -------------------------------------- + + +async def test_a_seq_only_cursor_maps_to_a_position_that_skips_nothing_and_upgrades() -> None: + scoped = "acct_hol_legacy_map" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + # r2 took a higher seq but a lower xid than r1: the late commit the old order could skip. + await insert_at(pool, scoped, "r0", 5) # read long ago, older than any unread change + await insert_at(pool, scoped, "r1", 20) + await insert_at(pool, scoped, "r2", 10) + await insert_at(pool, scoped, "r3", 30) + page = await store.changes_after( + account_id=scoped, cursor=legacy_cursor(scoped, await seq_of(pool, "r1")) + ) + # r2 is delivered; r1 is redelivered because its xid is above r2's; r0 is not, so the + # position is not simply xid 0. + assert ids(page) == ["r2", "r1", "r3"] + upgraded = decode_reporting_change_cursor_v1(page.cursor, scoped) + assert upgraded["xid"] == "30" + assert ids(await store.changes_after(account_id=scoped, cursor=page.cursor)) == [] + + # A seq-only cursor at the end of the feed has nothing unread: it maps to the snapshot's + # xmin, and an empty page still upgrades the cursor. + end = await store.changes_after( + account_id=scoped, cursor=legacy_cursor(scoped, await seq_of(pool, "r3")) + ) + assert ids(end) == [] + mapped = decode_reporting_change_cursor_v1(end.cursor, scoped) + assert mapped["seq"] == await seq_of(pool, "r3") + assert int(mapped["xid"]) > 30 + assert not is_legacy_account_change_cursor(mapped) + + +async def test_a_caught_up_seq_only_cursor_does_not_skip_a_change_from_an_in_flight_transaction() -> ( + None +): + scoped = "acct_hol_legacy_inflight" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "caught_up", 10) + head = await seq_of(pool, "caught_up") + opened = await pool.getconn() + try: + await opened.execute(INSERT_CHANGE, (scoped, "inflight")) # at or above xmin + during = await store.changes_after( + account_id=scoped, cursor=legacy_cursor(scoped, head) + ) + assert ids(during) == [] + await opened.commit() + rest = await until( + lambda: store.changes_after(account_id=scoped, cursor=during.cursor), ["inflight"] + ) + assert ids(rest) == ["inflight"], "the mapped position is below the open transaction" + finally: + await pool.putconn(opened) + + +async def test_a_seq_only_cursor_expires_when_any_change_with_a_higher_seq_was_pruned() -> None: + scoped = "acct_hol_legacy_pruned" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + # Pruned: (xid 20, seq s1) and (xid 10, seq s2). The highest (xid, seq) is s1's, but s2 > s1. + await insert_at(pool, scoped, "p1", 20, age_days=40) + await insert_at(pool, scoped, "p2", 10, age_days=40) + await insert_at(pool, scoped, "p3", 30) + s1, s2 = await seq_of(pool, "p1"), await seq_of(pool, "p2") + assert await store.prune_changes(change_retention_days=30) >= 2 + + async def expired(cursor: str) -> bool: + try: + await store.changes_after(account_id=scoped, cursor=cursor) + except ReportingChangeCursorExpiredError as error: + assert error.code == "CURSOR_EXPIRED" + return True + return False + + # p2 was pruned after seq s1: a tuple compare would miss it. + assert await expired(legacy_cursor(scoped, s1)) is True + assert await expired(legacy_cursor(scoped, "0")) is True + assert await expired(legacy_cursor(scoped, s2)) is False + # A cursor at the horizon has passed everything pruned; one below it in (xid, seq) order + # may have missed a pruned change. + at_horizon = encode_reporting_change_cursor_v1(account=scoped, seq=s1, xid="20") + assert await expired(at_horizon) is False + below = encode_reporting_change_cursor_v1(account=scoped, seq=s2, xid="10") + assert await expired(below) is True + assert ids(await store.changes_after(account_id=scoped)) == ["p3"], "no cursor: horizon" + + +async def test_a_seq_only_consumer_registration_holds_by_seq_until_it_saves_a_new_cursor() -> None: + scoped = "acct_hol_legacy_consumer" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "c1", 10, age_days=40) + await insert_at(pool, scoped, "c2", 20, age_days=40) + await insert_at(pool, scoped, "c3", 30, age_days=40) + s1 = await seq_of(pool, "c1") + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_feed_consumers" + " (consumer_name, account_id, cursor_xid, cursor_seq)" + " VALUES ('legacy-sink', %s, NULL, %s::bigint)", + (scoped, s1), + ) + + async def registered() -> tuple[Any, ...]: + return ( + await fetch_all( + pool, + "SELECT cursor_xid::text, cursor_seq::text FROM adcp_reporting_feed_consumers" + " WHERE consumer_name = 'legacy-sink'", + ) + )[0] + + await store.prune_changes(change_retention_days=30) + assert await remaining(pool, scoped) == ["c2", "c3"], "c1 was passed; c2 and c3 are held" + # Saving a seq-only cursor registers a mapped position, below every unread change. + assert await store.save_feed_consumer_cursor("legacy-sink", legacy_cursor(scoped, s1)) + assert (await registered())[0] == "20", "the oldest unread change is c2, at xid 20" + # A cursor in the new format replaces it, forward only. + page = await store.changes_after(account_id=scoped, cursor=legacy_cursor(scoped, s1)) + assert ids(page) == ["c2", "c3"] + assert await store.save_feed_consumer_cursor("legacy-sink", page.cursor) is True + assert await registered() == ("30", await seq_of(pool, "c3")) + + +# -- pruning in (xid, seq) order ------------------------------------------------------------- + + +async def test_an_account_consumer_holds_every_change_after_its_cursor_in_xid_seq_order() -> None: + scoped = "acct_hol_prune_order" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "o1", 10, age_days=40) + await insert_at(pool, scoped, "o2", 30, age_days=40) + await insert_at(pool, scoped, "o3", 20, age_days=40) # higher seq than o2, lower xid + o1, o3 = await seq_of(pool, "o1"), await seq_of(pool, "o3") + cursor = encode_reporting_change_cursor_v1(account=scoped, seq=o3, xid="20") + await store.save_feed_consumer_cursor("prune-order", cursor) + await store.prune_changes(change_retention_days=30) + assert await remaining(pool, scoped) == ["o2"], "o2 follows the cursor in xid order" + horizon = await fetch_all( + pool, + "SELECT pruned_through_xid::text, pruned_through_xid_seq::text," + " pruned_through_seq::text FROM adcp_reporting_change_account_horizon" + " WHERE account_id = %s", + (scoped,), + ) + assert horizon == [("20", o3, o3)] + # The consumer reads on from where it was and is not told it missed anything. + assert ids(await store.changes_after(account_id=scoped, cursor=cursor)) == ["o2"] + # A consumer that had only passed o1 has missed o3 (pruned): expired. + behind = encode_reporting_change_cursor_v1(account=scoped, seq=o1, xid="10") + with pytest.raises(ReportingChangeCursorExpiredError): + await store.changes_after(account_id=scoped, cursor=behind) + + +async def test_the_horizon_only_moves_forward_and_another_accounts_consumer_holds_nothing() -> None: + scoped = "acct_hol_prune_forward" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "f1", 50, age_days=40) + await store.save_feed_consumer_cursor( + "prune-elsewhere", + encode_reporting_change_cursor_v1(account="acct_hol_prune_elsewhere", seq="1", xid="1"), + ) + await store.prune_changes(change_retention_days=30) + assert await remaining(pool, scoped) == [] + + async def horizon_xid() -> str: + rows = await fetch_all( + pool, + "SELECT pruned_through_xid::text FROM adcp_reporting_change_account_horizon" + " WHERE account_id = %s", + (scoped,), + ) + return str(rows[0][0]) + + assert await horizon_xid() == "50" + await insert_at(pool, scoped, "f2", 40, age_days=40) # lower xid than the horizon + await store.prune_changes(change_retention_days=30) + assert await horizon_xid() == "50", "a lower (xid, seq) never lowers the horizon" + + +async def test_a_pruning_batch_cut_by_its_limit_leaves_nothing_retained_below_the_horizon() -> None: + scoped = "acct_hol_prune_batch" + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + await insert_at(pool, scoped, "a", 20, age_days=40) + await insert_at(pool, scoped, "b", 30, age_days=40) + await insert_at(pool, scoped, "c", 10, age_days=40) # the highest seq, the lowest xid + await store.prune_changes(change_retention_days=30, limit=2) + kept = await remaining(pool, scoped) + assert kept == ["b"], "victims are taken in (xid, seq) order: c then a" + assert ( + ids(await store.changes_after(account_id=scoped)) == kept + ), "a cold start loses nothing" + + +async def test_period_retirement_holds_a_period_in_xid_seq_order_for_account_consumers() -> None: + from adcp_reporting_storage.ledger.change_feed import feed_consumer_behind_change + + async with isolated_reporting_pool() as pool: + store = await feed_store(pool) + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_changes (xid, account_id, record_kind, record_id," + " obligation_id) VALUES ('30'::xid8, 'acct_a', 'revision', 'k1', 'o')," + " ('20'::xid8, 'acct_a', 'revision', 'k2', 'o')" + ) + k1, k2 = await seq_of(pool, "k1"), await seq_of(pool, "k2") + sql = ( + "SELECT change.record_id FROM adcp_reporting_changes change" + " JOIN adcp_reporting_feed_consumers consumer ON TRUE WHERE " + + feed_consumer_behind_change("consumer", "change") + + " ORDER BY change.xid, change.seq" + ) + # The consumer passed (xid 20, seq k2): k1 (xid 30, a lower seq) is still ahead of it. + await store.save_feed_consumer_cursor( + "hold", encode_reporting_change_cursor_v1(account="acct_a", seq=k2, xid="20") + ) + assert [row[0] for row in await fetch_all(pool, sql)] == ["k1"] + # A seq-only registration holds by seq alone. + async with pool.connection() as connection: + await connection.execute( + "UPDATE adcp_reporting_feed_consumers SET cursor_xid = NULL, cursor_seq = %s::bigint", + (k1,), + ) + assert [row[0] for row in await fetch_all(pool, sql)] == ["k2"] + + +# -- the migration ----------------------------------------------------------------------------- + +#: The change tables as the previous release created them. +PREVIOUS_CHANGES_DDL = """ +CREATE TABLE adcp_reporting_changes ( + seq BIGSERIAL PRIMARY KEY, xid xid8 NOT NULL DEFAULT pg_current_xact_id(), + account_id TEXT NOT NULL, + record_kind TEXT NOT NULL + CHECK (record_kind IN ('obligation', 'revision', 'adjustment', 'retirement')), + record_id TEXT NOT NULL, obligation_id TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()); +CREATE INDEX adcp_reporting_changes_account ON adcp_reporting_changes (account_id, seq); +CREATE TABLE adcp_reporting_change_account_horizon ( + account_id TEXT PRIMARY KEY, pruned_through_seq BIGINT NOT NULL); +INSERT INTO adcp_reporting_change_account_horizon VALUES ('acct_upgrade', 77); +CREATE TABLE adcp_reporting_feed_consumers ( + consumer_name TEXT PRIMARY KEY CHECK (consumer_name ~ '^[A-Za-z0-9_.:-]{1,128}$'), + account_id TEXT, cursor_xid xid8, cursor_seq BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), + CHECK ((account_id IS NULL) = (cursor_xid IS NOT NULL))); +INSERT INTO adcp_reporting_feed_consumers (consumer_name, account_id, cursor_xid, cursor_seq) + VALUES ('old-account-sink', 'acct_upgrade', NULL, 5); +""" + + +async def test_concurrent_appliers_of_the_changes_migration_all_succeed() -> None: + from adcp_reporting_storage.ledger.change_feed import REPORTING_CHANGES_DDL_PATH async with isolated_reporting_pool() as pool: + await PgReportingLedgerStore(pool=pool).create_schema() # the ledger the index needs + ddl = REPORTING_CHANGES_DDL_PATH.read_text() + + async def apply() -> None: + async with pool.connection() as connection: + await connection.execute(ddl) + + results = await asyncio.gather(*(apply() for _ in range(8)), return_exceptions=True) + assert [r for r in results if isinstance(r, BaseException)] == [] + + +async def test_reapplying_the_changes_migration_upgrades_an_older_install_in_place() -> None: + from adcp_reporting_storage.ledger.change_feed import REPORTING_CHANGES_DDL_PATH + from psycopg.errors import CheckViolation + + async with isolated_reporting_pool() as pool: + await PgReportingLedgerStore(pool=pool).create_schema() + async with pool.connection() as connection: + await connection.execute(PREVIOUS_CHANGES_DDL) + with pytest.raises(CheckViolation): + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_feed_consumers" + " (consumer_name, account_id, cursor_xid, cursor_seq)" + " VALUES ('old-deployment-sink', NULL, NULL, 5)" + ) + ddl = REPORTING_CHANGES_DDL_PATH.read_text() + for _ in range(2): # idempotent + async with pool.connection() as connection: + await connection.execute(ddl) + + assert await fetch_all( + pool, + "SELECT account_id, pruned_through_seq::text, pruned_through_xid::text," + " pruned_through_xid_seq::text FROM adcp_reporting_change_account_horizon", + ) == [("acct_upgrade", "77", "0", "0")], "the existing seq horizon is kept" + assert await fetch_all( + pool, + "SELECT consumer_name, cursor_xid::text, cursor_seq::text FROM" + " adcp_reporting_feed_consumers", + ) == [("old-account-sink", None, "5")], "a seq-only account consumer is kept" + # The deployment feed still needs an xid; the old check is gone, the new one stays. + with pytest.raises(CheckViolation): + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_feed_consumers" + " (consumer_name, account_id, cursor_xid, cursor_seq)" + " VALUES ('bad-deployment', NULL, NULL, 1)" + ) + async with pool.connection() as connection: + await connection.execute( + "INSERT INTO adcp_reporting_feed_consumers" + " (consumer_name, account_id, cursor_xid, cursor_seq)" + " VALUES ('new-account-sink', 'acct_upgrade', '9'::xid8, 1)" + ) + checks = await fetch_all( + pool, + "SELECT conname, pg_get_constraintdef(oid) FROM pg_constraint" + " WHERE conrelid = 'adcp_reporting_feed_consumers'::regclass AND contype = 'c'" + " ORDER BY conname", + ) + assert [name for name, _ in checks if "account_id IS NULL) =" in _] == [] + assert [name for name, _ in checks].count("adcp_reporting_feed_consumers_cursor_scope") == 1 + indexes = { + row[0] + for row in await fetch_all( + pool, + "SELECT indexname FROM pg_indexes WHERE schemaname = current_schema()" + " AND tablename = 'adcp_reporting_changes'", + ) + } + assert "adcp_reporting_changes_account_xid" in indexes + assert "adcp_reporting_changes_account" in indexes, "re-applying never drops an index" + + +async def test_the_account_feed_query_is_served_by_the_account_xid_seq_index_in_order() -> None: + from adcp_reporting_storage.ledger.change_feed import _page_query + + # Autocommit: rows written inside one open transaction carry that transaction's own xid, at or + # above the snapshot's xmin, and the planner would see a histogram nothing satisfies. + async with isolated_reporting_pool(autocommit=True) as pool: await feed_store(pool) async with pool.connection() as connection: await connection.execute( "INSERT INTO adcp_reporting_changes (account_id, record_kind, record_id," - " obligation_id) SELECT 'acct_filler_' || n, 'revision', 'rrev_filler_' || n," - " 'robl_filler' FROM generate_series(1, 2000) AS n" + " obligation_id) SELECT 'acct_filler_' || (n % 2000), 'revision'," + " 'rrev_filler_' || n, 'robl_filler' FROM generate_series(1, 100000) AS n" + ) + # Enough history for the account that a page of 500 is a small slice of it. + await connection.execute( + "INSERT INTO adcp_reporting_changes (account_id, record_kind, record_id," + " obligation_id) SELECT %s, 'revision', 'rrev_history_' || n, 'robl_history'" + " FROM generate_series(1, 3000) AS n", + (ACCOUNT,), ) await connection.execute("ANALYZE adcp_reporting_changes") # The very query the store runs for an account page, planned with its parameters. - plan = await ( - await connection.execute( - "EXPLAIN " + _page_query(account=True), - ("acct_hol_feed", 0, None, None, 500), - ) - ).fetchall() - assert "adcp_reporting_changes_account" in "\n".join(row[0] for row in plan) + plan = "\n".join( + row[0] + for row in await ( + await connection.execute( + "EXPLAIN " + _page_query(account=True), + (ACCOUNT, "0", 0, None, None, 500), + ) + ).fetchall() + ) + assert "adcp_reporting_changes_account_xid" in plan, plan + assert "Sort" not in plan, "the index already yields (xid, seq) order"