Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 60 additions & 17 deletions backend/src/apis/shared/sessions/metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -1840,6 +1840,37 @@ def _encode_list_cursor(session: SessionMetadata) -> str:
return base64.b64encode(json.dumps(payload).encode('utf-8')).decode('utf-8')


# --- Issue #175 Phase 3: contract the dual-scheme read once migration completes ---
_MIGRATION_MARKER_PK = "MIGRATION#session-sk"
_MIGRATION_MARKER_SK = "STATE"
# Memoised once observed True. The marker only ever goes unset -> set (the Phase 2
# backfill sets it after confirming zero legacy rows), never back, so caching a True
# result per-process is safe. A False/absent result is NOT cached, so a container
# that started before the backfill picks up the flip on a later list call.
_migration_complete_cache = False


def _is_session_migration_complete(table) -> bool:
"""True once the Phase 2 backfill has set the migration-complete marker.

Gates ``list_user_sessions`` from the dual-scheme union down to a GSI-only read.
Fails open (returns False -> keep dual-read) on any error, and stays in dual-read
for downstream/forked deployments that haven't run the backfill — so removing the
legacy branch here can never blank the sidebar on un-migrated data.
"""
global _migration_complete_cache
if _migration_complete_cache:
return True
try:
resp = table.get_item(Key={"PK": _MIGRATION_MARKER_PK, "SK": _MIGRATION_MARKER_SK})
if (resp.get("Item") or {}).get("complete"):
_migration_complete_cache = True
return True
except Exception as e:
logger.debug("migration marker check failed (staying in dual-read): %s", e)
return False


async def _list_user_sessions_cloud(
user_id: str,
table_name: str,
Expand Down Expand Up @@ -1885,25 +1916,13 @@ async def _list_user_sessions_cloud(
want = (limit + 1) if limit else None
pk = f'USER#{user_id}'

# Legacy source: base table, S#ACTIVE# prefix. Resuming after a cursor uses
# between('S#ACTIVE#', 'S#ACTIVE#{la}#{sid}') — a single range condition that
# keeps the prefix filter while bounding the upper end (inclusive; the exact
# cursor row is dropped by the strict-less filter below).
if cursor:
la, sid = cursor
legacy_cond = Key('PK').eq(pk) & Key('SK').between('S#ACTIVE#', f'S#ACTIVE#{la}#{sid}')
else:
legacy_cond = Key('PK').eq(pk) & Key('SK').begins_with('S#ACTIVE#')
legacy_params: Dict[str, Any] = {
'KeyConditionExpression': legacy_cond,
'ScanIndexForward': False,
}
if want:
legacy_params['Limit'] = want
legacy_sessions = _collect_valid_sessions(table, legacy_params, want)
# GSI-only once the migration is complete (marker set); otherwise dual-read
# the union so un-migrated legacy rows stay visible — for downstream/forked
# deployments that haven't run the Phase 2 backfill yet.
gsi_only = _is_session_migration_complete(table)

# Migrated source: SessionRecencyIndex GSI. GSI4_SK < '{la}#{sid}' is a clean
# strict-less resume. Degrade to legacy-only if the index isn't there yet.
# strict-less resume.
if cursor:
la, sid = cursor
gsi_cond = Key('GSI4_PK').eq(pk) & Key('GSI4_SK').lt(f'{la}#{sid}')
Expand All @@ -1916,6 +1935,7 @@ async def _list_user_sessions_cloud(
}
if want:
gsi_params['Limit'] = want
gsi_failed = False
try:
gsi_sessions = _collect_valid_sessions(table, gsi_params, want)
except ClientError as e:
Expand All @@ -1936,9 +1956,32 @@ async def _list_user_sessions_cloud(
"session listing", code
)
gsi_sessions = []
gsi_failed = True
else:
raise

# Legacy source: base table, S#ACTIVE# prefix. Skipped once migration is
# complete (GSI-only), UNLESS the GSI query failed — then fall back to legacy
# regardless so a transient index error never blanks the list. Resuming after
# a cursor uses between('S#ACTIVE#', 'S#ACTIVE#{la}#{sid}') — a single range
# condition that keeps the prefix filter while bounding the upper end
# (inclusive; the exact cursor row is dropped by the strict-less filter below).
if gsi_only and not gsi_failed:
legacy_sessions: list[SessionMetadata] = []
else:
if cursor:
la, sid = cursor
legacy_cond = Key('PK').eq(pk) & Key('SK').between('S#ACTIVE#', f'S#ACTIVE#{la}#{sid}')
else:
legacy_cond = Key('PK').eq(pk) & Key('SK').begins_with('S#ACTIVE#')
legacy_params: Dict[str, Any] = {
'KeyConditionExpression': legacy_cond,
'ScanIndexForward': False,
}
if want:
legacy_params['Limit'] = want
legacy_sessions = _collect_valid_sessions(table, legacy_params, want)

# Merge: sort by (lastMessageAt, sessionId) descending, drop anything not
# strictly older than the cursor (removes the inclusive legacy cursor row),
# dedupe by session_id.
Expand Down
77 changes: 77 additions & 0 deletions backend/tests/shared/test_sessions_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -1215,3 +1215,80 @@ async def test_end_to_end_create_activity_list_delete(self, sessions_metadata_ta
await SessionService().delete_session("u1", "s1")
sessions, _ = await list_user_sessions("u1")
assert sessions == []


class TestListContractOnMarker:
"""Issue #175 Phase 3 — list read contracts to GSI-only once the marker is set."""

@staticmethod
def _set_marker(table):
table.put_item(Item={"PK": "MIGRATION#session-sk", "SK": "STATE", "complete": True})

@pytest.mark.asyncio
async def test_dual_read_when_marker_absent(self, sessions_metadata_table, monkeypatch):
import apis.shared.sessions.metadata as md
monkeypatch.setattr(md, "_migration_complete_cache", False)
_put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
_put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")

sessions, _ = await md.list_user_sessions("u1")
assert sorted(s.session_id for s in sessions) == ["leg", "mig"] # union of both

@pytest.mark.asyncio
async def test_gsi_only_when_marker_set(self, sessions_metadata_table, monkeypatch):
import apis.shared.sessions.metadata as md
monkeypatch.setattr(md, "_migration_complete_cache", False)
_put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
_put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")
self._set_marker(sessions_metadata_table)

sessions, _ = await md.list_user_sessions("u1")
assert [s.session_id for s in sessions] == ["mig"] # legacy row excluded

@pytest.mark.asyncio
async def test_marker_result_is_memoised(self, sessions_metadata_table, monkeypatch):
import apis.shared.sessions.metadata as md
monkeypatch.setattr(md, "_migration_complete_cache", False)
self._set_marker(sessions_metadata_table)
_put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")

await md.list_user_sessions("u1")
assert md._migration_complete_cache is True # cached after first observation

@pytest.mark.asyncio
async def test_gsi_failure_falls_back_to_legacy_even_with_marker(self, sessions_metadata_table, monkeypatch):
"""Marker set but the GSI query errors → still read legacy so the list never blanks."""
import boto3
import apis.shared.sessions.metadata as md
from botocore.exceptions import ClientError

monkeypatch.setattr(md, "_migration_complete_cache", False)
_put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
self._set_marker(sessions_metadata_table)
real_table = sessions_metadata_table

class _GsiFailingTable:
def get_item(self, **kw):
return real_table.get_item(**kw) # marker lookup goes through

def query(self, **kw):
if kw.get("IndexName") == "SessionRecencyIndex":
raise ClientError(
{"Error": {"Code": "ValidationException",
"Message": "The table does not have the specified index: SessionRecencyIndex"}},
"Query",
)
return real_table.query(**kw)

class _FakeResource:
def Table(self, _name):
return _GsiFailingTable()

real_resource = boto3.resource
monkeypatch.setattr(
boto3, "resource",
lambda svc, **kw: _FakeResource() if svc == "dynamodb" else real_resource(svc, **kw),
)

sessions, _ = await md.list_user_sessions("u1")
assert [s.session_id for s in sessions] == ["leg"] # fell back to legacy, not blank