Fix session retry and statistics release blockers

papyrus/schemas/tracking.py

@@ -127,6 +127,9 @@ def ordered(self) -> "PageCoverage":
 
 class Activity(TrackingPayload):
     session_id: UUID | None = None
+    start_position: float | None = Field(default=None, ge=0, le=1)
+    end_position: float | None = Field(default=None, ge=0, le=1)
+    device_type: str | None = None
     book_id: UUID
     book_title: str = Field(min_length=1, max_length=500)
     start_time: AwareDatetime

papyrus/services/progress.py

@@ -1,5 +1,6 @@
 """Real activity-backed sessions and cross-device statistics."""
 
+from collections import defaultdict
 from datetime import UTC, date, datetime, time, timedelta
 from uuid import UUID, uuid4
 
@@ -27,7 +28,7 @@
 from papyrus.services.goal_progress import effective_activities, project_goal, union_length
 from papyrus.services.goals import activities
 from papyrus.services.library_sync import owned_row
-from papyrus.services.sync import apply_powersync_upload_batch
+from papyrus.services.sync import apply_powersync_upload_batch, lock_owner_mutations
 
 
 def session_response(activity: Activity) -> ReadingSession:
@@ -39,7 +40,9 @@ def session_response(activity: Activity) -> ReadingSession:
         end_time=activity.end_time,
         pages_read=activity.pages,
         duration_minutes=int((activity.end_time - activity.start_time).total_seconds()) // 60,
-        device_type=activity.source,
+        start_position=activity.start_position,
+        end_position=activity.end_position,
+        device_type=activity.device_type if activity.device_type is not None else activity.source,
         device_name=activity.device_id,
         created_at=activity.created_at,
     )
@@ -55,9 +58,11 @@ def grouped_sessions(entries: list[Activity]) -> list[ReadingSession]:
 
     for group in groups.values():
         first = min(group, key=lambda entry: entry.start_time)
-        end = max(entry.end_time for entry in group)
+        last = max(group, key=lambda entry: entry.end_time)
+        end = last.end_time
         result = session_response(first)
         result.end_time = end
+        result.end_position = last.end_position
 
         result.duration_minutes = (
             int(union_length([(entry.start_time.timestamp(), entry.end_time.timestamp()) for entry in group])) // 60
@@ -116,6 +121,8 @@ async def list_sessions(
 
 
 async def create_session(session: AsyncSession, user_id: UUID, request: CreateReadingSessionRequest) -> ReadingSession:
+    await lock_owner_mutations(session, user_id)
+
     book = await owned_row(
         session,
         user_id,
@@ -146,6 +153,9 @@ async def create_session(session: AsyncSession, user_id: UUID, request: CreateRe
                 original.book_id != request.book_id
                 or original.start_time != start
                 or (requested_end is not None and original.end_time != requested_end)
+                or original.start_position != request.start_position
+                or original.end_position != request.end_position
+                or original.device_type != request.device_type
                 or original.pages != (request.pages_read or 0)
                 or original.device_id != (request.device_name or "manual")
             ):
@@ -173,6 +183,9 @@ async def create_session(session: AsyncSession, user_id: UUID, request: CreateRe
             start_time=start,
             end_time=end,
             created_at=datetime.now(UTC),
+            start_position=request.start_position,
+            end_position=request.end_position,
+            device_type=request.device_type,
             pages=request.pages_read or 0,
             device_id=request.device_name or "manual",
             shelf_ids=shelves,
@@ -211,6 +224,61 @@ def totals_definition(start: datetime, end: datetime, book_id: UUID | None = Non
     )
 
 
+def overlaps_period(activity: Activity, start: datetime, end: datetime) -> bool:
+    if activity.start_time == activity.end_time:
+        return start <= activity.start_time < end
+
+    return activity.start_time < end and activity.end_time > start
+
+
+def current_reading_streak(ledger: list[Activity], now: datetime) -> int:
+    """Union complete day ranges and partial boundary days without walking long activities."""
+    boundaries: dict[date, list[tuple[float, float]]] = defaultdict(list)
+    qualified_ranges: list[tuple[int, int]] = []
+
+    for activity in ledger:
+        if activity.kind != "reading":
+            continue
+
+        start = activity.start_time.astimezone(UTC)
+        end = min(activity.end_time, now).astimezone(UTC)
+
+        if start >= end:
+            continue
+
+        if start.date() == end.date():
+            boundaries[start.date()].append((start.timestamp(), end.timestamp()))
+            continue
+
+        next_day = datetime.combine(start.date() + timedelta(days=1), time.min, UTC)
+        last_day = datetime.combine(end.date(), time.min, UTC)
+        boundaries[start.date()].append((start.timestamp(), next_day.timestamp()))
+        boundaries[end.date()].append((last_day.timestamp(), end.timestamp()))
+        first_complete, last_complete = start.date().toordinal() + 1, end.date().toordinal() - 1
+
+        if first_complete <= last_complete:
+            qualified_ranges.append((first_complete, last_complete))
+
+    for day, intervals in boundaries.items():
+        if union_length(intervals) >= 300:
+            qualified_ranges.append((day.toordinal(), day.toordinal()))
+
+    merged: list[tuple[int, int]] = []
+
+    for first, last in sorted(qualified_ranges):
+        if merged and first <= merged[-1][1] + 1:
+            merged[-1] = (merged[-1][0], max(merged[-1][1], last))
+        else:
+            merged.append((first, last))
+
+    today = now.astimezone(UTC).date().toordinal()
+
+    if merged and merged[-1][1] in {today, today - 1}:
+        return merged[-1][1] - merged[-1][0] + 1
+
+    return 0
+
+
 async def statistics(
     session: AsyncSession, user_id: UUID, start_date: date | None, end_date: date | None
 ) -> ReadingStatistics:
@@ -227,22 +295,17 @@ async def statistics(
     total = project_goal(totals_definition(start, end), ledger, now)
     real = effective_activities(ledger)
 
-    counted = [
-        entry for entry in real if entry.kind == "reading" and entry.start_time < end and entry.end_time >= start
-    ]
+    counted = [entry for entry in real if entry.kind == "reading" and overlaps_period(entry, start, end)]
 
     daily = []
-    reading_dates = []
     longest = streak = 0
 
     for index in range((last - first).days + 1):
         day = start + timedelta(days=index)
         next_day = day + timedelta(days=1)
         values = project_goal(totals_definition(day, next_day), ledger, now)
 
-        count = len(
-            {entry.session_id or entry.id for entry in counted if entry.start_time < next_day and entry.end_time > day}
-        )
+        count = len({entry.session_id or entry.id for entry in counted if overlaps_period(entry, day, next_day)})
 
         daily.append(
             DailyBreakdown(
@@ -254,19 +317,11 @@ async def statistics(
         )
 
         if values["seconds"] >= 300:
-            reading_dates.append(day.date())
             streak += 1
             longest = max(longest, streak)
         else:
             streak = 0
 
-    current = 0
-    cursor = today if today in reading_dates else today - timedelta(days=1)
-
-    while cursor in reading_dates:
-        current += 1
-        cursor -= timedelta(days=1)
-
     books = {entry.book_id: entry.book_title for entry in counted}
     breakdown = []
 
@@ -294,7 +349,7 @@ async def statistics(
             sessions_count=sessions_count,
             average_session_minutes=total["seconds"] / 60 / sessions_count if sessions_count else 0,
             reading_days=total["days"],
-            current_streak=current,
+            current_streak=current_reading_streak(real, now),
             longest_streak=longest,
         ),
         daily_breakdown=daily,

papyrus/services/sync.py

@@ -12,6 +12,11 @@
 from papyrus.services.library_sync import apply_library_mutation
 
 
+async def lock_owner_mutations(session: AsyncSession, user_id: UUID) -> None:
+    """Hold the owner's mutation lock until the surrounding transaction ends."""
+    await session.execute(select(User.user_id).where(User.user_id == user_id).with_for_update())
+
+
 async def apply_powersync_upload_batch(
     session: AsyncSession,
     user_id: UUID,
@@ -22,7 +27,7 @@ async def apply_powersync_upload_batch(
     media_paths_to_delete: list[Path] = []
 
     try:
-        await session.execute(select(User.user_id).where(User.user_id == user_id).with_for_update())
+        await lock_owner_mutations(session, user_id)
 
         for mutation in batch:
             applied, deleted_media_paths = await apply_library_mutation(session, user_id, mutation)

tests/api/routes/test_progress.py

@@ -46,13 +46,19 @@ async def test_create_reading_session(client: AsyncClient, auth_headers: dict[st
             "start_time": now.isoformat(),
             "start_position": 0.25,
             "end_position": 0.35,
+            "device_type": "android",
             "pages_read": 20,
         },
     )
 
     assert response.status_code == 200
     data = response.json()
     assert data["book_id"] == book_id
+    assert data["start_position"] == 0.25
+    assert data["end_position"] == 0.35
+    assert data["device_type"] == "android"
+    sessions = (await client.get("/v1/progress/sessions", headers=auth_headers)).json()
+    assert sessions["sessions"][0] == data
 
 
 async def test_get_reading_statistics(client: AsyncClient, auth_headers: dict[str, str]):
@@ -134,6 +140,9 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h
                             "book_id": book_id,
                             "book_title": "Reading book",
                             "source": "reader",
+                            "start_position": index / 10,
+                            "end_position": (index + 1) / 10,
+                            "device_type": "linux",
                             "start_time": (end - timedelta(minutes=20 - index * 10)).isoformat(),
                             "end_time": (end - timedelta(minutes=10 - index * 10)).isoformat(),
                             "created_at": end.isoformat(),
@@ -149,6 +158,9 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h
     sessions = (await client.get("/v1/progress/sessions", headers=auth_headers)).json()
     assert sessions["pagination"]["total"] == 1
     assert sessions["sessions"][0]["duration_minutes"] == 20
+    assert sessions["sessions"][0]["start_position"] == 0
+    assert sessions["sessions"][0]["end_position"] == 0.2
+    assert sessions["sessions"][0]["device_type"] == "linux"
 
     result = await client.get(
         "/v1/progress/statistics",
@@ -161,3 +173,119 @@ async def test_statistics_group_reader_checkpoints_as_one_session(client, auth_h
     assert statistics["totals"]["sessions_count"] == 1
     assert statistics["totals"]["average_session_minutes"] == 20
     assert statistics["books_breakdown"][0]["sessions_count"] == 1
+
+
+async def test_concurrent_session_retry_reuses_generated_timestamps(client, auth_headers, book_id, monkeypatch):
+    import asyncio
+
+    from papyrus.services import progress
+
+    upload = progress.apply_powersync_upload_batch
+
+    async def delayed_upload(*args, **kwargs):
+        await asyncio.sleep(0.1)
+        return await upload(*args, **kwargs)
+
+    monkeypatch.setattr(progress, "apply_powersync_upload_batch", delayed_upload)
+
+    payload = {
+        "session_id": str(uuid4()),
+        "book_id": book_id,
+        "start_time": (datetime.now(UTC) - timedelta(minutes=10)).isoformat(),
+        "start_position": 0.2,
+        "end_position": 0.4,
+        "device_type": "android",
+        "pages_read": 12,
+    }
+
+    first, second = await asyncio.gather(
+        client.post("/v1/progress/sessions", headers=auth_headers, json=payload),
+        client.post("/v1/progress/sessions", headers=auth_headers, json=payload),
+    )
+
+    assert first.status_code == second.status_code == 200
+    assert first.json() == second.json()
+
+    for field, value in [("start_position", 0.3), ("end_position", 0.5), ("device_type", "windows")]:
+        response = await client.post("/v1/progress/sessions", headers=auth_headers, json={**payload, field: value})
+
+        assert response.status_code == 400
+
+
+async def test_statistics_excludes_sessions_ending_at_period_start(client, auth_headers, book_id):
+    boundary = datetime(2026, 1, 2, tzinfo=UTC)
+
+    response = await client.post(
+        "/v1/progress/sessions",
+        headers=auth_headers,
+        json={
+            "book_id": book_id,
+            "start_time": (boundary - timedelta(minutes=10)).isoformat(),
+            "end_time": boundary.isoformat(),
+            "pages_read": 5,
+        },
+    )
+
+    assert response.status_code == 200
+    params = {"start_date": "2026-01-02", "end_date": "2026-01-02"}
+    result = await client.get("/v1/progress/statistics", headers=auth_headers, params=params)
+    statistics = result.json()
+    assert statistics["totals"]["sessions_count"] == 0
+    assert statistics["totals"]["pages_read"] == 0
+    assert statistics["totals"]["reading_time_minutes"] == 0
+    assert statistics["books_breakdown"] == []
+    assert statistics["daily_breakdown"][0]["sessions_count"] == 0
+
+    response = await client.post(
+        "/v1/progress/sessions",
+        headers=auth_headers,
+        json={
+            "book_id": book_id,
+            "start_time": boundary.isoformat(),
+            "end_time": boundary.isoformat(),
+            "pages_read": 3,
+        },
+    )
+
+    assert response.status_code == 200
+    result = await client.get("/v1/progress/statistics", headers=auth_headers, params=params)
+    statistics = result.json()
+    assert statistics["totals"]["sessions_count"] == 1
+    assert statistics["totals"]["pages_read"] == 3
+    assert statistics["daily_breakdown"][0]["sessions_count"] == 1
+
+
+async def test_current_streak_is_independent_of_statistics_period(client, auth_headers, book_id, monkeypatch):
+    from papyrus.services import progress
+
+    now = datetime(2026, 3, 1, 12, tzinfo=UTC)
+
+    class FrozenDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return now
+
+    monkeypatch.setattr(progress, "datetime", FrozenDateTime)
+
+    for days_ago in range(3):
+        end = now - timedelta(days=days_ago)
+
+        response = await client.post(
+            "/v1/progress/sessions",
+            headers=auth_headers,
+            json={
+                "book_id": book_id,
+                "start_time": (end - timedelta(minutes=10)).isoformat(),
+                "end_time": end.isoformat(),
+            },
+        )
+
+        assert response.status_code == 200
+
+    for params in [{}, {"start_date": "2026-02-01", "end_date": "2026-02-01"}]:
+        response = await client.get("/v1/progress/statistics", headers=auth_headers, params=params)
+        assert response.status_code == 200
+        assert response.json()["totals"]["current_streak"] == 3
+
+    current = (await client.get("/v1/progress/statistics", headers=auth_headers)).json()
+    assert current["totals"]["longest_streak"] == 1