gmail: retry messages after body fetch failures

nerve/sources/gmail.py

@@ -112,6 +112,7 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
 
         records: list[SourceRecord] = []
         newest_epoch: int | None = int(cursor) if cursor else None
+        body_fetch_failed = False
 
         try:
             # Step 1: Search for message IDs + metadata
@@ -139,6 +140,12 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
                     body, html_body, internal_date = body_result
                 elif isinstance(body_result, Exception):
                     logger.warning("Failed to fetch body for %s: %s", msg["id"], body_result)
+                    body_fetch_failed = True
+                    continue
+                else:
+                    logger.warning("Failed to fetch body for %s", msg["id"])
+                    body_fetch_failed = True
+                    continue
 
                 # Use internalDate for cursor tracking (matches Gmail's `after:`
                 # filter).  Fall back to the Date header only if internalDate
@@ -205,7 +212,14 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
         except Exception as e:
             logger.error("Gmail error for %s: %s", self.account, e)
 
-        next_cursor = str(newest_epoch) if newest_epoch else cursor
+        # Successful messages may be persisted now, but keep the cursor behind
+        # the whole batch so a transient failure cannot strand a header-only
+        # message. Inbox persistence deduplicates unchanged successful retries
+        # by ID.
+        next_cursor = (
+            cursor if body_fetch_failed
+            else str(newest_epoch) if newest_epoch else cursor
+        )
         return FetchResult(records=records, next_cursor=next_cursor, has_more=False)
 
     async def preprocess(self, records: list[SourceRecord]) -> list[SourceRecord]:
@@ -240,11 +254,12 @@ async def _search_messages(
 
     async def _fetch_message_body(
         self, message_id: str, env: dict, sem: asyncio.Semaphore,
-    ) -> tuple[str, str | None, int | None]:
+    ) -> tuple[str, str | None, int | None] | None:
         """Fetch the body text, HTML body, and internalDate of a single message.
 
         Returns:
             (text_body, html_body_or_none, internal_date_epoch_seconds).
+            ``None`` when the message could not be retrieved.
 
         ``gog gmail get`` puts one body variant in its top-level ``body``
         field.  For multipart/alternative messages it picks text/plain,
@@ -264,10 +279,13 @@ async def _fetch_message_body(
 
             if proc.returncode != 0:
                 logger.warning("gog gmail get %s failed: %s", message_id, stderr.decode()[:200])
-                return "", None, None
+                return None
 
             stdout_text = stdout.decode()
-            data = json.loads(stdout_text) if stdout_text.strip() else {}
+            if not stdout_text.strip():
+                logger.warning("gog gmail get %s returned no data", message_id)
+                return None
+            data = json.loads(stdout_text)
             body = data.get("body", "")
 
             # Extract internalDate from the raw Gmail API message object.

tests/test_gmail_source.py

@@ -0,0 +1,124 @@
+"""Regression tests for the Gmail source."""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from nerve.sources.gmail import GmailSource
+
+
+def _message(message_id: str, epoch: int) -> dict:
+    return {
+        "id": message_id,
+        "threadId": f"thread-{message_id}",
+        "subject": f"Message {message_id}",
+        "from": "sender@example.com",
+        "date": f"1970-01-01T00:{epoch // 60:02d}:{epoch % 60:02d}Z",
+        "labels": ["INBOX"],
+    }
+
+
+@pytest.mark.asyncio
+async def test_failed_body_fetch_keeps_cursor_until_message_recovers():
+    source = GmailSource("me@example.com", {})
+    messages = [_message("failed", 101), _message("healthy", 102)]
+    failed_body_recovers = False
+    queries: list[str] = []
+
+    async def search(query, limit, env):
+        queries.append(query)
+        return messages
+
+    async def fetch_body(message_id, env, sem):
+        if message_id == "failed" and not failed_body_recovers:
+            return None
+        epoch = 101 if message_id == "failed" else 102
+        return f"body for {message_id}", None, epoch
+
+    source._search_messages = search
+    source._fetch_message_body = fetch_body
+
+    first = await source.fetch(cursor="100")
+
+    assert [record.id for record in first.records] == ["healthy"]
+    assert first.next_cursor == "100"
+    assert "body for healthy" in first.records[0].content
+
+    failed_body_recovers = True
+    second = await source.fetch(cursor=first.next_cursor)
+
+    assert [record.id for record in second.records] == ["failed", "healthy"]
+    assert "body for failed" in second.records[0].content
+    assert second.next_cursor == "102"
+    assert queries == ["after:101 -in:spam -in:trash"] * 2
+
+
+@pytest.mark.asyncio
+async def test_body_fetch_exception_does_not_create_header_only_record():
+    source = GmailSource("me@example.com", {})
+
+    async def search(query, limit, env):
+        return [_message("failed", 101)]
+
+    async def fetch_body(message_id, env, sem):
+        raise TimeoutError("temporary failure")
+
+    source._search_messages = search
+    source._fetch_message_body = fetch_body
+
+    result = await source.fetch(cursor=None)
+
+    assert result.records == []
+    assert result.next_cursor is None
+
+
+@pytest.mark.asyncio
+async def test_partial_first_sync_waits_to_establish_cursor_until_recovery():
+    source = GmailSource("me@example.com", {})
+    messages = [_message("failed", 101), _message("healthy", 102)]
+    failed_body_recovers = False
+
+    async def search(query, limit, env):
+        assert query == "newer_than:1d -in:spam -in:trash"
+        return messages
+
+    async def fetch_body(message_id, env, sem):
+        if message_id == "failed" and not failed_body_recovers:
+            return None
+        epoch = 101 if message_id == "failed" else 102
+        return f"body for {message_id}", None, epoch
+
+    source._search_messages = search
+    source._fetch_message_body = fetch_body
+
+    first = await source.fetch(cursor=None)
+
+    assert [record.id for record in first.records] == ["healthy"]
+    assert first.next_cursor is None
+
+    failed_body_recovers = True
+    second = await source.fetch(cursor=first.next_cursor)
+
+    assert [record.id for record in second.records] == ["failed", "healthy"]
+    assert second.next_cursor == "102"
+
+
+@pytest.mark.asyncio
+async def test_nonzero_body_command_is_reported_as_failure():
+    source = GmailSource("me@example.com", {})
+    process = AsyncMock()
+    process.returncode = 1
+    process.communicate.return_value = (b"", b"temporary API failure")
+
+    with patch(
+        "nerve.sources.gmail.asyncio.create_subprocess_exec",
+        AsyncMock(return_value=process),
+    ):
+        result = await source._fetch_message_body(
+            "message-id", {}, asyncio.Semaphore(1),
+        )
+
+    assert result is None