From 7b8fbdedb97ba371782d16d6a629c0f8a2e69bf5 Mon Sep 17 00:00:00 2001 From: Steve McMaster Date: Sun, 26 Jul 2026 17:46:53 +0000 Subject: [PATCH 1/2] fix(pagination): fix reverse paging and reject stale-sort cursors Reverse (`reverse=true`) cursor pagination was broken across several listings. The cases list inverted its cursor WHERE clause but never inverted the ORDER BY, so the query selected every row preceding the cursor and then kept the first `limit` of them in forward order: paging back from page 3 returned page 1, not page 2. It also never set `prev_cursor` on a reverse response and never swapped `has_more`/`has_previous`, so the Previous control went dead after one step while the response still reported `has_previous=true`. Six other listings had subsets of the same defect: - tables `list_rows`: no ORDER BY inversion, so the same page-1 teleport - case table rows: the reverse ORDER BY was appended to the one already on the statement, making it a no-op, and cursors were derived from the already-reversed page and then swapped a second time - agent skills (skills and versions) and agent preset versions: the page was never reversed back into display order and neither the cursors nor the flags were swapped - workflow management: `has_more`/`has_previous` never swapped - EE inbox agent runs: reverse-mode cursor guards and flags Add `take_cursor_page` and `build_cursor_page` to `tracecat/pagination.py` and route all seven call sites through them so the trim/reverse/swap step has a single definition and a single set of tests. Also reject a cursor whose `sort_column` does not match the sort of the request carrying it. Previously the mismatch was detected and then the cursor filter was dropped entirely, returning the first page while still advertising `prev_cursor` and `has_previous=true`. `InvalidCursorError` subclasses `ValueError`, which the cases and tables routers already map to 400. Co-Authored-By: Claude Opus 5 (1M context) --- .../tracecat_ee/inbox/providers/agent_runs.py | 91 +++----- tests/unit/api/test_api_cases.py | 29 ++- tests/unit/test_cases_service.py | 176 +++++++++++++- tests/unit/test_cursor_page.py | 220 ++++++++++++++++++ tests/unit/test_tables_service.py | 59 ++++- tracecat/agent/preset/service.py | 41 ++-- tracecat/agent/skill/service.py | 80 +++---- tracecat/cases/rows/service.py | 65 +++--- tracecat/cases/service.py | 183 +++++++-------- tracecat/pagination.py | 127 ++++++++++ tracecat/tables/service.py | 166 ++++++------- tracecat/workflow/management/management.py | 55 ++--- 12 files changed, 905 insertions(+), 387 deletions(-) create mode 100644 tests/unit/test_cursor_page.py diff --git a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py b/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py index 8eaa14a824..2e50a4b8e3 100644 --- a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py +++ b/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py @@ -23,7 +23,12 @@ from tracecat.inbox.schemas import InboxItemRead, UserSummary, WorkflowSummary from tracecat.inbox.types import InboxGroup, InboxItemStatus, InboxItemType from tracecat.logger import logger -from tracecat.pagination import BaseCursorPaginator, CursorPaginatedResponse +from tracecat.pagination import ( + BaseCursorPaginator, + CursorPaginatedResponse, + build_cursor_page, + take_cursor_page, +) from tracecat_ee.agent.types import AgentWorkflowID # The error signal is fully persisted (AgentSession.last_error), so Temporal is @@ -379,69 +384,39 @@ async def list_items( stmt = base_stmt.order_by(order_clause, id_order).limit(limit + 1) result = await self.session.execute(stmt) - sessions = list(result.scalars().all()) - - # Check if there are more items - has_more = len(sessions) > limit - if has_more: - sessions = sessions[:limit] + scanned_sessions, has_more = take_cursor_page( + list(result.scalars().all()), limit=limit + ) - # The scan ran in scan_desc order; reverse back into display order so - # the page reads in the requested sort regardless of pagination - # direction. - if reverse: - sessions.reverse() + # The scan ran in scan_desc order; build_cursor_page reverses the page + # back into display order so it reads in the requested sort regardless + # of pagination direction, and swaps the cursors and flags accordingly. + page = build_cursor_page( + scanned_sessions, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda session: self.encode_cursor( + id=session.id, + sort_column=sort_col, + # Use the correct column value based on sort_col + sort_value=( + session.updated_at + if sort_col == "updated_at" + else session.created_at + ), + ), + ) # Enrich sessions - items = await self._enrich_sessions(sessions) - - # Generate cursors - next_cursor = None - prev_cursor = None - - if items: - if has_more: - last_item = items[-1] - # Get the session for this item to access the sort column value - last_session = next( - (s for s in sessions if s.id == last_item.source_id), None - ) - if last_session: - # Use the correct column value based on sort_col - sort_value = ( - last_session.updated_at - if sort_col == "updated_at" - else last_session.created_at - ) - next_cursor = self.encode_cursor( - id=last_item.id, - sort_column=sort_col, - sort_value=sort_value, - ) - if cursor: - first_item = items[0] - first_session = next( - (s for s in sessions if s.id == first_item.source_id), None - ) - if first_session: - # Use the correct column value based on sort_col - sort_value = ( - first_session.updated_at - if sort_col == "updated_at" - else first_session.created_at - ) - prev_cursor = self.encode_cursor( - id=first_item.id, - sort_column=sort_col, - sort_value=sort_value, - ) + items = await self._enrich_sessions(page.items) return CursorPaginatedResponse( items=items, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=cursor is not None, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, total_estimate=None, ) diff --git a/tests/unit/api/test_api_cases.py b/tests/unit/api/test_api_cases.py index cc1089b908..b752b4c4e7 100644 --- a/tests/unit/api/test_api_cases.py +++ b/tests/unit/api/test_api_cases.py @@ -36,7 +36,7 @@ TracecatConflictError, TracecatValidationError, ) -from tracecat.pagination import CursorPaginatedResponse +from tracecat.pagination import CursorPaginatedResponse, InvalidCursorError @pytest.fixture @@ -263,6 +263,33 @@ async def test_list_cases_validates_field_ids_even_when_page_is_empty( ) +@pytest.mark.anyio +async def test_list_cases_returns_400_for_a_stale_sort_cursor( + client: TestClient, + test_admin_role: Role, +) -> None: + """A cursor from another sort is a client error, not a silent first page.""" + with patch.object(cases_router, "CasesService") as MockService: + mock_svc = AsyncMock() + mock_svc.list_cases.side_effect = InvalidCursorError( + "Cursor was created for sort column 'created_at', " + "but this request sorts by 'priority'." + ) + MockService.return_value = mock_svc + + response = client.get( + "/cases", + params={ + "workspace_id": str(test_admin_role.workspace_id), + "cursor": "stale-cursor", + "order_by": "priority", + }, + ) + + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "sorts by 'priority'" in response.json()["detail"] + + @pytest.mark.anyio async def test_list_case_events_includes_comment_activity( client: TestClient, diff --git a/tests/unit/test_cases_service.py b/tests/unit/test_cases_service.py index 25475d67bd..299c3d0293 100644 --- a/tests/unit/test_cases_service.py +++ b/tests/unit/test_cases_service.py @@ -53,7 +53,7 @@ TracecatConflictError, TracecatNotFoundError, ) -from tracecat.pagination import CursorPaginationParams +from tracecat.pagination import CursorPaginationParams, InvalidCursorError from tracecat.tables.enums import SqlType pytestmark = pytest.mark.usefixtures("db") @@ -1746,6 +1746,180 @@ async def test_search_cases_aliases_list_cases( assert search_response.model_dump() == list_response.model_dump() + async def test_search_cases_reverse_pagination_returns_preceding_page( + self, cases_service: CasesService + ) -> None: + """Paging backward should land on the page immediately before the cursor.""" + for i in range(6): + await cases_service.create_case( + CaseCreate( + summary=f"Case {i}", + description=f"Case {i}", + status=CaseStatus.NEW, + priority=CasePriority.MEDIUM, + severity=CaseSeverity.LOW, + ) + ) + await asyncio.sleep(0.01) + + async def page(cursor: str | None, reverse: bool = False): + return await cases_service.search_cases( + params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse) + ) + + page_1 = await page(None) + page_2 = await page(page_1.next_cursor) + page_3 = await page(page_2.next_cursor) + + # Sanity check the forward walk before asserting on the backward one. + forward_ids = [ + item.id for item in (*page_1.items, *page_2.items, *page_3.items) + ] + assert len(set(forward_ids)) == 6 + + # Stepping back from page 3 must return page 2, not the first page. + assert page_3.prev_cursor is not None + back_to_2 = await page(page_3.prev_cursor, reverse=True) + assert [item.id for item in back_to_2.items] == [ + item.id for item in page_2.items + ] + assert back_to_2.has_more is True + assert back_to_2.has_previous is True + assert back_to_2.prev_cursor is not None + + # Reverse responses must keep advertising a usable prev_cursor. + back_to_1 = await page(back_to_2.prev_cursor, reverse=True) + assert [item.id for item in back_to_1.items] == [ + item.id for item in page_1.items + ] + assert back_to_1.has_previous is False + assert back_to_1.prev_cursor is None + assert back_to_1.has_more is True + + # And the reverse page's next_cursor must walk forward again. + forward_to_3 = await page(back_to_2.next_cursor) + assert [item.id for item in forward_to_3.items] == [ + item.id for item in page_3.items + ] + + async def test_search_cases_reverse_pagination_ascending_sort( + self, cases_service: CasesService + ) -> None: + """Backward paging should also invert an ascending sort correctly.""" + for i in range(6): + await cases_service.create_case( + CaseCreate( + summary=f"Ascending {i}", + description=f"Ascending {i}", + status=CaseStatus.NEW, + priority=CasePriority.MEDIUM, + severity=CaseSeverity.LOW, + ) + ) + await asyncio.sleep(0.01) + + async def page(cursor: str | None, reverse: bool = False): + return await cases_service.search_cases( + params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse), + order_by="created_at", + sort="asc", + ) + + page_1 = await page(None) + page_2 = await page(page_1.next_cursor) + page_3 = await page(page_2.next_cursor) + + back_to_2 = await page(page_3.prev_cursor, reverse=True) + + assert [item.id for item in back_to_2.items] == [ + item.id for item in page_2.items + ] + assert back_to_2.has_more is True + assert back_to_2.has_previous is True + + async def test_search_cases_reverse_pagination_enum_sort( + self, cases_service: CasesService + ) -> None: + """Backward paging should work for enum sorts, which rank rather than compare.""" + priorities = [ + CasePriority.CRITICAL, + CasePriority.HIGH, + CasePriority.MEDIUM, + CasePriority.LOW, + ] + for i, priority in enumerate(priorities): + await cases_service.create_case( + CaseCreate( + summary=f"Priority {i}", + description=f"Priority {i}", + status=CaseStatus.NEW, + priority=priority, + severity=CaseSeverity.LOW, + ) + ) + await asyncio.sleep(0.01) + + async def page(cursor: str | None, reverse: bool = False): + return await cases_service.search_cases( + params=CursorPaginationParams(limit=1, cursor=cursor, reverse=reverse), + order_by="priority", + sort="desc", + ) + + page_1 = await page(None) + page_2 = await page(page_1.next_cursor) + page_3 = await page(page_2.next_cursor) + + back_to_2 = await page(page_3.prev_cursor, reverse=True) + + assert [item.id for item in back_to_2.items] == [ + item.id for item in page_2.items + ] + assert back_to_2.items[0].priority == page_2.items[0].priority + assert back_to_2.has_more is True + assert back_to_2.has_previous is True + + async def test_search_cases_rejects_a_cursor_from_a_different_sort( + self, cases_service: CasesService + ) -> None: + """A stale-sort cursor must fail loudly instead of rewinding to page 1.""" + for i in range(3): + await cases_service.create_case( + CaseCreate( + summary=f"Stale cursor {i}", + description=f"Stale cursor {i}", + status=CaseStatus.NEW, + priority=CasePriority.MEDIUM, + severity=CaseSeverity.LOW, + ) + ) + await asyncio.sleep(0.01) + + page_1 = await cases_service.search_cases( + params=CursorPaginationParams(limit=1, cursor=None, reverse=False), + order_by="created_at", + ) + assert page_1.next_cursor is not None + + # Same cursor, different sort column: the cursor's sort value cannot be + # compared against the new ORDER BY. + with pytest.raises(InvalidCursorError): + await cases_service.search_cases( + params=CursorPaginationParams( + limit=1, cursor=page_1.next_cursor, reverse=False + ), + order_by="priority", + ) + + # The cursor still works for the sort it was issued under. + page_2 = await cases_service.search_cases( + params=CursorPaginationParams( + limit=1, cursor=page_1.next_cursor, reverse=False + ), + order_by="created_at", + ) + assert page_2.items[0].id != page_1.items[0].id + async def test_search_cases_gates_duration_selectinload( self, cases_service: CasesService ) -> None: diff --git a/tests/unit/test_cursor_page.py b/tests/unit/test_cursor_page.py new file mode 100644 index 0000000000..883e5511ca --- /dev/null +++ b/tests/unit/test_cursor_page.py @@ -0,0 +1,220 @@ +"""Tests for the shared keyset pagination page builder. + +These exercise the contract that every cursor-paginated service relies on: +a reverse scan runs in the inverted sort order, and the page it produces must +be handed back to the client in forward (display) semantics. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + +import pytest + +from tracecat.pagination import ( + CursorData, + InvalidCursorError, + build_cursor_page, + take_cursor_page, + validate_cursor_sort_column, +) + + +@dataclass(frozen=True) +class Row: + """A row in a list sorted newest-first by ``rank``.""" + + rank: int + + @property + def cursor(self) -> str: + return f"cursor-{self.rank}" + + +# Display order is descending by rank: rank 0 is the first row of page 1. +ROWS = [Row(rank=i) for i in range(100)] + + +def scan(*, limit: int, cursor: str | None, reverse: bool) -> list[Row]: + """Mimic a keyset scan: filter past the cursor, order, over-fetch by one. + + Forward scans walk down the display order; reverse scans walk back up it in + the inverted order, which is what puts the rows adjacent to the cursor + inside the LIMIT. + """ + anchor = None if cursor is None else int(cursor.removeprefix("cursor-")) + if reverse: + rows = [row for row in ROWS if anchor is not None and row.rank < anchor] + rows.sort(key=lambda row: row.rank, reverse=True) + else: + rows = [row for row in ROWS if anchor is None or row.rank > anchor] + rows.sort(key=lambda row: row.rank) + return rows[: limit + 1] + + +def page(*, limit: int, cursor: str | None = None, reverse: bool = False): + rows, has_more = take_cursor_page( + scan(limit=limit, cursor=cursor, reverse=reverse), limit=limit + ) + return build_cursor_page( + rows, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda row: row.cursor, + ) + + +def ranks(rows: list[Row]) -> list[int]: + return [row.rank for row in rows] + + +def test_take_cursor_page_trims_the_extra_row() -> None: + rows, has_more = take_cursor_page(ROWS[:11], limit=10) + + assert ranks(rows) == list(range(10)) + assert has_more is True + + +def test_take_cursor_page_reports_no_more_when_short() -> None: + rows, has_more = take_cursor_page(ROWS[:3], limit=10) + + assert ranks(rows) == [0, 1, 2] + assert has_more is False + + +def test_first_page_has_no_previous() -> None: + first = page(limit=20) + + assert ranks(first.items) == list(range(20)) + assert first.has_more is True + assert first.has_previous is False + assert first.prev_cursor is None + assert first.next_cursor == "cursor-19" + + +def test_forward_page_exposes_both_cursors() -> None: + second = page(limit=20, cursor="cursor-19") + + assert ranks(second.items) == list(range(20, 40)) + assert second.has_more is True + assert second.has_previous is True + assert second.next_cursor == "cursor-39" + assert second.prev_cursor == "cursor-20" + + +def test_reverse_page_returns_the_preceding_page_in_display_order() -> None: + # Regression: the reverse branch used to return page 1 from any page, since + # it selected everything before the cursor and then kept the top `limit`. + third = page(limit=20, cursor="cursor-39") + assert ranks(third.items) == list(range(40, 60)) + + back = page(limit=20, cursor=third.prev_cursor, reverse=True) + + assert ranks(back.items) == list(range(20, 40)) + assert back.has_more is True + assert back.has_previous is True + + +def test_reverse_page_cursors_round_trip() -> None: + third = page(limit=20, cursor="cursor-39") + back = page(limit=20, cursor=third.prev_cursor, reverse=True) + + # Stepping back again must reach page 1, not repeat page 2. + assert back.prev_cursor == "cursor-20" + further_back = page(limit=20, cursor=back.prev_cursor, reverse=True) + assert ranks(further_back.items) == list(range(20)) + + # And stepping forward from the reverse page must return page 3. + assert back.next_cursor == "cursor-39" + forward = page(limit=20, cursor=back.next_cursor) + assert ranks(forward.items) == ranks(third.items) + + +def test_reverse_page_onto_the_first_page_reports_no_previous() -> None: + second = page(limit=20, cursor="cursor-19") + back = page(limit=20, cursor=second.prev_cursor, reverse=True) + + assert ranks(back.items) == list(range(20)) + # Nothing precedes page 1, but page 2 is still ahead of it. + assert back.has_previous is False + assert back.prev_cursor is None + assert back.has_more is True + assert back.next_cursor == "cursor-19" + + +def test_empty_page_has_no_cursors() -> None: + empty = build_cursor_page( + [], + cursor="cursor-99", + reverse=True, + has_more=False, + encode_cursor=lambda row: "unreachable", + ) + + assert empty.items == [] + assert empty.next_cursor is None + assert empty.prev_cursor is None + assert empty.has_more is False + assert empty.has_previous is False + + +def test_validate_cursor_sort_column_returns_the_sort_value() -> None: + cursor = CursorData(id="abc", sort_column="created_at", sort_value="2026-01-01") + + assert validate_cursor_sort_column(cursor, sort_column="created_at") == ( + datetime(2026, 1, 1) + ) + + +def test_validate_cursor_sort_column_rejects_a_different_sort() -> None: + """A cursor from another sort must 400, not silently rewind to page 1.""" + cursor = CursorData(id="abc", sort_column="created_at", sort_value="2026-01-01") + + with pytest.raises(InvalidCursorError, match="sorts by 'priority'"): + validate_cursor_sort_column(cursor, sort_column="priority") + + +def test_validate_cursor_sort_column_rejects_a_sortless_cursor() -> None: + """Cursors predating sort-aware pagination carry no sort column.""" + cursor = CursorData(id="abc") + + with pytest.raises(InvalidCursorError): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_a_missing_sort_value() -> None: + cursor = CursorData(id="abc", sort_column="created_at") + + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_a_wrongly_typed_sort_value() -> None: + """Enum and task-count sorts order by a rank, so their cursors carry ints.""" + cursor = CursorData(id="abc", sort_column="priority", sort_value="high") + + with pytest.raises(InvalidCursorError, match="wrong type"): + validate_cursor_sort_column(cursor, sort_column="priority", expected_type=int) + + +def test_invalid_cursor_error_is_a_value_error() -> None: + """Routers map ValueError to 400; the subclass keeps that mapping.""" + assert issubclass(InvalidCursorError, ValueError) + + +@pytest.mark.parametrize("reverse", [False, True]) +def test_build_cursor_page_does_not_mutate_the_input(reverse: bool) -> None: + rows = list(ROWS[:5]) + original = list(rows) + + build_cursor_page( + rows, + cursor="cursor-9", + reverse=reverse, + has_more=True, + encode_cursor=lambda row: row.cursor, + ) + + assert rows == original diff --git a/tests/unit/test_tables_service.py b/tests/unit/test_tables_service.py index b7a90c871d..52313eb8af 100644 --- a/tests/unit/test_tables_service.py +++ b/tests/unit/test_tables_service.py @@ -17,7 +17,7 @@ from tracecat.db.models import Table, TableColumn, Workspace from tracecat.exceptions import TracecatAuthorizationError, TracecatNotFoundError from tracecat.logger import logger -from tracecat.pagination import CursorPaginationParams +from tracecat.pagination import CursorPaginationParams, InvalidCursorError from tracecat.tables.common import ( ColumnHasDuplicateValuesError, handle_default_value, @@ -1437,6 +1437,63 @@ async def test_list_rows_reverse_pagination_flags( assert reverse_page.has_more is True assert reverse_page.has_previous is False + async def test_list_rows_reverse_pagination_returns_preceding_page( + self, tables_service: TablesService, table: Table + ) -> None: + """Paging backward should land on the page immediately before the cursor.""" + for i in range(6): + await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Row{i}", "age": i}) + ) + + page_1 = await _list_rows_page(tables_service, table, limit=2) + page_2 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_1.next_cursor + ) + page_3 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_2.next_cursor + ) + + assert page_3.prev_cursor is not None + back_to_2 = await _list_rows_page( + tables_service, table, limit=2, cursor=page_3.prev_cursor, reverse=True + ) + + assert [row["id"] for row in back_to_2.items] == [ + row["id"] for row in page_2.items + ] + assert back_to_2.has_more is True + assert back_to_2.has_previous is True + + forward_to_3 = await _list_rows_page( + tables_service, table, limit=2, cursor=back_to_2.next_cursor + ) + assert [row["id"] for row in forward_to_3.items] == [ + row["id"] for row in page_3.items + ] + + async def test_list_rows_rejects_a_cursor_from_a_different_sort( + self, tables_service: TablesService, table: Table + ) -> None: + """A stale-sort cursor must fail loudly instead of rewinding to page 1.""" + for i in range(3): + await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Stale{i}", "age": i}) + ) + + page_1 = await _list_rows_page(tables_service, table, limit=1) + assert page_1.next_cursor is not None + + with pytest.raises(InvalidCursorError): + await tables_service.list_rows( + table, + params=CursorPaginationParams( + limit=1, cursor=page_1.next_cursor, reverse=False + ), + order_by="age", + sort="asc", + ) + async def test_table_editor_list_rows_reverse_pagination_flags( self, tables_service: TablesService, table: Table ) -> None: diff --git a/tracecat/agent/preset/service.py b/tracecat/agent/preset/service.py index dc7f11e009..026416a44d 100644 --- a/tracecat/agent/preset/service.py +++ b/tracecat/agent/preset/service.py @@ -81,6 +81,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.registry.actions.service import RegistryActionsService from tracecat.secrets import secrets_manager @@ -1361,33 +1363,26 @@ async def list_versions( updated_at, ) in result.tuples().all() ] - has_more = len(versions) > params.limit - items = versions[: params.limit] - - next_cursor = None - if has_more and items: - last_version = items[-1] - next_cursor = paginator.encode_cursor( - last_version.id, - sort_column="version", - sort_value=last_version.version, - ) + scanned, has_more = take_cursor_page(versions, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first_version = items[0] - prev_cursor = paginator.encode_cursor( - first_version.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda version: paginator.encode_cursor( + version.id, sort_column="version", - sort_value=first_version.version, - ) + sort_value=version.version, + ), + ) return CursorPaginatedResponse( - items=list(items), - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) diff --git a/tracecat/agent/skill/service.py b/tracecat/agent/skill/service.py index b3152659fb..af3d75d687 100644 --- a/tracecat/agent/skill/service.py +++ b/tracecat/agent/skill/service.py @@ -71,6 +71,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.service import BaseWorkspaceService, requires_entitlement from tracecat.storage import blob @@ -1628,33 +1630,26 @@ async def list_skills( stmt = stmt.order_by(Skill.updated_at.desc(), Skill.id.desc()) stmt = stmt.limit(params.limit + 1) skills = (await self.session.execute(stmt)).scalars().all() - has_more = len(skills) > params.limit - items = skills[: params.limit] - - next_cursor = None - if has_more and items: - last = items[-1] - next_cursor = paginator.encode_cursor( - last.id, - sort_column="updated_at", - sort_value=last.updated_at, - ) + scanned, has_more = take_cursor_page(skills, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first = items[0] - prev_cursor = paginator.encode_cursor( - first.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda skill: paginator.encode_cursor( + skill.id, sort_column="updated_at", - sort_value=first.updated_at, - ) + sort_value=skill.updated_at, + ), + ) return CursorPaginatedResponse( - items=[self._build_skill_read_minimal(skill) for skill in items], - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=[self._build_skill_read_minimal(skill) for skill in page.items], + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) @@ -2202,26 +2197,19 @@ async def list_versions( stmt = stmt.order_by(SkillVersion.version.desc(), SkillVersion.id.desc()) stmt = stmt.limit(params.limit + 1) versions = (await self.session.execute(stmt)).scalars().all() - has_more = len(versions) > params.limit - items = versions[: params.limit] - - next_cursor = None - if has_more and items: - last = items[-1] - next_cursor = paginator.encode_cursor( - last.id, - sort_column="version", - sort_value=last.version, - ) + scanned, has_more = take_cursor_page(versions, limit=params.limit) - prev_cursor = None - if params.cursor and items: - first = items[0] - prev_cursor = paginator.encode_cursor( - first.id, + page = build_cursor_page( + scanned, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda version: paginator.encode_cursor( + version.id, sort_column="version", - sort_value=first.version, - ) + sort_value=version.version, + ), + ) return CursorPaginatedResponse( items=[ @@ -2238,12 +2226,12 @@ async def list_versions( created_at=version.created_at, updated_at=version.updated_at, ) - for version in items + for version in page.items ], - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) @requires_entitlement(Entitlement.AGENT_ADDONS) diff --git a/tracecat/cases/rows/service.py b/tracecat/cases/rows/service.py index f7dc8c1c39..eea757c38e 100644 --- a/tracecat/cases/rows/service.py +++ b/tracecat/cases/rows/service.py @@ -24,7 +24,12 @@ from tracecat.cases.service import CaseEventsService from tracecat.db.models import Case, CaseTableRow, Table from tracecat.exceptions import TracecatNotFoundError -from tracecat.pagination import BaseCursorPaginator, CursorPaginatedResponse +from tracecat.pagination import ( + BaseCursorPaginator, + CursorPaginatedResponse, + build_cursor_page, + take_cursor_page, +) from tracecat.service import BaseWorkspaceService from tracecat.tables.service import TablesService @@ -67,7 +72,6 @@ async def list_rows( CaseTableRow.case_id == case_id, ) .options(selectinload(CaseTableRow.case)) - .order_by(CaseTableRow.created_at.desc(), CaseTableRow.id.desc()) ) if cursor: @@ -98,9 +102,6 @@ async def list_rows( ), ) ) - stmt = stmt.order_by( - CaseTableRow.created_at.asc(), CaseTableRow.id.asc() - ) else: stmt = stmt.where( or_( @@ -112,44 +113,42 @@ async def list_rows( ) ) + # Reverse pagination scans away from the cursor in ascending order so + # LIMIT keeps the rows nearest the cursor; build_cursor_page puts them + # back into display order. + if reverse: + stmt = stmt.order_by(CaseTableRow.created_at.asc(), CaseTableRow.id.asc()) + else: + stmt = stmt.order_by(CaseTableRow.created_at.desc(), CaseTableRow.id.desc()) + stmt = stmt.limit(limit + 1) result = await self.session.execute(stmt) links = result.scalars().all() - has_more = len(links) > limit - items = links[:limit] if has_more else links - has_previous = cursor is not None + scanned_links, has_more = take_cursor_page(links, limit=limit) - if reverse: - items = list(reversed(items)) - - hydrated = await self._hydrate_links(items, include_row_data=include_row_data) - - next_cursor = None - prev_cursor = None - if items and has_more: - next_cursor = paginator.encode_cursor( - items[-1].id, - sort_column="created_at", - sort_value=items[-1].created_at, - ) - if items and cursor: - prev_cursor = paginator.encode_cursor( - items[0].id, + page = build_cursor_page( + scanned_links, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda link: paginator.encode_cursor( + link.id, sort_column="created_at", - sort_value=items[0].created_at, - ) + sort_value=link.created_at, + ), + ) - if reverse: - next_cursor, prev_cursor = prev_cursor, next_cursor - has_more, has_previous = has_previous, has_more + hydrated = await self._hydrate_links( + page.items, include_row_data=include_row_data + ) return CursorPaginatedResponse( items=hydrated, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def link_row( diff --git a/tracecat/cases/service.py b/tracecat/cases/service.py index 304eaf1f3b..93d753b736 100644 --- a/tracecat/cases/service.py +++ b/tracecat/cases/service.py @@ -127,6 +127,9 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, + validate_cursor_sort_column, ) from tracecat.service import BaseWorkspaceService, requires_entitlement from tracecat.tables.common import ( @@ -493,81 +496,84 @@ async def search_cases( cursor_data = paginator.decode_cursor(params.cursor) cursor_id = uuid.UUID(cursor_data.id) - # Check if cursor was created with the same sort column (for proper pagination) - cursor_sort_value = cursor_data.sort_value - cursor_has_sort_value = ( - cursor_data.sort_column == sort_column and cursor_sort_value is not None + # A cursor from a different sort cannot filter this query. Reject it + # instead of dropping the filter, which would silently return the + # first page while still reporting has_previous. + # Enum and task-count sorts order by a computed rank, so their + # cursors carry that rank rather than the column value. + sort_cursor_value = validate_cursor_sort_column( + cursor_data, + sort_column=sort_column, + expected_type=int + if sort_column == "tasks" or enum_sort_values is not None + else None, ) - if cursor_has_sort_value and sort_column == "tasks": - cursor_has_sort_value = isinstance(cursor_sort_value, int) - elif cursor_has_sort_value and enum_sort_values is not None: - cursor_has_sort_value = isinstance(cursor_sort_value, int) - - if cursor_has_sort_value: - sort_filter_col = sort_attr - sort_cursor_value = cursor_sort_value - - # Composite filtering: (sort_col, id) matches ORDER BY - # Use id as tie-breaker since it's always unique - if sort_direction == "asc": - if params.reverse: - # Going backward: get records before cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col < sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id < cursor_id, - ), - ) - ) - else: - # Going forward: get records after cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col > sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id > cursor_id, - ), - ) + sort_filter_col = sort_attr + + # Composite filtering: (sort_col, id) matches ORDER BY + # Use id as tie-breaker since it's always unique + if sort_direction == "asc": + if params.reverse: + # Going backward: get records before cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col < sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id < cursor_id, + ), ) + ) else: - # Descending order - if params.reverse: - # Going backward: get records after cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col > sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id > cursor_id, - ), - ) + # Going forward: get records after cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col > sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id > cursor_id, + ), ) - else: - # Going forward: get records before cursor in sort order - stmt = stmt.where( - or_( - sort_filter_col < sort_cursor_value, - and_( - sort_filter_col == sort_cursor_value, - Case.id < cursor_id, - ), - ) + ) + else: + # Descending order + if params.reverse: + # Going backward: get records after cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col > sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id > cursor_id, + ), + ) + ) + else: + # Going forward: get records before cursor in sort order + stmt = stmt.where( + or_( + sort_filter_col < sort_cursor_value, + and_( + sort_filter_col == sort_cursor_value, + Case.id < cursor_id, + ), ) + ) # Apply sorting: (sort_col, id) for stable pagination - # Use id as tie-breaker unless we're already sorting by id + # Reverse pagination scans away from the cursor in the inverted sort + # order, so the rows nearest the cursor are the ones kept by LIMIT. + # build_cursor_page puts them back into display order. + scan_ascending = (sort_direction == "asc") != params.reverse if sort_column == "id": # No tie-breaker needed when sorting by id (already unique) - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_attr.asc()) else: stmt = stmt.order_by(sort_attr.desc()) else: # Add id as tie-breaker for non-unique columns - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_attr.asc(), Case.id.asc()) else: stmt = stmt.order_by(sort_attr.desc(), Case.id.desc()) @@ -577,17 +583,10 @@ async def search_cases( result = await self.session.execute(stmt) all_cases = result.scalars().all() - # Check if there are more items - has_more = len(all_cases) > params.limit - cases = all_cases[: params.limit] if has_more else all_cases + scanned_cases, has_more = take_cursor_page(all_cases, limit=params.limit) # Fetch task counts for all cases in one query (needed for cursor generation if sorting by tasks) - task_counts = await self.get_task_counts([case.id for case in cases]) - - # Generate cursors with sort column info for proper pagination - next_cursor = None - prev_cursor = None - has_previous = params.cursor is not None + task_counts = await self.get_task_counts([case.id for case in scanned_cases]) def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: """Encode cursor sort values using the same semantics as ORDER BY.""" @@ -599,31 +598,19 @@ def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: ) return getattr(case, sort_column, None) - if has_more and cases: - last_case = cases[-1] - sort_value = get_cursor_sort_value(last_case) - next_cursor = paginator.encode_cursor( - last_case.id, + # Generate cursors with sort column info for proper pagination + page = build_cursor_page( + scanned_cases, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda case: paginator.encode_cursor( + case.id, sort_column=sort_column, - sort_value=sort_value, - ) - - if params.cursor and cases: - first_case = cases[0] - sort_value = get_cursor_sort_value(first_case) - # For reverse pagination, swap the cursor meaning - if params.reverse: - next_cursor = paginator.encode_cursor( - first_case.id, - sort_column=sort_column, - sort_value=sort_value, - ) - else: - prev_cursor = paginator.encode_cursor( - first_case.id, - sort_column=sort_column, - sort_value=sort_value, - ) + sort_value=get_cursor_sort_value(case), + ), + ) + cases = page.items # Convert to CaseReadMinimal objects with tags and dropdown values case_items = [] @@ -685,10 +672,10 @@ def get_cursor_sort_value(case: Case) -> datetime | str | int | float | None: return CursorPaginatedResponse( items=case_items, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, total_estimate=total_estimate, ) diff --git a/tracecat/pagination.py b/tracecat/pagination.py index 6e5f8d3aa7..40ec3778bd 100644 --- a/tracecat/pagination.py +++ b/tracecat/pagination.py @@ -2,6 +2,8 @@ import base64 import json +from collections.abc import Callable, Sequence +from dataclasses import dataclass from datetime import datetime from typing import TypeVar from uuid import UUID @@ -45,6 +47,94 @@ class CursorPaginatedResponse[T](BaseModel): ) +class InvalidCursorError(ValueError): + """A cursor that cannot be applied to the query it was submitted with. + + Raised when the cursor's sort column or sort value does not line up with + the current sort, which happens when a client changes the sort without + restarting pagination. Dropping the cursor filter instead would silently + return the first page while still reporting ``has_previous``. + + Subclasses ``ValueError`` so routers surface it as 400, the same way + ``decode_cursor`` already reports a structurally malformed cursor. + """ + + +@dataclass(slots=True) +class CursorPage[T]: + """A single page of keyset-paginated rows in display order.""" + + items: list[T] + next_cursor: str | None + prev_cursor: str | None + has_more: bool + has_previous: bool + + +def take_cursor_page[T](rows: Sequence[T], *, limit: int) -> tuple[list[T], bool]: + """Trim an over-fetched keyset scan to a page. + + Args: + rows: Rows returned by a scan that requested ``limit + 1`` rows. + limit: Page size requested by the caller. + + Returns: + The page rows (still in scan order) and whether the scan found more + rows beyond the page in the direction it was scanning. + """ + has_more = len(rows) > limit + return list(rows[:limit]), has_more + + +def build_cursor_page[T]( + rows: Sequence[T], + *, + cursor: str | None, + reverse: bool, + has_more: bool, + encode_cursor: Callable[[T], str], +) -> CursorPage[T]: + """Derive page items, cursors, and flags from a trimmed keyset scan. + + Reverse pagination scans away from the cursor in the inverted sort order, + so the caller must invert its ``ORDER BY`` (and the cursor predicate) when + ``reverse`` is set. This function then puts the rows back into display + order and swaps the cursors and flags, since the scan's "more rows ahead" + is the page's "more rows behind". + + Args: + rows: Page rows in scan order, already trimmed by ``take_cursor_page``. + cursor: The cursor the scan started from, if any. + reverse: Whether the scan ran backwards from ``cursor``. + has_more: Whether the scan found rows beyond the page. + encode_cursor: Encodes the cursor anchored at a given row. + + Returns: + The page in display order with cursors and flags in forward semantics. + """ + items = list(rows) + next_cursor = encode_cursor(items[-1]) if has_more and items else None + prev_cursor = encode_cursor(items[0]) if cursor is not None and items else None + has_previous = cursor is not None + + if reverse: + items.reverse() + next_cursor, prev_cursor = prev_cursor, next_cursor + # In reverse mode "next" walks back toward the page we came from, which + # is only reachable when this page produced an anchor cursor. Tying it + # to a bare `cursor is not None` would advertise has_more=True with + # next_cursor=None on an empty page, enabling a dead pagination control. + has_more, has_previous = next_cursor is not None, has_more + + return CursorPage( + items=items, + next_cursor=next_cursor, + prev_cursor=prev_cursor, + has_more=has_more, + has_previous=has_previous, + ) + + class CursorData(BaseModel): """Internal structure for cursor data.""" @@ -73,6 +163,43 @@ def parse_datetime_string( return v +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = None, +) -> str | int | float | datetime: + """Return the cursor's sort value, or raise if it cannot filter this query. + + Args: + cursor: The decoded cursor. + sort_column: The sort column of the current request. + expected_type: Type the sort value must have, when the column's cursor + representation is narrower than the query's column type (enum sorts + encode a rank, not the enum value). + + Raises: + InvalidCursorError: The cursor belongs to a different sort. + """ + if cursor.sort_column != sort_column: + raise InvalidCursorError( + f"Cursor was created for sort column {cursor.sort_column!r}, " + f"but this request sorts by {sort_column!r}. " + "Restart pagination without a cursor after changing the sort." + ) + if cursor.sort_value is None: + raise InvalidCursorError( + f"Cursor is missing a sort value for column {sort_column!r}. " + "Restart pagination without a cursor." + ) + if expected_type is not None and not isinstance(cursor.sort_value, expected_type): + raise InvalidCursorError( + f"Cursor sort value for column {sort_column!r} has the wrong type. " + "Restart pagination without a cursor." + ) + return cursor.sort_value + + class BaseCursorPaginator: """Base class for cursor-based pagination.""" diff --git a/tracecat/tables/service.py b/tracecat/tables/service.py index 9ff7037c99..8b18a4e66b 100644 --- a/tracecat/tables/service.py +++ b/tracecat/tables/service.py @@ -41,6 +41,9 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, + validate_cursor_sort_column, ) from tracecat.service import BaseWorkspaceService from tracecat.tables.common import ( @@ -1453,76 +1456,76 @@ async def list_rows( cursor_id = UUID(cursor_data.id) - # Check if cursor was created with the same sort column - cursor_sort_value = cursor_data.sort_value - cursor_has_sort_value = ( - cursor_data.sort_column == sort_column and cursor_sort_value is not None + # A cursor from a different sort cannot filter this query. Reject it + # instead of dropping the filter, which would silently return the + # first page while still reporting has_previous. + sort_cursor_value = validate_cursor_sort_column( + cursor_data, sort_column=sort_column ) - if cursor_has_sort_value: - # Use sort column value for cursor filtering - sort_cursor_value = cursor_sort_value - - # Composite filtering: (sort_col, id) matches ORDER BY - if sort_direction == "asc": - if params.reverse: - # Going backward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) - ) - else: - # Going forward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) + # Composite filtering: (sort_col, id) matches ORDER BY + if sort_direction == "asc": + if params.reverse: + # Going backward: get records before cursor in sort order + stmt = stmt.where( + sa.or_( + sort_col < sort_cursor_value, + sa.and_( + sort_col == sort_cursor_value, + sa.column("id") < cursor_id, + ), ) + ) else: - # Descending order - if params.reverse: - # Going backward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) + # Going forward: get records after cursor in sort order + stmt = stmt.where( + sa.or_( + sort_col > sort_cursor_value, + sa.and_( + sort_col == sort_cursor_value, + sa.column("id") > cursor_id, + ), ) - else: - # Going forward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) + ) + else: + # Descending order + if params.reverse: + # Going backward: get records after cursor in sort order + stmt = stmt.where( + sa.or_( + sort_col > sort_cursor_value, + sa.and_( + sort_col == sort_cursor_value, + sa.column("id") > cursor_id, + ), + ) + ) + else: + # Going forward: get records before cursor in sort order + stmt = stmt.where( + sa.or_( + sort_col < sort_cursor_value, + sa.and_( + sort_col == sort_cursor_value, + sa.column("id") < cursor_id, + ), ) + ) # Apply sorting: (sort_col, id) for stable pagination - # Use id as tie-breaker unless we're already sorting by id + # Reverse pagination scans away from the cursor in the inverted sort + # order so LIMIT keeps the rows nearest the cursor; build_cursor_page + # puts them back into display order. + scan_ascending = (sort_direction == "asc") != params.reverse if sort_column == "id": # No tie-breaker needed when sorting by id (already unique) - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_col.asc()) else: stmt = stmt.order_by(sort_col.desc()) else: # Add id as tie-breaker for non-unique columns - if sort_direction == "asc": + if scan_ascending: stmt = stmt.order_by(sort_col.asc(), sa.column("id").asc()) else: stmt = stmt.order_by(sort_col.desc(), sa.column("id").desc()) @@ -1552,46 +1555,27 @@ async def list_rows( raise # Check if there are more items - has_more = len(rows) > params.limit - if has_more: - rows = rows[: params.limit] - has_previous = params.cursor is not None + scanned_rows, has_more = take_cursor_page(rows, limit=params.limit) # Generate cursors with sort column info for proper pagination - next_cursor = None - prev_cursor = None - - if rows: - if has_more: - # Generate next cursor from the last item - last_item = rows[-1] - next_cursor = BaseCursorPaginator.encode_cursor( - last_item["id"], - sort_column=sort_column, - sort_value=last_item.get(sort_column), - ) - - if params.cursor: - # If we used a cursor to get here, we can go back - first_item = rows[0] - prev_cursor = BaseCursorPaginator.encode_cursor( - first_item["id"], - sort_column=sort_column, - sort_value=first_item.get(sort_column), - ) - - # If we were doing reverse pagination, swap the cursors and reverse items - if params.reverse: - rows = list(reversed(rows)) - next_cursor, prev_cursor = prev_cursor, next_cursor - has_more, has_previous = has_previous, has_more + page = build_cursor_page( + scanned_rows, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + encode_cursor=lambda row: BaseCursorPaginator.encode_cursor( + row["id"], + sort_column=sort_column, + sort_value=row.get(sort_column), + ), + ) return CursorPaginatedResponse( - items=rows, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=has_previous, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def batch_insert_rows( diff --git a/tracecat/workflow/management/management.py b/tracecat/workflow/management/management.py index 821b9b3463..c1ca7c42fb 100644 --- a/tracecat/workflow/management/management.py +++ b/tracecat/workflow/management/management.py @@ -56,6 +56,8 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + build_cursor_page, + take_cursor_page, ) from tracecat.registry.lock.service import RegistryLockService from tracecat.registry.lock.types import RegistryLock @@ -522,12 +524,7 @@ async def list_workflows( stmt = stmt.options(selectinload(Workflow.tags)) results = await self.session.execute(stmt) - raw_items = list(results.all()) - - # Check if there are more items - has_more = len(raw_items) > params.limit - if has_more: - raw_items = raw_items[: params.limit] + raw_items, has_more = take_cursor_page(list(results.all()), limit=params.limit) # Process results into the expected format items = [] @@ -560,37 +557,25 @@ async def list_workflows( items.append((workflow, latest_defn, trigger_summary)) # Generate cursors - next_cursor = None - prev_cursor = None - - if items: - if has_more: - last_workflow = items[-1][0] # Get the workflow from the tuple - next_cursor = paginator.encode_cursor( - last_workflow.id, - sort_column="created_at", - sort_value=last_workflow.created_at, - ) - - if params.cursor: - first_workflow = items[0][0] # Get the workflow from the tuple - prev_cursor = paginator.encode_cursor( - first_workflow.id, - sort_column="created_at", - sort_value=first_workflow.created_at, - ) - - # If we were doing reverse pagination, swap the cursors and reverse items - if params.reverse: - items = list(reversed(items)) - next_cursor, prev_cursor = prev_cursor, next_cursor + page = build_cursor_page( + items, + cursor=params.cursor, + reverse=params.reverse, + has_more=has_more, + # Get the workflow from the tuple + encode_cursor=lambda item: paginator.encode_cursor( + item[0].id, + sort_column="created_at", + sort_value=item[0].created_at, + ), + ) return CursorPaginatedResponse( - items=items, - next_cursor=next_cursor, - prev_cursor=prev_cursor, - has_more=has_more, - has_previous=params.cursor is not None, + items=page.items, + next_cursor=page.next_cursor, + prev_cursor=page.prev_cursor, + has_more=page.has_more, + has_previous=page.has_previous, ) async def get_workflow( From dced07ba5f2ce608f6cf5d258123e483b77f2f75 Mon Sep 17 00:00:00 2001 From: Steve McMaster Date: Mon, 3 Aug 2026 16:45:25 -0400 Subject: [PATCH 2/2] fix(pagination): follow NULL sort cursors and guard inbox cross-sort Addresses review feedback on #3131. Table rows sort on user-defined columns, which are nullable by default, so list_rows legitimately encodes a cursor whose sort value is NULL whenever a page boundary lands inside the NULL block. Treating that server-issued cursor as malformed returned 400 on page 2 and made every later row unreachable. CursorData.has_sort_value now distinguishes an explicitly serialized sort_value: null from a legacy cursor that omits the key, and validate_cursor_sort_column grows allow_null for columns that admit NULLs. The new keyset_filter builds the matching predicate: NULL placement follows PostgreSQL's defaults, which invert consistently (ASC NULLS LAST reversed is DESC NULLS FIRST), so an ascending scan treats NULL as greater than every value and a descending scan as less, and only the id tie-breaker orders rows inside the NULL block. list_rows uses it, collapsing four duplicated cursor branches into one scan direction and spelling out NULLS FIRST/LAST in ORDER BY so the ordering matches the predicate. NOT NULL columns pass nullable=False and keep their original two-branch predicate. The agent runs inbox provider validated no cursor at all, so a created_at anchor submitted with order_by=updated_at compiled into a predicate against the wrong column and could skip or duplicate rows. Both the grouped and ungrouped decode paths now validate before applying the keyset predicate. The ungrouped path also normalizes sort_col to the two columns the query can actually order by, matching the grouped path, so an unrecognized order_by cannot mint a cursor labelled with a column the scan never sorted on. Co-Authored-By: Claude Opus 5 (1M context) --- .../tracecat_ee/inbox/providers/agent_runs.py | 22 +- tests/unit/test_cursor_page.py | 188 ++++++++++++++++++ tests/unit/test_inbox_agent_runs_cursor.py | 83 ++++++++ tests/unit/test_tables_service.py | 77 ++++++- tracecat/pagination.py | 104 +++++++++- tracecat/tables/service.py | 111 +++++------ 6 files changed, 514 insertions(+), 71 deletions(-) create mode 100644 tests/unit/test_inbox_agent_runs_cursor.py diff --git a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py b/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py index 2e50a4b8e3..59ae936b60 100644 --- a/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py +++ b/packages/tracecat-ee/tracecat_ee/inbox/providers/agent_runs.py @@ -28,6 +28,7 @@ CursorPaginatedResponse, build_cursor_page, take_cursor_page, + validate_cursor_sort_column, ) from tracecat_ee.agent.types import AgentWorkflowID @@ -296,8 +297,10 @@ async def list_items( updated_after=updated_after, ) - # Determine sort column and direction - sort_col = order_by or "created_at" + # Determine sort column and direction. Normalize to the two columns the + # query can actually order by, so an unrecognized `order_by` cannot mint + # a cursor labelled with a column the scan never sorted on. + sort_col = "updated_at" if order_by == "updated_at" else "created_at" sort_desc = sort != "asc" # Scan in the direction that walks toward the rows adjacent to the # cursor. Reverse pagination flips the scan so the LIMIT keeps the rows @@ -317,7 +320,12 @@ async def list_items( if cursor: try: cursor_data = self.decode_cursor(cursor) - cursor_value = cursor_data.sort_value + # A cursor from a different sort cannot filter this query: + # applying a created_at anchor to updated_at skips or duplicates + # rows. Reject it instead of silently mis-filtering. + cursor_value = validate_cursor_sort_column( + cursor_data, sort_column=sort_col, expected_type=datetime + ) cursor_id = uuid.UUID(cursor_data.id) # Select the correct column based on sort_col @@ -591,7 +599,13 @@ async def _list_items_grouped( if cursor: try: cursor_data = self.decode_cursor(cursor) - last_key = (cursor_data.sort_value, uuid.UUID(cursor_data.id)) + # Same cross-sort guard as the ungrouped path: a created_at + # anchor applied to an updated_at scan resumes at the wrong + # position and skips or repeats sessions. + scan_value = validate_cursor_sort_column( + cursor_data, sort_column=sort_col, expected_type=datetime + ) + last_key = (scan_value, uuid.UUID(cursor_data.id)) except (ValueError, KeyError) as e: # Surface malformed cursors as a client error so the router can # return 400 instead of silently restarting the scan. diff --git a/tests/unit/test_cursor_page.py b/tests/unit/test_cursor_page.py index 883e5511ca..b758db9fab 100644 --- a/tests/unit/test_cursor_page.py +++ b/tests/unit/test_cursor_page.py @@ -11,11 +11,14 @@ from datetime import datetime import pytest +import sqlalchemy as sa from tracecat.pagination import ( + BaseCursorPaginator, CursorData, InvalidCursorError, build_cursor_page, + keyset_filter, take_cursor_page, validate_cursor_sort_column, ) @@ -199,11 +202,196 @@ def test_validate_cursor_sort_column_rejects_a_wrongly_typed_sort_value() -> Non validate_cursor_sort_column(cursor, sort_column="priority", expected_type=int) +def test_validate_cursor_sort_column_accepts_an_explicit_null_when_allowed() -> None: + """A nullable sort column can legitimately anchor a page on a NULL value.""" + cursor = BaseCursorPaginator.decode_cursor( + BaseCursorPaginator.encode_cursor("abc", sort_column="score", sort_value=None) + ) + + assert cursor.has_sort_value is True + assert ( + validate_cursor_sort_column(cursor, sort_column="score", allow_null=True) + is None + ) + + +def test_validate_cursor_sort_column_rejects_a_null_by_default() -> None: + """Sorts on NOT NULL columns keep treating a NULL anchor as unusable.""" + cursor = BaseCursorPaginator.decode_cursor( + BaseCursorPaginator.encode_cursor( + "abc", sort_column="created_at", sort_value=None + ) + ) + + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="created_at") + + +def test_validate_cursor_sort_column_rejects_an_omitted_sort_value_when_null_allowed() -> ( + None +): + """An absent field is a legacy cursor, not a row whose sort value is NULL.""" + cursor = CursorData.model_validate({"id": "abc", "sort_column": "score"}) + + assert cursor.has_sort_value is False + with pytest.raises(InvalidCursorError, match="missing a sort value"): + validate_cursor_sort_column(cursor, sort_column="score", allow_null=True) + + def test_invalid_cursor_error_is_a_value_error() -> None: """Routers map ValueError to 400; the subclass keeps that mapping.""" assert issubclass(InvalidCursorError, ValueError) +# Keyset filtering over a nullable sort column, exercised on in-memory SQLite. +# NULL placement is pinned with explicit NULLS FIRST/LAST, so the ordering +# matches PostgreSQL's defaults and the predicate is checked against real rows +# rather than a rendered SQL string. +_KEYSET_TABLE = sa.Table( + "keyset_rows", + sa.MetaData(), + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("score", sa.Integer, nullable=True), +) +# Ids paired with a nullable sort value, deliberately interleaving NULLs with +# ties so both the tie-breaker and the NULL/non-NULL transition are crossed. +_KEYSET_ROWS = [ + (1, None), + (2, 10), + (3, None), + (4, 5), + (5, 10), + (6, None), + (7, 1), + (8, 5), +] +# (score ASC NULLS LAST, id ASC). The descending display order is its reverse. +_ASCENDING_IDS = [7, 4, 8, 2, 5, 1, 3, 6] + + +@pytest.fixture +def keyset_conn(): + engine = sa.create_engine("sqlite://") + with engine.begin() as conn: + _KEYSET_TABLE.create(conn) + conn.execute( + _KEYSET_TABLE.insert(), + [{"id": id_, "score": score} for id_, score in _KEYSET_ROWS], + ) + with engine.connect() as conn: + yield conn + engine.dispose() + + +def _keyset_scan(conn, *, cursor: str | None, ascending: bool, limit: int): + """Run one over-fetched keyset scan, mirroring what the services build.""" + stmt = sa.select(_KEYSET_TABLE.c.id, _KEYSET_TABLE.c.score) + if cursor is not None: + cursor_data = BaseCursorPaginator.decode_cursor(cursor) + stmt = stmt.where( + keyset_filter( + _KEYSET_TABLE.c.score, + _KEYSET_TABLE.c.id, + sort_value=validate_cursor_sort_column( + cursor_data, sort_column="score", allow_null=True + ), + id_value=int(cursor_data.id), + ascending=ascending, + ) + ) + sort_order = ( + _KEYSET_TABLE.c.score.asc().nulls_last() + if ascending + else _KEYSET_TABLE.c.score.desc().nulls_first() + ) + id_order = _KEYSET_TABLE.c.id.asc() if ascending else _KEYSET_TABLE.c.id.desc() + return list(conn.execute(stmt.order_by(sort_order, id_order).limit(limit + 1))) + + +def _keyset_page( + conn, *, cursor: str | None, ascending: bool, limit: int, reverse: bool = False +): + """One page in display order, following the cursor contract end to end.""" + scanned = _keyset_scan( + conn, cursor=cursor, ascending=ascending != reverse, limit=limit + ) + rows, has_more = take_cursor_page(scanned, limit=limit) + return build_cursor_page( + rows, + cursor=cursor, + reverse=reverse, + has_more=has_more, + encode_cursor=lambda row: BaseCursorPaginator.encode_cursor( + row.id, sort_column="score", sort_value=row.score + ), + ) + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_walks_every_row_across_the_null_boundary( + keyset_conn, ascending: bool +) -> None: + """Regression: a page anchored on a NULL sort value used to be unreachable. + + Nullable columns let a page boundary land inside the NULL block, so the + server issues a cursor whose sort value is NULL. Following it must resume + inside that block rather than reject the cursor or restart the scan. + """ + expected = _ASCENDING_IDS if ascending else list(reversed(_ASCENDING_IDS)) + + seen: list[int] = [] + cursor: str | None = None + while True: + page = _keyset_page(keyset_conn, cursor=cursor, ascending=ascending, limit=2) + seen.extend(row.id for row in page.items) + if not page.has_more: + break + cursor = page.next_cursor + + assert seen == expected + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_reverses_across_the_null_boundary( + keyset_conn, ascending: bool +) -> None: + """Paging back from a NULL-adjacent page must land on the page before it.""" + expected = _ASCENDING_IDS if ascending else list(reversed(_ASCENDING_IDS)) + + page_1 = _keyset_page(keyset_conn, cursor=None, ascending=ascending, limit=3) + page_2 = _keyset_page( + keyset_conn, cursor=page_1.next_cursor, ascending=ascending, limit=3 + ) + assert [row.id for row in page_2.items] == expected[3:6] + + back = _keyset_page( + keyset_conn, + cursor=page_2.prev_cursor, + ascending=ascending, + limit=3, + reverse=True, + ) + + assert [row.id for row in back.items] == expected[:3] + + +@pytest.mark.parametrize("ascending", [True, False]) +def test_keyset_filter_omits_null_branches_for_not_null_columns( + ascending: bool, +) -> None: + """NOT NULL sorts keep the tight two-branch predicate they had before.""" + predicate = keyset_filter( + _KEYSET_TABLE.c.score, + _KEYSET_TABLE.c.id, + sort_value=5, + id_value=4, + ascending=ascending, + nullable=False, + ) + + assert "NULL" not in str(predicate.compile(compile_kwargs={"literal_binds": True})) + + @pytest.mark.parametrize("reverse", [False, True]) def test_build_cursor_page_does_not_mutate_the_input(reverse: bool) -> None: rows = list(ROWS[:5]) diff --git a/tests/unit/test_inbox_agent_runs_cursor.py b/tests/unit/test_inbox_agent_runs_cursor.py new file mode 100644 index 0000000000..e4d6e7b225 --- /dev/null +++ b/tests/unit/test_inbox_agent_runs_cursor.py @@ -0,0 +1,83 @@ +"""Cursor-contract tests for the agent runs inbox provider. + +The provider paginates on either ``created_at`` or ``updated_at``. Applying a +cursor minted for one column to a scan ordered by the other resumes at the +wrong keyset position, which silently skips or repeats sessions, so both the +ungrouped and grouped decode paths must reject a cross-sort cursor. + +These exercise cursor validation only, which happens before any query runs, so +the session is a stub. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import MagicMock +from uuid import uuid4 + +import pytest +from tracecat_ee.inbox.providers.agent_runs import AgentRunsInboxProvider + +from tracecat.auth.types import Role +from tracecat.inbox.types import InboxGroup +from tracecat.pagination import BaseCursorPaginator + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def provider() -> AgentRunsInboxProvider: + role = Role( + type="user", workspace_id=uuid4(), user_id=uuid4(), service_id="tracecat-api" + ) + return AgentRunsInboxProvider(session=MagicMock(), role=role) + + +def _cursor(sort_column: str) -> str: + return BaseCursorPaginator.encode_cursor( + uuid4(), + sort_column=sort_column, + sort_value=datetime(2026, 1, 1, tzinfo=UTC), + ) + + +async def test_list_items_rejects_a_cursor_from_a_different_sort( + provider: AgentRunsInboxProvider, +) -> None: + with pytest.raises(ValueError, match="sorts by 'updated_at'"): + await provider.list_items(cursor=_cursor("created_at"), order_by="updated_at") + + +async def test_grouped_list_items_rejects_a_cursor_from_a_different_sort( + provider: AgentRunsInboxProvider, +) -> None: + with pytest.raises(ValueError, match="sorts by 'updated_at'"): + await provider.list_items( + cursor=_cursor("created_at"), + order_by="updated_at", + group=InboxGroup.RUNNING, + ) + + +@pytest.mark.parametrize("group", [None, InboxGroup.RUNNING]) +async def test_list_items_rejects_a_sortless_cursor( + provider: AgentRunsInboxProvider, group: InboxGroup | None +) -> None: + """Cursors predating sort-aware pagination carry no sort column.""" + cursor = BaseCursorPaginator.encode_cursor(uuid4()) + + with pytest.raises(ValueError, match="Cursor was created for sort column None"): + await provider.list_items(cursor=cursor, order_by="created_at", group=group) + + +@pytest.mark.parametrize("group", [None, InboxGroup.RUNNING]) +async def test_list_items_rejects_a_non_datetime_sort_value( + provider: AgentRunsInboxProvider, group: InboxGroup | None +) -> None: + """Both columns are timestamps; anything else cannot filter the keyset.""" + cursor = BaseCursorPaginator.encode_cursor( + uuid4(), sort_column="created_at", sort_value=7 + ) + + with pytest.raises(ValueError, match="wrong type"): + await provider.list_items(cursor=cursor, order_by="created_at", group=group) diff --git a/tests/unit/test_tables_service.py b/tests/unit/test_tables_service.py index 52313eb8af..14cbf144c5 100644 --- a/tests/unit/test_tables_service.py +++ b/tests/unit/test_tables_service.py @@ -1,7 +1,7 @@ from collections.abc import Iterator from datetime import UTC, date, datetime, timedelta, timezone from decimal import Decimal -from typing import Any +from typing import Any, Literal from uuid import UUID, uuid4 import pytest @@ -17,7 +17,11 @@ from tracecat.db.models import Table, TableColumn, Workspace from tracecat.exceptions import TracecatAuthorizationError, TracecatNotFoundError from tracecat.logger import logger -from tracecat.pagination import CursorPaginationParams, InvalidCursorError +from tracecat.pagination import ( + BaseCursorPaginator, + CursorPaginationParams, + InvalidCursorError, +) from tracecat.tables.common import ( ColumnHasDuplicateValuesError, handle_default_value, @@ -1494,6 +1498,75 @@ async def test_list_rows_rejects_a_cursor_from_a_different_sort( sort="asc", ) + @pytest.mark.parametrize("sort", ["asc", "desc"]) + async def test_list_rows_follows_a_null_sort_value_cursor( + self, + tables_service: TablesService, + table: Table, + sort: Literal["asc", "desc"], + ) -> None: + """A page boundary inside the NULL block must stay followable. + + Custom columns are nullable by default, so ``list_rows`` legitimately + encodes ``sort_value=None`` whenever a page ends on an unset value. + Rejecting that server-issued cursor made every row after the first page + unreachable. + """ + unset_ids = set() + for i in range(3): + row = await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Unset{i}"}) + ) + unset_ids.add(str(row["id"])) + # age 0 also guards the NULL check against a falsy-value regression. + set_ids = set() + for i in range(2): + row = await tables_service.insert_row( + table, TableRowInsert(data={"name": f"Set{i}", "age": i}) + ) + set_ids.add(str(row["id"])) + + async def page(cursor: str | None, *, reverse: bool = False): + return await tables_service.list_rows( + table, + params=CursorPaginationParams(limit=2, cursor=cursor, reverse=reverse), + order_by="age", + sort=sort, + ) + + seen: list[str] = [] + pages: list[list[str]] = [] + null_anchored = False + cursor: str | None = None + while True: + result = await page(cursor) + ids = [str(row["id"]) for row in result.items] + pages.append(ids) + seen.extend(ids) + if not result.has_more: + break + assert result.next_cursor is not None + anchor = BaseCursorPaginator.decode_cursor(result.next_cursor) + assert anchor.sort_column == "age" + null_anchored |= anchor.has_sort_value and anchor.sort_value is None + cursor = result.next_cursor + assert len(pages) <= 5, "pagination did not advance" + + assert null_anchored, "expected a page boundary anchored on a NULL sort value" + assert seen == list(dict.fromkeys(seen)) + assert set(seen) == unset_ids | set_ids + + # PostgreSQL sorts NULLs last ascending and first descending, so the + # unset rows form a contiguous block at one end of the walk. + assert set(seen[-3:] if sort == "asc" else seen[:3]) == unset_ids + + # Paging back must land on the preceding page rather than restarting. + page_1 = await page(None) + page_2 = await page(page_1.next_cursor) + assert page_2.prev_cursor is not None + back = await page(page_2.prev_cursor, reverse=True) + assert [str(row["id"]) for row in back.items] == pages[0] + async def test_table_editor_list_rows_reverse_pagination_flags( self, tables_service: TablesService, table: Table ) -> None: diff --git a/tracecat/pagination.py b/tracecat/pagination.py index 40ec3778bd..a4bc292740 100644 --- a/tracecat/pagination.py +++ b/tracecat/pagination.py @@ -5,7 +5,7 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass from datetime import datetime -from typing import TypeVar +from typing import Any, Literal, TypeVar, overload from uuid import UUID import sqlalchemy as sa @@ -162,13 +162,47 @@ def parse_datetime_string( return v return v + @property + def has_sort_value(self) -> bool: + """Whether the cursor carries an explicit sort value. + + ``encode_cursor`` always serializes ``sort_value``, so a cursor issued + by this server has the key even when the anchor row's sort column is + NULL. An absent key means a cursor from before sort-aware pagination + (or a hand-built one), which cannot filter anything. Keeping the two + apart is what lets a nullable sort follow its own ``sort_value: null`` + cursor instead of rejecting it as malformed. + """ + return "sort_value" in self.model_fields_set + + +@overload +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = ..., + allow_null: Literal[False] = ..., +) -> str | int | float | datetime: ... + + +@overload +def validate_cursor_sort_column( + cursor: CursorData, + *, + sort_column: str, + expected_type: type | tuple[type, ...] | None = ..., + allow_null: Literal[True], +) -> str | int | float | datetime | None: ... + def validate_cursor_sort_column( cursor: CursorData, *, sort_column: str, expected_type: type | tuple[type, ...] | None = None, -) -> str | int | float | datetime: + allow_null: bool = False, +) -> str | int | float | datetime | None: """Return the cursor's sort value, or raise if it cannot filter this query. Args: @@ -177,9 +211,15 @@ def validate_cursor_sort_column( expected_type: Type the sort value must have, when the column's cursor representation is narrower than the query's column type (enum sorts encode a rank, not the enum value). + allow_null: Whether the sort column admits NULLs. When set, a cursor + anchored on a NULL sort value returns ``None`` instead of raising, + and the caller must pair it with a NULL-aware keyset predicate (see + ``keyset_filter``). Legacy cursors that omit the field entirely are + still rejected. Raises: - InvalidCursorError: The cursor belongs to a different sort. + InvalidCursorError: The cursor belongs to a different sort, carries no + usable sort value, or carries one of the wrong type. """ if cursor.sort_column != sort_column: raise InvalidCursorError( @@ -188,6 +228,8 @@ def validate_cursor_sort_column( "Restart pagination without a cursor after changing the sort." ) if cursor.sort_value is None: + if allow_null and cursor.has_sort_value: + return None raise InvalidCursorError( f"Cursor is missing a sort value for column {sort_column!r}. " "Restart pagination without a cursor." @@ -200,6 +242,62 @@ def validate_cursor_sort_column( return cursor.sort_value +def keyset_filter( + sort_col: sa.ColumnElement[Any], + id_col: sa.ColumnElement[Any], + *, + sort_value: str | int | float | datetime | None, + id_value: Any, + ascending: bool, + nullable: bool = True, +) -> sa.ColumnElement[bool]: + """Build the predicate selecting rows strictly after a cursor anchor. + + Args: + sort_col: The column (or expression) the query sorts by. + id_col: The unique tie-breaker column, sorted alongside ``sort_col``. + sort_value: The anchor row's sort value; ``None`` anchors inside the + NULL block. + id_value: The anchor row's tie-breaker value. + ascending: The direction of the *scan*, not of the requested sort. + Reverse pagination inverts the scan so ``LIMIT`` keeps the rows + nearest the cursor, and this predicate must follow that inversion. + nullable: Whether ``sort_col`` can hold NULLs. Pass ``False`` for + NOT NULL columns to keep the predicate free of NULL branches; such + a column can never produce a ``sort_value`` of ``None``. + + NULL placement follows PostgreSQL's defaults, which invert consistently: + ``ASC`` puts NULLs last and ``DESC`` puts them first, so ``ASC NULLS LAST`` + reversed is exactly ``DESC NULLS FIRST``. An ascending scan therefore + treats NULL as greater than every value, a descending scan treats it as + less than every value, and inside the NULL block only ``id_col`` orders + rows. Callers must order by the matching ``nulls_last()``/``nulls_first()``. + """ + if ascending: + if sort_value is None: + # NULLs sort last, so nothing outside the NULL block follows. + return sa.and_(sort_col.is_(None), id_col > id_value) + after = [sort_col > sort_value] + if nullable: + # NULLs sort after every value, and `>` is unknown against them. + after.append(sort_col.is_(None)) + after.append(sa.and_(sort_col == sort_value, id_col > id_value)) + return sa.or_(*after) + + if sort_value is None: + # NULLs sort first, so every non-NULL row follows the NULL block. + return sa.or_( + sort_col.is_not(None), + sa.and_(sort_col.is_(None), id_col < id_value), + ) + # NULLs sort before every value, so a non-NULL anchor already excludes + # them: `<` is unknown against NULL and drops those rows. + return sa.or_( + sort_col < sort_value, + sa.and_(sort_col == sort_value, id_col < id_value), + ) + + class BaseCursorPaginator: """Base class for cursor-based pagination.""" diff --git a/tracecat/tables/service.py b/tracecat/tables/service.py index 8b18a4e66b..b3937dec97 100644 --- a/tracecat/tables/service.py +++ b/tracecat/tables/service.py @@ -41,7 +41,9 @@ BaseCursorPaginator, CursorPaginatedResponse, CursorPaginationParams, + InvalidCursorError, build_cursor_page, + keyset_filter, take_cursor_page, validate_cursor_sort_column, ) @@ -1447,6 +1449,21 @@ async def list_rows( sort_col = sa.column(self._sanitize_identifier(sort_column)) + # Custom columns default to nullable, so a page boundary can legitimately + # land on a NULL sort value. System columns are NOT NULL by construction. + if sort_column in _TABLE_SYSTEM_COLUMNS: + sort_column_nullable = False + else: + column_meta = next( + (col for col in table.columns if col.name == sort_column), None + ) + sort_column_nullable = column_meta is None or column_meta.nullable + + # Reverse pagination scans away from the cursor in the inverted sort + # order so LIMIT keeps the rows nearest the cursor; build_cursor_page + # puts them back into display order. + scan_ascending = (sort_direction == "asc") != params.reverse + # Apply cursor-based pagination with sort-column-aware filtering if params.cursor: try: @@ -1458,77 +1475,47 @@ async def list_rows( # A cursor from a different sort cannot filter this query. Reject it # instead of dropping the filter, which would silently return the - # first page while still reporting has_previous. + # first page while still reporting has_previous. A NULL sort value + # is accepted for nullable columns: this service issues such cursors + # whenever a page boundary falls inside the NULL block. sort_cursor_value = validate_cursor_sort_column( - cursor_data, sort_column=sort_column + cursor_data, sort_column=sort_column, allow_null=True ) + if sort_cursor_value is None and not sort_column_nullable: + raise InvalidCursorError( + f"Cursor carries a NULL sort value for NOT NULL column " + f"{sort_column!r}. Restart pagination without a cursor." + ) # Composite filtering: (sort_col, id) matches ORDER BY - if sort_direction == "asc": - if params.reverse: - # Going backward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) - ) - else: - # Going forward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) - ) - else: - # Descending order - if params.reverse: - # Going backward: get records after cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col > sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") > cursor_id, - ), - ) - ) - else: - # Going forward: get records before cursor in sort order - stmt = stmt.where( - sa.or_( - sort_col < sort_cursor_value, - sa.and_( - sort_col == sort_cursor_value, - sa.column("id") < cursor_id, - ), - ) - ) + stmt = stmt.where( + keyset_filter( + sort_col, + sa.column("id"), + sort_value=sort_cursor_value, + id_value=cursor_id, + ascending=scan_ascending, + nullable=sort_column_nullable, + ) + ) - # Apply sorting: (sort_col, id) for stable pagination - # Reverse pagination scans away from the cursor in the inverted sort - # order so LIMIT keeps the rows nearest the cursor; build_cursor_page - # puts them back into display order. - scan_ascending = (sort_direction == "asc") != params.reverse + # Apply sorting: (sort_col, id) for stable pagination. NULL placement is + # spelled out so it matches the keyset predicate above rather than + # relying on the server default. + order_clause = ( + sort_col.asc().nulls_last() + if scan_ascending + else sort_col.desc().nulls_first() + ) if sort_column == "id": # No tie-breaker needed when sorting by id (already unique) - if scan_ascending: - stmt = stmt.order_by(sort_col.asc()) - else: - stmt = stmt.order_by(sort_col.desc()) + stmt = stmt.order_by(order_clause) else: # Add id as tie-breaker for non-unique columns - if scan_ascending: - stmt = stmt.order_by(sort_col.asc(), sa.column("id").asc()) - else: - stmt = stmt.order_by(sort_col.desc(), sa.column("id").desc()) + id_order = ( + sa.column("id").asc() if scan_ascending else sa.column("id").desc() + ) + stmt = stmt.order_by(order_clause, id_order) # Fetch limit + 1 to determine if there are more items stmt = stmt.limit(params.limit + 1)