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"