Read aggregates in one page, and refuse to page them

Paged $apply results from this feed are silently wrong. The row count is right -
it matches the equivalent SQL exactly, 86184 for ssh successes and 91453 for the
(ip,response_status) grouping - but the rows are not: some are duplicated and
others dropped, differently on every run.

    $top=100000 (1 page)   86184 rows, 86184 distinct ip, 0 duplicated
    $top=100000 (again)    86184 rows, 86184 distinct ip - identical
    $top=25000  (4 pages)  86184 rows, 61348 distinct ip, 24836 duplicated
    $top=25000  (again)    86184 rows, 57882 distinct ip, 28302 duplicated
    $top=10000  (9 pages)  86184 rows, 58991 distinct ip

SQL over the same window agrees with the single-page read, so the engine, filter
and aggregate are all correct; each page appears to be a fresh execution of an
unordered query, so $skip lands somewhere different every time.

This had already reached production and left no trace: the derived lists carried
130408 lines for 80048 real http hosts and 123701 for 90221 https, so refresh was
re-grabbing duplicates inside a single run while tens of thousands of known hosts
were missing from both the refresh queue and the discovery exclusion.

$top now sits at the server's ceiling, where every query this project issues fits
in one page, and grouped_max/distinct_values raise rather than follow a nextLink.
The failure mode to design against is a plausible answer, not an error -
responsive.py keeps the previous list when a read raises and cannot tell a corrupt
answer from a good one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: joocer

src/ichnos/odata.py

@@ -46,10 +46,32 @@
 DEFAULT_ODATA_PREFIX = "/api/v4"
 DEFAULT_TOKEN_URL = "https://authenticate.opteryx.app/token"
 
-MAX_TOP = 25000
-"""The service's documented ceiling for a single request's `$top`; a larger value is
-rejected with a 400. Set deliberately high rather than left to the default of 100 -
-this module's reads are bulk reads, and the default would turn one page into 250."""
+MAX_TOP = 100000
+"""The service's ceiling for a single request's `$top` - 100001 is rejected with
+"$top must be between 0 and 100000". Set to exactly the ceiling, and that is not just
+about efficiency: a read that fits in one page is the only read this feed answers
+correctly.
+
+Paginated `$apply` results are unstable. The row *count* is right and matches the
+equivalent SQL exactly, but the contents are not: rows are duplicated and others
+dropped, differently on every run. Measured against ssh observations, one query,
+same data, minutes apart:
+
+    $top=100000 (1 page)    86184 rows, 86184 distinct ip, 0 duplicated
+    $top=100000 (again)     86184 rows, 86184 distinct ip, 0 duplicated - identical
+    $top=25000  (4 pages)   86184 rows, 61348 distinct ip, 24836 duplicated
+    $top=25000  (again)     86184 rows, 57882 distinct ip, 28302 duplicated
+    $top=10000  (9 pages)   86184 rows, 58991 distinct ip
+
+SQL over the same window returns 86184 rows and 86184 distinct ip, so the engine, the
+filter and the aggregate are all correct - each page appears to be a fresh execution of
+an unordered query, so `$skip` lands somewhere different every time.
+
+The damage was silent. Nothing errors, the row count looks right, and the derived
+known-responsive lists simply carried duplicates and were missing hosts: 130408 lines
+for 80048 real http hosts, 123701 for 90221 https. Hence `_reject_paged_result` below -
+at 100000 nothing this project reads pages today, and the day something does, it must
+fail loudly rather than quietly produce a plausible wrong answer."""
 
 
 class ODataError(Exception):
@@ -130,10 +152,15 @@ def iter_rows(
     base_url: str = DEFAULT_ODATA_BASE,
     prefix: str = DEFAULT_ODATA_PREFIX,
     get: Optional[Callable[..., Any]] = None,
+    single_page: bool = False,
 ) -> Iterator[Dict[str, Any]]:
     """Yield every row of a query, following `@odata.nextLink` until the feed stops
     offering one.
 
+    `single_page=True` refuses to follow the link at all and raises instead, for callers
+    whose answer would be silently wrong if it did - see MAX_TOP for the measurements.
+    Paging this feed returns the right number of rows with the wrong rows in them.
+
     `path` is the three-part `{workspace}/{collection}/{dataset}` address - the same
     triple the Upload API's `Target` uses. `query` is a pre-encoded query string; it is
     not built from a dict on purpose, see the module docstring.
@@ -157,6 +184,13 @@ def iter_rows(
         next_link = payload.get("@odata.nextLink")
         if not next_link:
             break
+        if single_page:
+            raise ODataError(
+                f"{path}: result needs more than one page at $top={MAX_TOP}, and paged "
+                "reads from this feed are not trustworthy - the row count is right but "
+                "rows are duplicated and dropped non-deterministically (see MAX_TOP). "
+                "Narrow the query or fix the feed; do not page it."
+            )
         # Relative, and already percent-encoded - concatenate, never re-encode.
         url = next_link if next_link.startswith("http") else f"{base_url}{next_link}"
 
@@ -198,7 +232,8 @@ def grouped_max(
     query = f"$apply={quote(apply_expr, safe='()/,')}&$top={top}"
     return [
         row
-        for row in iter_rows(path, query, token=token, base_url=base_url, prefix=prefix, get=get)
+        for row in iter_rows(path, query, token=token, base_url=base_url, prefix=prefix,
+                             get=get, single_page=True)
         if all(row.get(c) is not None for c in columns)
     ]
 
@@ -230,7 +265,8 @@ def distinct_values(
 
     values = []
     for row in iter_rows(
-        path, query, token=token, base_url=base_url, prefix=prefix, get=get
+        path, query, token=token, base_url=base_url, prefix=prefix, get=get,
+        single_page=True,
     ):
         value = row.get(column)
         if value is not None:

tests/test_responsive.py

@@ -6,6 +6,8 @@
 from ichnos.odata import ODataError
 from ichnos.odata import distinct_values
 from ichnos.odata import fetch_access_token
+from ichnos.odata import grouped_max
+from ichnos.odata import iter_rows
 from ichnos.responsive import fetch_responsive_hosts
 from ichnos.responsive import read_responsive_file
 from ichnos.responsive import read_responsive_hosts
@@ -49,7 +51,7 @@ def test_distinct_values_composes_the_filter_inside_apply():
     )
     assert "%24apply" not in calls[0]  # we build it, we don't re-encode it
     assert "$apply=filter(protocol%20eq%20%27http%27)/groupby((ip))" in calls[0]
-    assert "$top=25000" in calls[0]
+    assert "$top=100000" in calls[0]
 
 
 def test_iter_rows_follows_nextlink_without_re_encoding_it():
@@ -60,7 +62,8 @@ def test_iter_rows_follows_nextlink_without_re_encoding_it():
         {"value": [{"ip": "1.1.1.1"}], "@odata.nextLink": "/api/v4/ws/coll/observations?%24skip=1"},
         {"value": [{"ip": "2.2.2.2"}]},
     ]
-    ips = distinct_values("ws/coll/observations", "ip", get=_fake_get(pages, calls))
+    ips = [r["ip"] for r in iter_rows("ws/coll/observations", "$top=1",
+                                      get=_fake_get(pages, calls))]
     assert ips == ["1.1.1.1", "2.2.2.2"]
     assert calls[1] == "https://odata.opteryx.app/api/v4/ws/coll/observations?%24skip=1"
 
@@ -78,7 +81,7 @@ def get(url, headers=None, timeout=None):
         return FakeResponse(pages[0])
 
     with pytest.raises(ODataError) as exc:
-        distinct_values("ws/coll/observations", "ip", get=get)
+        list(iter_rows("ws/coll/observations", "$top=1", get=get))
     assert "500" in str(exc.value)
 
 
@@ -254,3 +257,44 @@ def test_a_host_that_only_ever_failed_is_not_a_refresh_target():
     )
 
     assert [ip for ip, _ in hosts] == ["real"]
+
+
+def test_an_aggregate_read_refuses_to_paginate():
+    """Paged `$apply` results from this feed are silently wrong. The row count is right
+    and matches the equivalent SQL exactly, but the contents are not - rows are
+    duplicated and others dropped, differently on every run. Measured against ssh
+    observations, same query, same data, minutes apart:
+
+        $top=100000 (1 page)   86184 rows, 86184 distinct ip, 0 duplicated
+        $top=100000 (again)    86184 rows, 86184 distinct ip - byte-identical
+        $top=25000  (4 pages)  86184 rows, 61348 distinct ip, 24836 duplicated
+        $top=25000  (again)    86184 rows, 57882 distinct ip, 28302 duplicated
+
+    So the failure mode to design against is not an error, it is a plausible answer -
+    and it had already reached production, where the derived lists carried 130408 lines
+    for 80048 real http hosts. Nothing this project reads needs a second page at
+    $top=100000, and the day something does it must stop: responsive.py keeps the
+    previous list when a read raises, and has no way to tell a corrupt answer from a
+    good one."""
+    paged = [
+        {"value": [{"ip": "203.0.113.1", "last_at": "2026-08-01T00:00:00Z"}],
+         "@odata.nextLink": "/api/v4/ws/coll/observations?%24skip=1"},
+        {"value": [{"ip": "203.0.113.2", "last_at": "2026-08-02T00:00:00Z"}]},
+    ]
+
+    with pytest.raises(ODataError, match="more than one page"):
+        grouped_max("ws/coll/observations", "ip", "observed_at", "last_at",
+                    get=_fake_get(paged))
+
+
+def test_a_single_page_aggregate_read_is_returned_normally():
+    """The guard must not fire on the normal case - every real query fits today."""
+    rows = grouped_max(
+        "ws/coll/observations", ("ip", "response_status"), "observed_at", "last_at",
+        get=_fake_get([{"value": [
+            {"ip": "203.0.113.1", "response_status": "success",
+             "last_at": "2026-08-01T00:00:00Z"},
+        ]}]),
+    )
+    assert rows == [{"ip": "203.0.113.1", "response_status": "success",
+                     "last_at": "2026-08-01T00:00:00Z"}]