diff --git a/services/graph/app/fsma/chain_integrity.py b/services/graph/app/fsma/chain_integrity.py index 93e8347de..3c0de5986 100644 --- a/services/graph/app/fsma/chain_integrity.py +++ b/services/graph/app/fsma/chain_integrity.py @@ -8,7 +8,7 @@ from __future__ import annotations import time -from typing import Any, Dict, List, Optional +from typing import Any, AsyncIterator, Callable, Dict, List, Optional, Sequence import structlog from shared.fsma_rules import TimeArrowRule, TraceEvent as SharedTraceEvent @@ -18,6 +18,99 @@ logger = structlog.get_logger("fsma-utils") +# Rows fetched per round trip. Results are read page by page with keyset +# pagination on a unique key, so every matching row is reported, nothing is +# silently dropped at a fixed cap, and each round trip stays bounded. Unlike +# SKIP/offset paging, a row deleted (or inserted) between pages cannot shift +# the next page past a row that has not been read yet: each page starts +# strictly after the last key already returned. +_PAGE_SIZE = 2000 + + +def _sort_key(*fields: str) -> Callable[[Any], tuple]: + """Python sort key for records: ascending, nulls last (Neo4j ASC order).""" + + def key(record: Any) -> tuple: + return tuple((record[f] is None, "" if record[f] is None else str(record[f])) for f in fields) + + return key + + +def _after_condition(width: int) -> str: + """Cypher for ``_page_key > $after_key``, compared element by element.""" + terms = [] + for i in range(width): + equal = [f"_page_key[{j}] = $after_key[{j}]" for j in range(i)] + terms.append("(" + " AND ".join(equal + [f"_page_key[{i}] > $after_key[{i}]"]) + ")") + return " OR ".join(terms) + + +_PAGE_MARKER = "// @page" + + +def _hit(record: Any) -> bool: + """True unless the row carries a ``_hit`` flag that is not true. + + A null flag (Cypher ``null``) is not a hit; queries coalesce it to false. + """ + try: + return record["_hit"] is True + except (KeyError, LookupError): + return True + + +async def _run_paged( + session: Any, + query: str, + *, + anchors: str, + key: Sequence[str], + order: Optional[Callable[[Any], Any]] = None, + **params: Any, +) -> AsyncIterator[Any]: + """Yield every record of ``query``, fetched ``_PAGE_SIZE`` anchors at a time. + + ``query`` holds a ``// @page`` line right after the MATCH that binds the + ``anchors`` variables. The keyset filter, ORDER BY and LIMIT are put + there, so each page reads only its own anchors instead of the whole match. + ``key`` lists Cypher expressions over the anchors that are unique + together. The rest of the query must carry ``_page_key`` along and return + exactly one row per anchor; a row whose ``_hit`` is false is used only to + advance the cursor and is not yielded. Records are yielded once all pages + are read, sorted by ``order`` when given. + """ + if _PAGE_MARKER not in query: + raise ValueError("paged query needs a // @page marker") + key_expr = ", ".join(f"coalesce(toString({c}), '')" for c in key) + page_clause = ( + f"WITH {anchors}, [{key_expr}] AS _page_key\n" + f"WHERE $after_key IS NULL OR {_after_condition(len(key))}\n" + f"WITH {anchors}, _page_key\n" + "ORDER BY _page_key[0]" + "".join(f", _page_key[{i}]" for i in range(1, len(key))) + "\nLIMIT $page_size" + ) + paged_query = query.replace(_PAGE_MARKER, page_clause, 1) + records: List[Any] = [] + after_key: Optional[List[str]] = None + while True: + result = await session.run(paged_query, after_key=after_key, page_size=_PAGE_SIZE, **params) + page = [record async for record in result] + records.extend(r for r in page if _hit(r)) + if len(page) < _PAGE_SIZE: + break + after_key = list(page[-1]["_page_key"]) + if order is not None: + records.sort(key=order) + for record in records: + yield record + + +# Broken-chain rows are (shipping event, lot) pairs: unique together. They +# are reported in event-date order, then event id, then lot. +_CHAIN_ANCHORS = "l, shipping" +_CHAIN_KEY = ("shipping.event_id", "l.tlc") +_CHAIN_ORDER = _sort_key("event_date", "event_id", "lot_tlc") + + async def find_gaps( client: Neo4jClient, tenant_id: str, @@ -37,9 +130,7 @@ async def find_gaps( callers read every tenant's events, so callers must pass a tenant. """ if not tenant_id or not str(tenant_id).strip(): - raise ValueError( - "find_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported" - ) + raise ValueError("find_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported") start_time = time.time() @@ -57,7 +148,9 @@ async def find_gaps( e.responsible_party_contact IS NULL OR e.responsible_party_contact = '' ) + // @page RETURN + _page_key, e.event_id as event_id, e.type as type, e.event_date as event_date, @@ -65,13 +158,16 @@ async def find_gaps( CASE WHEN e.event_date IS NULL OR e.event_date = '' THEN 'missing_date' ELSE '' END + CASE WHEN NOT EXISTS { MATCH (e)<-[:UNDERWENT]-(l:Lot) WHERE l.tenant_id = $tenant_id } THEN ',missing_lot' ELSE '' END + CASE WHEN e.responsible_party_contact IS NULL OR e.responsible_party_contact = '' THEN ',missing_responsible_party_contact' ELSE '' END as gaps - LIMIT 2000 """ gaps = [] + # Keep the first occurrence of each event so no event is reported twice. + seen_event_ids: set = set() async with client.session() as session: - result = await session.run(query, tenant_id=tenant_id) - async for record in result: + async for record in _run_paged(session, query, anchors="e", key=("e.event_id",), tenant_id=tenant_id): + if record["event_id"] in seen_event_ids: + continue + seen_event_ids.add(record["event_id"]) gaps.append( { "event_id": record["event_id"], @@ -140,7 +236,8 @@ async def find_broken_chains( # ------------------------------------------------------------------ # # Query 1: SHIPPING with no origin event at all # Optimization: single-hop UNDERWENT only (no variable-length path), - # LIMIT 2000 caps result set on large tenants for sub-second response. + # Results are keyset-paged (_run_paged) instead of capped, so large tenants + # get every violation. # ------------------------------------------------------------------ # # #1301: Every event node joined via UNDERWENT must be tenant-scoped, # not just the Lot. A cross-tenant UNDERWENT edge (possible because of @@ -158,15 +255,15 @@ async def find_broken_chains( WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING', 'TRANSFORMATION'] AND origin.tenant_id = $tenant_id } + // @page RETURN + _page_key, shipping.event_id AS event_id, shipping.type AS type, shipping.event_date AS event_date, shipping.document_id AS document_id, l.tlc AS lot_tlc, l.product_description AS product_description - ORDER BY shipping.event_date - LIMIT 2000 """ # ------------------------------------------------------------------ # @@ -183,8 +280,9 @@ async def find_broken_chains( WHERE l.tenant_id = $tenant_id AND shipping.tenant_id = $tenant_id AND shipping.event_date IS NOT NULL - WITH l, shipping - MATCH (l)-[:UNDERWENT]->(origin:TraceEvent) + // @page + // OPTIONAL so every paged (lot, shipping) anchor yields one row. + OPTIONAL MATCH (l)-[:UNDERWENT]->(origin:TraceEvent) WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING', 'TRANSFORMATION'] AND origin.tenant_id = $tenant_id AND origin.event_date IS NOT NULL @@ -192,12 +290,13 @@ async def find_broken_chains( // encounter order, so head(collect(...)) is the id of the earliest // origin matching min(origin.event_date). Without this the reported // origin_event_id can point at a different origin than origin_date. - WITH l, shipping, origin ORDER BY origin.event_date ASC - WITH l, shipping, + WITH l, shipping, _page_key, origin ORDER BY origin.event_date ASC + WITH l, shipping, _page_key, min(origin.event_date) AS earliest_origin_date, head(collect(origin.event_id)) AS earliest_origin_id - WHERE earliest_origin_date > shipping.event_date RETURN + _page_key, + coalesce(earliest_origin_date > shipping.event_date, false) AS _hit, shipping.event_id AS event_id, shipping.type AS type, shipping.event_date AS event_date, @@ -206,8 +305,6 @@ async def find_broken_chains( l.product_description AS product_description, earliest_origin_id AS origin_event_id, earliest_origin_date AS origin_date - ORDER BY shipping.event_date - LIMIT 2000 """ violations: List[Dict[str, Any]] = [] @@ -215,8 +312,14 @@ async def find_broken_chains( async with client.session() as session: # --- Missing origin violations -------------------------------- - result = await session.run(missing_origin_query, tenant_id=tenant_id) - async for record in result: + async for record in _run_paged( + session, + missing_origin_query, + anchors=_CHAIN_ANCHORS, + key=_CHAIN_KEY, + order=_CHAIN_ORDER, + tenant_id=tenant_id, + ): eid = record["event_id"] if eid in seen_event_ids: continue @@ -236,8 +339,14 @@ async def find_broken_chains( ) # --- Temporal paradox violations ------------------------------ - result = await session.run(temporal_paradox_query, tenant_id=tenant_id) - async for record in result: + async for record in _run_paged( + session, + temporal_paradox_query, + anchors=_CHAIN_ANCHORS, + key=_CHAIN_KEY, + order=_CHAIN_ORDER, + tenant_id=tenant_id, + ): eid = record["event_id"] if eid in seen_event_ids: continue @@ -308,20 +417,21 @@ async def find_broken_chains( WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING'] AND origin.tenant_id = $tenant_id } - WITH l, shipping + // @page MATCH (l)-[:UNDERWENT]->(origin:TraceEvent) WHERE origin.type IN ['CREATION', 'RECEIVING', 'INITIAL_PACKING'] AND origin.tenant_id = $tenant_id - WITH shipping, l, + WITH shipping, l, _page_key, collect({ event_id: origin.event_id, type: origin.type, event_date: origin.event_date, merkle_hash: origin.merkle_hash }) AS origins - WHERE shipping.merkle_prev_hash IS NULL - OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash) RETURN + _page_key, + coalesce(shipping.merkle_prev_hash IS NULL + OR NOT any(o IN origins WHERE o.merkle_hash = shipping.merkle_prev_hash), false) AS _hit, shipping.event_id AS event_id, shipping.type AS type, shipping.event_date AS event_date, @@ -331,12 +441,11 @@ async def find_broken_chains( l.tlc AS lot_tlc, l.product_description AS product_description, origins AS expected_predecessors - ORDER BY shipping.event_date - LIMIT 2000 """ - result = await session.run(crypto_gap_query, tenant_id=tenant_id) - async for record in result: + async for record in _run_paged( + session, crypto_gap_query, anchors=_CHAIN_ANCHORS, key=_CHAIN_KEY, order=_CHAIN_ORDER, tenant_id=tenant_id + ): eid = record["event_id"] if eid in seen_event_ids: continue @@ -392,9 +501,7 @@ async def find_all_gaps( the underlying gap queries, which require a tenant). """ if not tenant_id or not str(tenant_id).strip(): - raise ValueError( - "find_all_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported" - ) + raise ValueError("find_all_gaps requires a non-empty tenant_id; null-tenant queries are no longer supported") gaps = await find_gaps(client, tenant_id) broken_chains = await find_broken_chains(client, tenant_id) diff --git a/services/graph/app/fsma/data_quality.py b/services/graph/app/fsma/data_quality.py index c50b5e087..4a00e1c48 100644 --- a/services/graph/app/fsma/data_quality.py +++ b/services/graph/app/fsma/data_quality.py @@ -14,6 +14,7 @@ import structlog from ..neo4j_utils import Neo4jClient +from .chain_integrity import _run_paged, _sort_key from .types import DataQualityReport, KDECompletenessMetrics, OrphanLot logger = structlog.get_logger("fsma-utils") @@ -128,20 +129,21 @@ async def find_orphaned_lots( MATCH (trans:TraceEvent)-[:CONSUMED]->(l) WHERE trans.tenant_id = $tenant_id } + // @page // Get the most recent event for this lot OPTIONAL MATCH (l)-[:UNDERWENT]->(last_event:TraceEvent) WHERE last_event.tenant_id = $tenant_id - WITH l, last_event + WITH l, _page_key, last_event ORDER BY last_event.event_date DESC - WITH l, collect(last_event)[0] as most_recent_event - - // Filter by stagnant days - WHERE most_recent_event IS NULL - OR most_recent_event.event_date IS NULL - OR most_recent_event.event_date <= $cutoff_date + WITH l, _page_key, collect(last_event)[0] as most_recent_event + // Stagnant-days filter, as a flag so every paged lot yields one row RETURN + _page_key, + coalesce(most_recent_event IS NULL + OR most_recent_event.event_date IS NULL + OR most_recent_event.event_date <= $cutoff_date, false) AS _hit, l.tlc as tlc, l.product_description as product_description, l.quantity as quantity, @@ -149,15 +151,24 @@ async def find_orphaned_lots( l.created_at as created_at, most_recent_event.type as last_event_type, most_recent_event.event_date as last_event_date - ORDER BY most_recent_event.event_date ASC - LIMIT 2000 """ orphans = [] + # Keep the first occurrence of each lot so no lot is reported twice. + seen_tlcs: set = set() async with client.session() as session: - result = await session.run(query, tenant_id=tenant_id, cutoff_date=cutoff_date) - - async for record in result: + async for record in _run_paged( + session, + query, + anchors="l", + key=("l.tlc",), + order=_sort_key("last_event_date", "tlc"), + tenant_id=tenant_id, + cutoff_date=cutoff_date, + ): + if record["tlc"] in seen_tlcs: + continue + seen_tlcs.add(record["tlc"]) # Calculate stagnant days. The parsed date is naive, so anchor # it to UTC before subtracting: an aware minus naive # subtraction raises TypeError, which the fallback below would diff --git a/services/graph/tests/test_data_quality.py b/services/graph/tests/test_data_quality.py index cbfd8fc60..0a4cc69cf 100644 --- a/services/graph/tests/test_data_quality.py +++ b/services/graph/tests/test_data_quality.py @@ -302,6 +302,7 @@ async def test_orphan_detection_with_results(self): "created_at": "2025-01-01", "last_event_type": "RECEIVING", "last_event_date": "2025-01-05", + "_hit": True, # the query flags reported lots; null is not a hit }.get(key) # Mock async iterator that yields one record diff --git a/services/graph/tests/test_fsma_module_qa.py b/services/graph/tests/test_fsma_module_qa.py index 64f120d83..b3f5cfdab 100644 --- a/services/graph/tests/test_fsma_module_qa.py +++ b/services/graph/tests/test_fsma_module_qa.py @@ -232,14 +232,18 @@ async def test_exists_check_scopes_lot(self): class TestBrokenChainsQueries: @pytest.mark.asyncio - async def test_crypto_gap_query_has_row_cap(self): - """All three broken-chain queries must cap rows for large tenants.""" + async def test_broken_chain_queries_are_paged_not_capped(self): + """R1: each broken-chain query fetches bounded pages, never a silent 2000-row cap.""" client, session = _empty_client() await find_broken_chains(client, tenant_id=TENANT) assert session.run.call_count == 3 for i, call in enumerate(session.run.call_args_list): cypher = _flatten(call.args[0]) - assert "LIMIT 2000" in cypher, f"query {i} has no row cap" + assert "LIMIT 2000" not in cypher, f"query {i} still has a fixed row cap" + assert "LIMIT $page_size" in cypher, f"query {i} is not paged" + assert "SKIP" not in cypher, f"query {i} uses offset paging" + assert "$after_key" in cypher, f"query {i} is not keyset-paged" + assert call.kwargs["page_size"] > 0 @pytest.mark.asyncio async def test_paradox_query_sorts_origins_before_collect(self): diff --git a/services/graph/tests/test_graph_row_paging_r1.py b/services/graph/tests/test_graph_row_paging_r1.py new file mode 100644 index 000000000..27be5db66 --- /dev/null +++ b/services/graph/tests/test_graph_row_paging_r1.py @@ -0,0 +1,385 @@ +"""R1: chain_integrity / data_quality must not silently cap results at 2000 rows. + +Rows are fully synthetic: every id is a fresh ``uuid4`` and business fields +are empty or ``None``. Where a sortable field needs a value, the row index is +used. The fake Neo4j session honours ``$after_key`` / ``$page_size`` so the +tests exercise real keyset paging. +""" + +from __future__ import annotations + +import re +import sys +import uuid +from collections import Counter +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +sys.path.insert(0, str(Path(__file__).resolve().parents[3])) + +from services.graph.app.fsma import chain_integrity +from services.graph.app.fsma.chain_integrity import find_broken_chains, find_gaps +from services.graph.app.fsma.data_quality import find_orphaned_lots + +TENANT = str(uuid.uuid4()) +TOTAL_ROWS = 4501 # > 2 full pages of 2000, plus a partial page + + +class _Result: + def __init__(self, rows): + self._rows = rows + + def __aiter__(self): + return self._gen() + + async def _gen(self): + for r in self._rows: + yield r + + +class _PagingSession: + """Fake session: routes by query text and serves keyset pages. + + The page key columns are read from the generated Cypher; each page holds + the rows whose key is strictly after ``$after_key``, sorted by key. + ``on_page(n, rows)`` runs after page ``n`` is served, so a test can + delete rows between pages. ``dup_in_page(n, page)`` may return a changed + page ``n`` (for example one row served twice) before it is sent. + """ + + def __init__(self, rows_for, on_page=None, dup_in_page=None): + self._rows_for = rows_for + self._on_page = on_page + self._dup_in_page = dup_in_page + self.calls: list[tuple[str, dict]] = [] + + async def run(self, query, **params): + self.calls.append((query, params)) + assert "LIMIT 2000" not in query + assert "SKIP" not in query + assert "LIMIT $page_size" in query + assert "CALL {" not in query + assert "// @page" not in query + exprs = re.findall(r"coalesce\(toString\((\w+\.\w+)\), ''\)", query) + assert exprs, query + rows = self._rows_for(query) + # The cursor filter runs on anchor properties (``l.tlc``); map each to + # the returned column (``tlc``, or ``lot_tlc`` for chain rows). + cols = [] + for expr in exprs: + col = expr.split(".")[1] + if rows and col not in rows[0]: + col = f"lot_{col}" + cols.append(col) + + def key(r): + return ["" if r[c] is None else str(r[c]) for c in cols] + + after = params["after_key"] + page = sorted((r for r in rows if after is None or key(r) > after), key=key) + page = [{**r, "_page_key": key(r)} for r in page[: params["page_size"]]] + if self._dup_in_page: + page = self._dup_in_page(len(self.calls), page) + if self._on_page: + self._on_page(len(self.calls), rows) + return _Result(page) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +def _client(session): + client = MagicMock() + client.session = MagicMock(return_value=session) + return client + + +def _assert_each_once(input_ids: list[str], output_ids: list[str]) -> None: + counts = Counter(output_ids) + assert set(counts) == set(input_ids) + assert all(n == 1 for n in counts.values()) + + +def _shipping_rows(n: int) -> list[dict]: + return [ + { + "event_id": str(uuid.uuid4()), + "type": "SHIPPING", + "event_date": f"{i:05d}", # sortable: row index + "document_id": None, + "lot_tlc": str(uuid.uuid4()), + "product_description": "", + "merkle_hash": None, + "merkle_prev_hash": None, + "expected_predecessors": [], + "gaps": "", + } + for i in range(n) + ] + + +def test_page_size_is_positive(): + assert chain_integrity._PAGE_SIZE > 0 + + +@pytest.mark.asyncio +async def test_find_gaps_returns_more_than_2000_rows(): + rows = _shipping_rows(TOTAL_ROWS) + session = _PagingSession(lambda q: rows) + gaps = await find_gaps(_client(session), tenant_id=TENANT) + assert len(gaps) == TOTAL_ROWS + _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps]) + assert len(session.calls) == 3 + assert session.calls[0][1]["after_key"] is None + assert all(c[1]["page_size"] == 2000 for c in session.calls) + assert all(c[1]["tenant_id"] == TENANT for c in session.calls) + + +@pytest.mark.asyncio +async def test_find_gaps_drops_rows_repeated_across_pages(): + rows = _shipping_rows(TOTAL_ROWS) + # Simulate a row shifting forward between pages: page 2 repeats the + # last row of page 1. + shifted = rows[:2000] + [rows[1999]] + rows[2000:] + session = _PagingSession(lambda q: shifted) + gaps = await find_gaps(_client(session), tenant_id=TENANT) + _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps]) + + +@pytest.mark.asyncio +async def test_find_broken_chains_returns_more_than_2000_rows(): + rows = _shipping_rows(TOTAL_ROWS) + + def rows_for(q): + # missing_origin query only; temporal_paradox and crypto_gap empty. + if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q: + return rows + return [] + + session = _PagingSession(rows_for) + violations = await find_broken_chains(_client(session), tenant_id=TENANT) + assert len(violations) == TOTAL_ROWS + _assert_each_once([r["lot_tlc"] for r in rows], [v["lot_tlc"] for v in violations]) + # Reported in event-date order (the row index here). + assert [v["event_date"] for v in violations] == [f"{i:05d}" for i in range(TOTAL_ROWS)] + + +def _orphan_rows(n: int) -> list[dict]: + return [ + { + "tlc": str(uuid.uuid4()), + "product_description": "", + "quantity": None, + "unit_of_measure": "", + "created_at": None, + "last_event_type": "", + "last_event_date": None, + } + for _ in range(n) + ] + + +@pytest.mark.asyncio +async def test_find_orphaned_lots_returns_more_than_2000_rows(): + rows = _orphan_rows(TOTAL_ROWS) + session = _PagingSession(lambda q: rows) + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + assert len(orphans) == TOTAL_ROWS + _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans]) + assert len(session.calls) == 3 + assert [o.tlc for o in orphans] == sorted(r["tlc"] for r in rows) + + +@pytest.mark.asyncio +async def test_find_orphaned_lots_drops_rows_repeated_across_pages(): + rows = _orphan_rows(TOTAL_ROWS) + shifted = rows[:2000] + [rows[1999]] + rows[2000:] + session = _PagingSession(lambda q: shifted) + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans]) + + +@pytest.mark.asyncio +async def test_exact_page_boundary_issues_one_empty_followup(): + rows = _shipping_rows(2000) + session = _PagingSession(lambda q: rows) + gaps = await find_gaps(_client(session), tenant_id=TENANT) + assert len(gaps) == 2000 + _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps]) + assert len(session.calls) == 2 + + +def _delete_after_first_page(victims): + def on_page(n, rows): + if n == 1: + for v in victims: + rows.remove(v) + + return on_page + + +@pytest.mark.asyncio +async def test_find_gaps_deletion_between_pages_skips_nothing(): + rows = _shipping_rows(TOTAL_ROWS) + ordered = sorted(rows, key=lambda r: r["event_id"]) + # One already-read row and one unread row vanish after page 1. With + # offset paging the next page would start one row late and skip a row. + read, unread = ordered[10], ordered[3000] + session = _PagingSession(lambda q: rows, on_page=_delete_after_first_page([read, unread])) + expected = [r["event_id"] for r in rows if r is not unread] + gaps = await find_gaps(_client(session), tenant_id=TENANT) + _assert_each_once(expected, [g["event_id"] for g in gaps]) + + +@pytest.mark.asyncio +async def test_find_broken_chains_deletion_between_pages_skips_nothing(): + rows = _shipping_rows(TOTAL_ROWS) + ordered = sorted(rows, key=lambda r: (r["event_id"], r["lot_tlc"])) + read = ordered[0] + + def rows_for(q): + if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q: + return rows + return [] + + unread = ordered[2500] + session = _PagingSession(rows_for, on_page=_delete_after_first_page([read, unread])) + expected = [r["event_id"] for r in rows if r is not unread] + violations = await find_broken_chains(_client(session), tenant_id=TENANT) + _assert_each_once(expected, [v["event_id"] for v in violations]) + + +@pytest.mark.asyncio +async def test_find_orphaned_lots_deletion_between_pages_skips_nothing(): + rows = _orphan_rows(TOTAL_ROWS) + ordered = sorted(rows, key=lambda r: r["tlc"]) + read, unread = ordered[5], ordered[4400] + session = _PagingSession(lambda q: rows, on_page=_delete_after_first_page([read, unread])) + expected = [r["tlc"] for r in rows if r is not unread] + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + _assert_each_once(expected, [o.tlc for o in orphans]) + + +def _repeat_row_in_first_page(n, page): + # Serve row 4 a second time, inside page 1, before the cursor moves on. + if n == 1: + return page[:5] + [dict(page[4])] + page[5:] + return page + + +@pytest.mark.asyncio +async def test_find_gaps_drops_row_repeated_within_one_page(): + rows = _shipping_rows(TOTAL_ROWS) + session = _PagingSession(lambda q: rows, dup_in_page=_repeat_row_in_first_page) + gaps = await find_gaps(_client(session), tenant_id=TENANT) + _assert_each_once([r["event_id"] for r in rows], [g["event_id"] for g in gaps]) + + +@pytest.mark.asyncio +async def test_find_broken_chains_drops_row_repeated_within_one_page(): + rows = _shipping_rows(TOTAL_ROWS) + + def rows_for(q): + if "NOT EXISTS" in q and "merkle_prev_hash" not in q and "origin_date" not in q: + return rows + return [] + + session = _PagingSession(rows_for, dup_in_page=_repeat_row_in_first_page) + violations = await find_broken_chains(_client(session), tenant_id=TENANT) + _assert_each_once([r["event_id"] for r in rows], [v["event_id"] for v in violations]) + + +@pytest.mark.asyncio +async def test_find_orphaned_lots_drops_row_repeated_within_one_page(): + rows = _orphan_rows(TOTAL_ROWS) + session = _PagingSession(lambda q: rows, dup_in_page=_repeat_row_in_first_page) + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + _assert_each_once([r["tlc"] for r in rows], [o.tlc for o in orphans]) + + +@pytest.mark.asyncio +async def test_non_hit_anchor_rows_advance_cursor_but_are_not_reported(): + rows = _orphan_rows(TOTAL_ROWS) + for i, r in enumerate(rows): + r["_hit"] = i % 2 == 0 + session = _PagingSession(lambda q: rows) + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + _assert_each_once([r["tlc"] for r in rows if r["_hit"]], [o.tlc for o in orphans]) + assert len(session.calls) == 3 + + +@pytest.mark.asyncio +async def test_page_filter_and_limit_sit_inside_the_match(): + session = _PagingSession(lambda q: []) + await find_broken_chains(_client(session), tenant_id=TENANT) + await find_gaps(_client(session), tenant_id=TENANT) + for query, _ in session.calls: + assert "CALL {" not in query + # The cursor filter and LIMIT come before the final RETURN. + assert query.index("LIMIT $page_size") < query.rindex("RETURN") + assert query.index("$after_key") < query.rindex("RETURN") + + +def _hit_expr(query: str) -> str: + """The RETURN expression aliased ``_hit``, whitespace collapsed.""" + ret = query[query.rindex("RETURN") :] + return " ".join(ret[: ret.index("AS _hit")].split()) + + +@pytest.mark.asyncio +async def test_hit_flags_coalesce_cypher_null_to_false(): + session = _PagingSession(lambda q: []) + await find_broken_chains(_client(session), tenant_id=TENANT) + await find_orphaned_lots(_client(session), tenant_id=TENANT) + hits = [_hit_expr(q) for q, _ in session.calls if "AS _hit" in q] + assert len(hits) == 3 # temporal paradox, crypto gap, orphan + assert all(h.split("_page_key,", 1)[1].strip().startswith("coalesce(") for h in hits) + assert all(h.endswith(", false)") for h in hits) + + +@pytest.mark.asyncio +async def test_null_merkle_hash_hit_is_not_a_crypto_gap(): + # Null merkle_hash on shipping and origin: the comparison is null in + # Cypher, so the row must not be reported. + rows = _shipping_rows(5) + for r in rows: + r["_hit"] = None + + def rows_for(q): + return rows if "merkle_prev_hash" in q else [] + + session = _PagingSession(rows_for) + violations = await find_broken_chains(_client(session), tenant_id=TENANT) + assert violations == [] + + +@pytest.mark.asyncio +async def test_uncomparable_dates_hit_is_not_an_orphan(): + # last_event_date empty vs a cutoff date: the comparison is null. + rows = _orphan_rows(5) + for r in rows: + r["last_event_date"] = "" + r["_hit"] = None + session = _PagingSession(lambda q: rows) + orphans = await find_orphaned_lots(_client(session), tenant_id=TENANT) + assert orphans == [] + + +@pytest.mark.asyncio +async def test_page_clause_orders_and_limits_in_its_own_with(): + # openCypher: WITH [ORDER BY] [SKIP] [LIMIT] [WHERE]. A WHERE + # before ORDER BY is not valid, so filter and order use separate WITHs. + session = _PagingSession(lambda q: []) + await find_gaps(_client(session), tenant_id=TENANT) + for query, _ in session.calls: + lines = [ln.strip() for ln in query.splitlines()] + w = next(i for i, ln in enumerate(lines) if ln.startswith("WHERE $after_key IS NULL")) + assert lines[w + 1].startswith("WITH ") + assert lines[w + 2].startswith("ORDER BY _page_key[0]") + assert lines[w + 3] == "LIMIT $page_size"