[F106] compound (timestamp, id) keyset cursor for message pagination

get_messages used strict timestamp inequalities with a non-deterministic
order_by(timestamp.desc()), so equal-timestamp messages were cut by limit
on one page and excluded (strict < T / > T) from the next — they vanished
across pages. Bundled the (timestamp, id) pair into a MessageCursor dataclass
so the next page resumes exactly past the cursor's id at the shared
timestamp (or_: strictly-older OR same-timestamp-smaller-id for before; the
mirror for after), with a deterministic order_by(timestamp.desc(), id.desc())
so the last-item cursor is unambiguous. id is None for a legacy timestamp-
only cursor (strict inequality, prior behavior). The route builds cursors
from the flat before/before_id + after/after_id HTTP params; the schema now
carries the tie-breaker ids. Also clears PLR0913 (cursors replace the
before_id/after_id params).
This commit is contained in:
Renn F
2026-06-28 21:56:10 +02:00
parent c8bb16e634
commit b1d5dbb240
4 changed files with 201 additions and 19 deletions
+13 -2
View File
@@ -25,6 +25,7 @@ from roboco.services.messaging import (
MessageCreateRequest as ServiceMessageRequest, MessageCreateRequest as ServiceMessageRequest,
) )
from roboco.services.messaging import ( from roboco.services.messaging import (
MessageCursor,
get_messaging_service, get_messaging_service,
) )
@@ -45,10 +46,20 @@ async def list_messages(
"""List messages in a session (session existence is checked in the service).""" """List messages in a session (session existence is checked in the service)."""
messaging = get_messaging_service(db) messaging = get_messaging_service(db)
try: try:
before_cursor = (
MessageCursor(params.before, params.before_id)
if params.before is not None
else None
)
after_cursor = (
MessageCursor(params.after, params.after_id)
if params.after is not None
else None
)
messages, has_more = await messaging.list_messages_for_session( messages, has_more = await messaging.list_messages_for_session(
session_id=params.session_id, session_id=params.session_id,
before=params.before, before=before_cursor,
after=params.after, after=after_cursor,
message_type=params.type_filter, message_type=params.type_filter,
limit=params.limit, limit=params.limit,
) )
+6
View File
@@ -23,6 +23,12 @@ class ListMessagesParams(BaseModel):
session_id: UUID session_id: UUID
before: datetime | None = None before: datetime | None = None
after: datetime | None = None after: datetime | None = None
# Keyset-pagination tie-breakers: pass the last (before) / first (after)
# message's id alongside its timestamp so equal-timestamp messages are not
# skipped across pages. Optional — a timestamp-only cursor keeps the
# legacy strict-inequality behavior.
before_id: UUID | None = None
after_id: UUID | None = None
type_filter: MessageType | None = None type_filter: MessageType | None = None
limit: int = Field(50, ge=1, le=100) limit: int = Field(50, ge=1, le=100)
+71 -14
View File
@@ -16,7 +16,7 @@ from datetime import UTC, datetime, timedelta
from typing import Any, ClassVar, cast from typing import Any, ClassVar, cast
from uuid import UUID from uuid import UUID
from sqlalchemy import select from sqlalchemy import and_, or_, select
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload from sqlalchemy.orm import joinedload, selectinload
@@ -76,6 +76,23 @@ def _minutes_to_timedelta(value: int | None) -> timedelta | None:
# ============================================================================= # =============================================================================
@dataclass(frozen=True, slots=True)
class MessageCursor:
"""A keyset-pagination cursor over messages: a ``(timestamp, id)`` pair.
The ``id`` tie-breaks equal timestamps so paging across same-timestamp
messages skips nothing: the next page resumes past the cursor's id at
the shared timestamp rather than being excluded by a strict
``timestamp < cursor`` / ``timestamp > cursor`` inequality (which drops
every row sharing the cursor's timestamp — they are cut by ``limit`` on
the prior page and never returned). ``id`` is ``None`` for a legacy
timestamp-only cursor (strict inequality, the prior behavior).
"""
timestamp: datetime
id: UUID | None = None
class MessagingService(BaseService): class MessagingService(BaseService):
""" """
Service for managing all messaging operations. Service for managing all messaging operations.
@@ -1571,8 +1588,8 @@ class MessagingService(BaseService):
self, self,
*, *,
session_id: UUID, session_id: UUID,
before: datetime | None, before: MessageCursor | None,
after: datetime | None, after: MessageCursor | None,
message_type: MessageType | None, message_type: MessageType | None,
limit: int, limit: int,
) -> tuple[list[MessageTable], bool]: ) -> tuple[list[MessageTable], bool]:
@@ -1580,7 +1597,10 @@ class MessagingService(BaseService):
Existing `get_messages` skips the session-existence check — this Existing `get_messages` skips the session-existence check — this
variant raises NotFoundError when the session is missing so routes variant raises NotFoundError when the session is missing so routes
can return a clean 404 without issuing their own query. can return a clean 404 without issuing their own query. The ``before``
/ ``after`` cursors carry the keyset ``(timestamp, id)`` pair so
callers can paginate without skipping equal-timestamp messages across
pages.
""" """
await self.get_session_or_raise(session_id) await self.get_session_or_raise(session_id)
return await self.get_messages( return await self.get_messages(
@@ -1668,8 +1688,8 @@ class MessagingService(BaseService):
async def get_messages( async def get_messages(
self, self,
session_id: UUID, session_id: UUID,
before: datetime | None = None, before: MessageCursor | None = None,
after: datetime | None = None, after: MessageCursor | None = None,
message_type: MessageType | None = None, message_type: MessageType | None = None,
limit: int = 50, limit: int = 50,
) -> tuple[list[MessageTable], bool]: ) -> tuple[list[MessageTable], bool]:
@@ -1678,8 +1698,12 @@ class MessagingService(BaseService):
Args: Args:
session_id: Session to get messages from session_id: Session to get messages from
before: Get messages before this timestamp before: Keyset cursor — return messages older than (or, when the
after: Get messages after this timestamp cursor carries an id, equal-timestamp-but-smaller-id than) this
position. ``None`` for no upper bound.
after: Keyset cursor — return messages newer than (or
equal-timestamp-but-larger-id than) this position. ``None`` for
no lower bound.
message_type: Filter by message type message_type: Filter by message type
limit: Maximum messages to return limit: Maximum messages to return
@@ -1688,15 +1712,48 @@ class MessagingService(BaseService):
""" """
query = select(MessageTable).where(MessageTable.session_id == session_id) query = select(MessageTable).where(MessageTable.session_id == session_id)
if before: if before is not None:
query = query.where(MessageTable.timestamp < before) if before.id is not None:
if after: # Compound keyset cursor: resume past (before.timestamp,
query = query.where(MessageTable.timestamp > after) # before.id) in desc order. Rows where (timestamp, id) <
# (before.timestamp, before.id): strictly older, OR same
# timestamp with a smaller id. Without this tie-break,
# equal-timestamp rows cut by ``limit`` on the prior page are
# excluded by the strict ``< before.timestamp`` and vanish.
query = query.where(
or_(
MessageTable.timestamp < before.timestamp,
and_(
MessageTable.timestamp == before.timestamp,
MessageTable.id < before.id,
),
)
)
else:
query = query.where(MessageTable.timestamp < before.timestamp)
if after is not None:
if after.id is not None:
query = query.where(
or_(
MessageTable.timestamp > after.timestamp,
and_(
MessageTable.timestamp == after.timestamp,
MessageTable.id > after.id,
),
)
)
else:
query = query.where(MessageTable.timestamp > after.timestamp)
if message_type: if message_type:
query = query.where(MessageTable.type == message_type) query = query.where(MessageTable.type == message_type)
# Get one extra to check if there are more # Deterministic total order: timestamp desc, then id desc so the
query = query.order_by(MessageTable.timestamp.desc()).limit(limit + 1) # "last item" cursor is unambiguous for keyset pagination (without the
# id tie-break, same-timestamp rows have a non-deterministic order and
# the cursor is ambiguous).
query = query.order_by(
MessageTable.timestamp.desc(), MessageTable.id.desc()
).limit(limit + 1)
result = await self.session.execute(query) result = await self.session.execute(query)
messages = list(result.scalars().all()) messages = list(result.scalars().all())
+111 -3
View File
@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
from types import SimpleNamespace from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast from typing import TYPE_CHECKING, Any, cast
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, patch
from uuid import uuid4 from uuid import UUID, uuid4
from uuid import uuid4 as _u from uuid import uuid4 as _u
import pytest import pytest
@@ -38,6 +38,7 @@ from roboco.models.session import (
from roboco.services.base import ConflictError, NotFoundError from roboco.services.base import ConflictError, NotFoundError
from roboco.services.messaging import ( from roboco.services.messaging import (
ApiSessionCreate, ApiSessionCreate,
MessageCursor,
MessagingService, MessagingService,
get_messaging_service, get_messaging_service,
) )
@@ -1106,8 +1107,8 @@ async def test_get_messages_with_filters(msg_setup: dict) -> None:
cutoff = datetime.now(UTC) - timedelta(hours=1) cutoff = datetime.now(UTC) - timedelta(hours=1)
msgs, _ = await svc.get_messages( msgs, _ = await svc.get_messages(
sess.id, sess.id,
before=datetime.now(UTC) + timedelta(hours=1), before=MessageCursor(datetime.now(UTC) + timedelta(hours=1)),
after=cutoff, after=MessageCursor(cutoff),
message_type=MessageType.DIALOGUE, message_type=MessageType.DIALOGUE,
) )
assert isinstance(msgs, list) assert isinstance(msgs, list)
@@ -1130,6 +1131,113 @@ async def test_get_messages_with_limit(msg_setup: dict) -> None:
assert has_more is True assert has_more is True
async def _seed_messages_same_timestamp(
svc: MessagingService, session: Any, aid: UUID, count: int
) -> tuple[UUID, list[UUID]]:
"""Create ``count`` messages in one session and force them ALL to the same
timestamp so the equal-timestamp pagination skip is reproducible. Returns
``(session_id, message_ids)``."""
ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
sess = await svc.create_session(SessionCreateRequest(group_id=grp.id))
ids: list[UUID] = []
for i in range(count):
m = await svc.send_message(
MessageCreateRequest(agent_id=aid, session_id=sess.id, content=f"m-{i}")
)
ids.append(m.id)
fixed = datetime.now(UTC)
rows = (
(
await session.execute(
select(MessageTable).where(MessageTable.session_id == sess.id)
)
)
.scalars()
.all()
)
for row in rows:
row.timestamp = fixed
await session.flush()
return sess.id, ids
@pytest.mark.asyncio
async def test_get_messages_compound_before_cursor_no_skip_on_equal_timestamps(
msg_setup: dict,
) -> None:
"""Equal-timestamp messages must not be skipped across pages (F106).
With a strict ``timestamp < before`` cursor and ``order_by(timestamp.desc())``,
messages sharing the page's last timestamp are cut by ``limit`` on page 1
and excluded (``< T``) from page 2 — they vanish. The compound
``(timestamp, id)`` cursor tie-breaks on the unique id so the next page
resumes exactly past the last id (no skip, no duplicate). Requires the
deterministic ``order_by(timestamp.desc(), id.desc())`` so the "last item"
cursor is unambiguous.
"""
svc = msg_setup["svc"]
session = svc.session
aid = msg_setup["agent_id"]
total = 5
sess_id, ids = await _seed_messages_same_timestamp(svc, session, aid, count=total)
page_size = 3
page1, has_more = await svc.get_messages(sess_id, limit=page_size)
assert has_more is True
assert len(page1) == page_size
# Deterministic order: same timestamp → by id.desc()
page1_ids = [m.id for m in page1]
assert page1_ids == sorted(page1_ids, reverse=True)
last = page1[-1]
page2, has_more2 = await svc.get_messages(
sess_id, before=MessageCursor(last.timestamp, last.id), limit=page_size
)
# All 5 covered, no skip, no duplicate.
seen = {m.id for m in page1} | {m.id for m in page2}
assert seen == set(ids)
assert len(page1) + len(page2) == total
assert has_more2 is False
@pytest.mark.asyncio
async def test_get_messages_compound_after_cursor_no_skip_on_equal_timestamps(
msg_setup: dict,
) -> None:
"""Forward pagination (``after``) with the compound ``(timestamp, id)``
cursor tie-breaks on id so newer-direction pagination across equal
timestamps skips nothing either (F106).
With a strict ``timestamp > after`` cursor, every row sharing the cursor's
timestamp is EXCLUDED — so forward-paginating from a middle message would
return nothing (all rows are at the same timestamp, none strictly greater).
The compound cursor includes same-timestamp rows with a larger id.
"""
svc = msg_setup["svc"]
session = svc.session
aid = msg_setup["agent_id"]
total = 5
fetch_limit = 10
mid_index = 2
newer_count = 2
sess_id, _ids = await _seed_messages_same_timestamp(svc, session, aid, count=total)
# Full desc ordering: [id_largest, ..., id_smallest].
all_msgs, _ = await svc.get_messages(sess_id, limit=fetch_limit)
assert len(all_msgs) == total
# Pick the middle row as the forward cursor.
mid = all_msgs[mid_index]
newer = {m.id for m in all_msgs[:newer_count]} # the rows above mid (larger id)
page, _ = await svc.get_messages(
sess_id, after=MessageCursor(mid.timestamp, mid.id), limit=fetch_limit
)
# The same-timestamp rows newer than mid are returned — NOT skipped.
assert {m.id for m in page} == newer
assert len(page) == newer_count
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_message_context_redirects_when_session_closed( async def test_get_message_context_redirects_when_session_closed(
msg_setup: dict, msg_setup: dict,