Skip to content

Commit 13ca2c5

Browse files
bokelleyclaude
andauthored
feat(reporting): add current-revision host reads (#1473)
* feat(reporting): add current-revision host reads Adds get_current_revision and list_current_revisions (adcontextprotocol/adcp#7996 section 3.7): serving-grade header reads for pacing, alerts and warehouse sinks. The current revision is the official revision if one exists, otherwise the snapshot nothing supersedes yet. list_current_revisions pages obligations by period start with a cursor the JavaScript SDK also reads, and is where a change-feed consumer resynchronizes after CURSOR_EXPIRED. Part of #1457. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> * fix(reporting): validate the current revisions cursor position and index its order Rejects cursors whose start is not a timestamp with a zone before the cast, and adds an obligations index on account, period start and id for hosts that enable the change feed. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Sonnet 5.5 <noreply@anthropic.com>
1 parent b7012e0 commit 13ca2c5

6 files changed

Lines changed: 304 additions & 0 deletions

File tree

‎docs/reporting-row-storage.md‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -328,3 +328,27 @@ await store.changes_after(account_id="acct_a", cursor=page.cursor)
328328
- A registered consumer holds back `prune_changes` until it has passed those changes in its own feed order, up to
329329
`max_feed_hold_days` (default 7) after its last update, so a dead consumer cannot block retention. Schedule
330330
`prune_changes(change_retention_days=30)` next to the other maintenance. Cursors are the same bytes in both SDKs.
331+
332+
## Host reads
333+
334+
In-process consumers such as pacing, alerts and a warehouse sink read the ledger directly. Both reads return revision
335+
headers; rows come from the verified paged reader.
336+
337+
```python
338+
current = await store.get_current_revision(account_id="acct_a", reporting_obligation_id="robl_1")
339+
page = await store.list_current_revisions(
340+
account_id="acct_a", period_start_from=start, period_start_to=end, delivery_config_id="daily", limit=100
341+
)
342+
while page.cursor:
343+
page = await store.list_current_revisions(..., cursor=page.cursor)
344+
```
345+
346+
The current revision is the official revision if one exists, otherwise the snapshot nothing supersedes yet.
347+
`list_current_revisions` pages obligations whose period starts in `[period_start_from, period_start_to)` in period order
348+
and omits obligations without a revision. Its cursor is the format the JavaScript SDK uses. A change-feed consumer
349+
that receives `CURSOR_EXPIRED` resynchronizes from this read.
350+
351+
`list_current_revisions` pages by `(period_start, reporting_obligation_id)`. Enabling `change_feed` creates
352+
`reporting_obligations_account_period_idx` to back that order; a host without it can create the same index itself on
353+
`reporting_obligations (account_id, period_start, reporting_obligation_id)`. A cursor whose position is not a
354+
timestamp with a zone raises `ValueError`, never a driver error.

‎src/adcp/reporting/ledger/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,7 @@
134134
issue_id_for_occurrence,
135135
project_obligation_health,
136136
)
137+
from adcp.reporting.ledger.host_reads import ReportingCurrentRevisionsPageV1
137138
from adcp.reporting.ledger.models import (
138139
ConsumerStatusRecord,
139140
ConsumerStatusValue,
@@ -268,6 +269,7 @@
268269
"ReportingConfigurationGenerationKey",
269270
"ReportingControlTotalRecord",
270271
"ReportingCurrencyError",
272+
"ReportingCurrentRevisionsPageV1",
271273
"ReportingDefinitionBinding",
272274
"ReportingDeliveryEscalation",
273275
"ReportingDeliveryPrincipal",
Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
"""Serving-grade reads for in-process consumers (pacing, alerts, warehouse sinks).
2+
3+
Shared SDK persistence spec, adcontextprotocol/adcp#7996, section 3.7. These return
4+
revision *headers*; rows are read separately through the verified paged reader, and
5+
nothing here reads from a warehouse. ``PgReportingLedgerStore.get_current_revision`` and
6+
``list_current_revisions`` implement them.
7+
"""
8+
9+
from __future__ import annotations
10+
11+
from dataclasses import dataclass
12+
13+
from adcp.reporting.ledger.models import ReportingRevisionRecord
14+
15+
__all__ = ["ReportingCurrentRevisionsPageV1"]
16+
17+
18+
@dataclass(frozen=True, slots=True)
19+
class ReportingCurrentRevisionsPageV1:
20+
"""A page of current revision headers; ``cursor`` is ``None`` on the last page."""
21+
22+
revisions: tuple[ReportingRevisionRecord, ...]
23+
cursor: str | None

‎src/adcp/reporting/ledger/pg.py‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
from __future__ import annotations
7474

7575
import asyncio
76+
import base64
7677
import hashlib
7778
import json
7879
from collections.abc import AsyncIterator, Callable, Sequence
@@ -93,6 +94,7 @@
9394
from adcp.reporting.currency import require_frozen_currency
9495
from adcp.reporting.evidence import ReportingCanonicalDigest, ReportingControlTotalRecord
9596
from adcp.reporting.ledger.health import issue_id_for_occurrence
97+
from adcp.reporting.ledger.host_reads import ReportingCurrentRevisionsPageV1
9698
from adcp.reporting.ledger.models import (
9799
ConsumerStatusRecord,
98100
LedgerRecordKind,
@@ -1664,6 +1666,117 @@ async def get_revision(
16641666
).fetchone()
16651667
return _revision_from_row(row) if row else None
16661668

1669+
async def get_current_revision(
1670+
self, *, account_id: str, reporting_obligation_id: str
1671+
) -> ReportingRevisionRecord | None:
1672+
"""The obligation's current revision header: the official revision if one exists,
1673+
otherwise the snapshot nothing supersedes yet.
1674+
1675+
One indexed lookup. Rows are read separately through the verified paged reader.
1676+
"""
1677+
query = "".join(
1678+
(
1679+
"SELECT ",
1680+
_REVISION_COLUMNS,
1681+
" FROM reporting_revisions r WHERE r.account_id = %s",
1682+
" AND r.reporting_obligation_id = %s AND ",
1683+
_CURRENT_REVISION_FILTER,
1684+
_CURRENT_REVISION_ORDER,
1685+
" LIMIT 1",
1686+
)
1687+
)
1688+
async with self._connection() as connection:
1689+
row = await (
1690+
await connection.execute(query, (account_id, reporting_obligation_id))
1691+
).fetchone()
1692+
return _revision_from_row(row) if row else None
1693+
1694+
async def list_current_revisions(
1695+
self,
1696+
*,
1697+
account_id: str,
1698+
period_start_from: datetime,
1699+
period_start_to: datetime,
1700+
delivery_config_id: str | None = None,
1701+
limit: int = 100,
1702+
cursor: str | None = None,
1703+
) -> ReportingCurrentRevisionsPageV1:
1704+
"""Current revision headers for an account's obligations whose period starts in
1705+
``[period_start_from, period_start_to)``, ordered by period start.
1706+
1707+
Obligations without a revision yet are omitted. Page with ``cursor``; the
1708+
cursor format matches the JavaScript SDK's. A change-feed consumer that
1709+
received ``CURSOR_EXPIRED`` resynchronizes from this read.
1710+
"""
1711+
if type(limit) is not int or not 1 <= limit <= 1000:
1712+
raise ValueError("limit must be between 1 and 1000")
1713+
after: tuple[str, str] | None = None
1714+
if cursor is not None:
1715+
try:
1716+
padded = cursor + "=" * (-len(cursor) % 4)
1717+
parsed = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")))
1718+
start, identity = parsed["start"], parsed["id"]
1719+
if not isinstance(start, str) or not isinstance(identity, str):
1720+
raise ValueError("invalid")
1721+
# A cursor the JavaScript SDK wrote or a corrupted one must fail as a
1722+
# cursor error here, not as a driver error from the timestamp cast.
1723+
moment = datetime.fromisoformat(start.replace("Z", "+00:00"))
1724+
if moment.tzinfo is None:
1725+
raise ValueError("invalid")
1726+
after = (start, identity)
1727+
except (ValueError, KeyError, TypeError, UnicodeError, AttributeError):
1728+
raise ValueError("current revisions cursor is invalid") from None
1729+
query = "".join(
1730+
(
1731+
"SELECT ",
1732+
_OUTER_REVISION_COLUMNS,
1733+
", to_char(o.period_start AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"')",
1734+
" FROM reporting_obligations o JOIN LATERAL (SELECT ",
1735+
_REVISION_COLUMNS,
1736+
" FROM reporting_revisions r WHERE r.account_id = o.account_id",
1737+
" AND r.reporting_obligation_id = o.reporting_obligation_id AND ",
1738+
_CURRENT_REVISION_FILTER,
1739+
_CURRENT_REVISION_ORDER,
1740+
" LIMIT 1) r ON TRUE",
1741+
" WHERE o.account_id = %s AND o.period_start >= %s AND o.period_start < %s",
1742+
" AND (%s::text IS NULL OR o.delivery_config_id = %s)",
1743+
" AND (%s::timestamptz IS NULL OR (o.period_start, o.reporting_obligation_id)",
1744+
" > (%s::timestamptz, %s::text))",
1745+
" ORDER BY o.period_start, o.reporting_obligation_id LIMIT %s",
1746+
)
1747+
)
1748+
async with self._connection() as connection:
1749+
rows = await (
1750+
await connection.execute(
1751+
query,
1752+
(
1753+
account_id,
1754+
period_start_from,
1755+
period_start_to,
1756+
delivery_config_id,
1757+
delivery_config_id,
1758+
after[0] if after else None,
1759+
after[0] if after else None,
1760+
after[1] if after else None,
1761+
limit + 1,
1762+
),
1763+
)
1764+
).fetchall()
1765+
page = rows[:limit]
1766+
next_cursor = None
1767+
if len(rows) > limit:
1768+
last = page[-1]
1769+
next_cursor = (
1770+
base64.urlsafe_b64encode(
1771+
json.dumps({"start": last[-1], "id": last[2]}, separators=(",", ":")).encode()
1772+
)
1773+
.rstrip(b"=")
1774+
.decode("ascii")
1775+
)
1776+
return ReportingCurrentRevisionsPageV1(
1777+
tuple(_revision_from_row(row[:-1]) for row in page), next_cursor
1778+
)
1779+
16671780
async def read_revision_rows(
16681781
self,
16691782
*,
@@ -3064,6 +3177,17 @@ def _translate_integrity_error(error: Exception) -> LedgerConflictError:
30643177
" canonical_content_digest, managed_control_totals"
30653178
)
30663179

3180+
_OUTER_REVISION_COLUMNS = ", ".join(f"r.{name.strip()}" for name in _REVISION_COLUMNS.split(","))
3181+
# The current revision of an obligation is the official one if it exists, otherwise the
3182+
# snapshot nothing supersedes yet. An official revision is terminal, so it is never superseded.
3183+
_CURRENT_REVISION_FILTER = (
3184+
"NOT EXISTS (SELECT 1 FROM reporting_revisions s WHERE s.account_id = r.account_id"
3185+
" AND s.supersedes_reporting_revision_id = r.reporting_revision_id)"
3186+
)
3187+
_CURRENT_REVISION_ORDER = (
3188+
" ORDER BY (r.finality = 'official') DESC, r.created_at DESC, r.reporting_revision_id DESC"
3189+
)
3190+
30673191
_ADJUSTMENT_COLUMNS = (
30683192
"reporting_adjustment_id, account_id, adjusts_reporting_revision_id, reason_code,"
30693193
" reason_detail, accounting_period_start, accounting_period_end, control_total_deltas,"

‎src/adcp/reporting/ledger/reporting_changes.sql‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,12 @@
1818

1919
SELECT pg_advisory_xact_lock(hashtext('adcp.reporting.schema'), hashtext(current_schema()));
2020

21+
-- Backs list_current_revisions: an account's obligations in (period_start, id) order. Kept out of
22+
-- the base ledger DDL, whose parent catalog objects are pinned; extra indexes are ignored by the
23+
-- catalog checks.
24+
CREATE INDEX IF NOT EXISTS reporting_obligations_account_period_idx
25+
ON reporting_obligations (account_id, period_start, reporting_obligation_id);
26+
2127
CREATE TABLE IF NOT EXISTS adcp_reporting_changes (
2228
seq BIGSERIAL PRIMARY KEY,
2329
xid xid8 NOT NULL DEFAULT pg_current_xact_id(),
Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
"""Current-revision reads for in-process consumers: one lookup per obligation, paged by period."""
2+
3+
from __future__ import annotations
4+
5+
import base64
6+
import json
7+
from datetime import timedelta
8+
from typing import Any
9+
10+
import pytest
11+
12+
from adcp.reporting.ledger import PgReportingLedgerStore
13+
14+
from ._generation_support import START, configuration, isolated_reporting_pool, obligation_for
15+
from .test_reporting_row_storage_pg import fetch_all, many_rows, typed_revision
16+
17+
18+
async def obligations(store: PgReportingLedgerStore, count: int) -> list[Any]:
19+
created = []
20+
for index in range(count):
21+
config = configuration("acct_a", consumer_id=f"buyer{index}")
22+
await store.put_configuration(config)
23+
created.append(await store.commit_obligation(obligation_for(config)))
24+
return created
25+
26+
27+
async def test_the_current_revision_is_the_unsuperseded_snapshot_then_the_official_one() -> None:
28+
async with isolated_reporting_pool() as pool:
29+
store = PgReportingLedgerStore(pool=pool)
30+
await store.create_schema()
31+
config = configuration()
32+
await store.put_configuration(config)
33+
obligation = await store.commit_obligation(obligation_for(config))
34+
key = {
35+
"account_id": obligation.account_id,
36+
"reporting_obligation_id": obligation.reporting_obligation_id,
37+
}
38+
assert await store.get_current_revision(**key) is None
39+
40+
rows = many_rows(2)
41+
first = typed_revision(obligation, rows, suffix="a")
42+
await store.commit_revision(first, rows)
43+
assert (
44+
await store.get_current_revision(**key)
45+
).reporting_revision_id == first.reporting_revision_id
46+
second = typed_revision(
47+
obligation, rows, suffix="b", supersedes=first.reporting_revision_id
48+
)
49+
await store.commit_revision(second, rows)
50+
assert (
51+
await store.get_current_revision(**key)
52+
).reporting_revision_id == second.reporting_revision_id
53+
54+
# Scoped to the account: another account's lookup finds nothing.
55+
assert (
56+
await store.get_current_revision(
57+
account_id="acct_other", reporting_obligation_id=key["reporting_obligation_id"]
58+
)
59+
is None
60+
)
61+
62+
63+
async def test_current_revisions_page_by_period_start_with_a_javascript_compatible_cursor() -> None:
64+
async with isolated_reporting_pool() as pool:
65+
store = PgReportingLedgerStore(pool=pool, change_feed=True)
66+
await store.create_schema()
67+
created = await obligations(store, 5)
68+
rows = many_rows(1)
69+
for index, obligation in enumerate(created[:4]): # the fifth has no revision yet
70+
await store.commit_revision(typed_revision(obligation, rows, suffix=f"s{index}"), rows)
71+
72+
window = {
73+
"account_id": "acct_a",
74+
"period_start_from": START - timedelta(days=1),
75+
"period_start_to": START + timedelta(days=1),
76+
}
77+
first = await store.list_current_revisions(**window, limit=3)
78+
assert len(first.revisions) == 3 and first.cursor is not None
79+
# The cursor is the format the JavaScript SDK reads.
80+
padded = first.cursor + "=" * (-len(first.cursor) % 4)
81+
decoded = json.loads(base64.urlsafe_b64decode(padded))
82+
assert set(decoded) == {"start", "id"} and decoded["start"].endswith("Z")
83+
second = await store.list_current_revisions(**window, limit=3, cursor=first.cursor)
84+
assert len(second.revisions) == 1 and second.cursor is None
85+
seen = [r.reporting_revision_id for r in (*first.revisions, *second.revisions)]
86+
assert len(set(seen)) == 4
87+
88+
outside = await store.list_current_revisions(
89+
account_id="acct_a",
90+
period_start_from=START + timedelta(days=1),
91+
period_start_to=START + timedelta(days=2),
92+
)
93+
assert outside.revisions == () and outside.cursor is None
94+
none = await store.list_current_revisions(**{**window, "account_id": "acct_z"})
95+
assert none.revisions == ()
96+
filtered = await store.list_current_revisions(**window, delivery_config_id="nope")
97+
assert filtered.revisions == ()
98+
for bad in (0, 1001):
99+
with pytest.raises(ValueError):
100+
await store.list_current_revisions(**window, limit=bad)
101+
with pytest.raises(ValueError):
102+
await store.list_current_revisions(**window, cursor="!!")
103+
104+
def cursor_of(body: object) -> str:
105+
return base64.urlsafe_b64encode(json.dumps(body).encode()).rstrip(b"=").decode()
106+
107+
# Well-formed JSON that is not a usable position fails as a cursor error, never as a
108+
# driver error from the timestamp cast.
109+
for body in (
110+
{"start": "garbage", "id": "x"},
111+
{"start": "2026-09-01T00:00:00", "id": "x"}, # no zone
112+
{"start": 5, "id": "x"},
113+
{"start": "2026-09-01T00:00:00Z", "id": 7},
114+
{"id": "x"},
115+
["start", "id"],
116+
):
117+
with pytest.raises(ValueError, match="cursor is invalid"):
118+
await store.list_current_revisions(**window, cursor=cursor_of(body))
119+
# Hosts that enable the change feed get an index on the account's period
120+
# starts behind the page order.
121+
assert await fetch_all(
122+
pool,
123+
"SELECT 1 FROM pg_indexes WHERE schemaname = current_schema()"
124+
" AND indexname = 'reporting_obligations_account_period_idx'",
125+
) == [(1,)]

0 commit comments

Comments
 (0)