From 0eb9a7d8b8b435664252dce563db1d25cfbe7195 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 17:34:27 +0200 Subject: [PATCH] gmail: retry messages after body fetch failures --- nerve/sources/gmail.py | 26 ++++++-- tests/test_gmail_source.py | 124 +++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 4 deletions(-) create mode 100644 tests/test_gmail_source.py diff --git a/nerve/sources/gmail.py b/nerve/sources/gmail.py index 6ea29e03..f86d72b2 100644 --- a/nerve/sources/gmail.py +++ b/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. diff --git a/tests/test_gmail_source.py b/tests/test_gmail_source.py new file mode 100644 index 00000000..21f6d67e --- /dev/null +++ b/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