From 0eb9a7d8b8b435664252dce563db1d25cfbe7195 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 17:34:27 +0200 Subject: [PATCH 1/2] 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 From 0525085a1eeddb7d55c0e3507204513fda66fd67 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 14 Aug 2026 18:59:58 +0200 Subject: [PATCH 2/2] gmail: keep failed body fetches pending by id --- docs/sources.md | 5 +- nerve/sources/gmail.py | 192 +++++++++++++++++++++++++++++++------ tests/test_gmail_source.py | 83 +++++++++++++--- 3 files changed, 239 insertions(+), 41 deletions(-) diff --git a/docs/sources.md b/docs/sources.md index 1e45593d..b5f67579 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -66,10 +66,11 @@ A persistent cron job (`inbox-processor`) runs every 15 minutes: ### Gmail - **Adapter:** `nerve/sources/gmail.py` — uses `gog gmail messages search` + `gog gmail get` CLI -- **Cursor:** Epoch timestamp from Gmail's `internalDate` (the receive timestamp Gmail uses for `after:` filtering) +- **Cursor:** JSON containing the newest successfully fetched `internalDate` epoch timestamp plus per-message retry state; legacy epoch-only cursors are accepted - **First run:** Fetches emails from the last 24 hours (`newer_than:1d`) -- **Subsequent runs:** Uses `after:` with client-side dedup (Gmail's `after:` has ~2s tolerance window) +- **Subsequent runs:** Uses `after:` with client-side dedup (Gmail's `after:` has ~2s tolerance window), and retries pending message IDs directly even after they fall out of the bounded search window - **Two-step fetch:** Search returns metadata only; body + `internalDate` are fetched per-message via `gog gmail get` (up to 5 concurrent) +- **Body failures:** Failed fetches are omitted from the inbox and stored with their search metadata, status, and attempt count so one deterministic failure cannot block or lose newer messages - **Default schedule:** `*/15 * * * *` (every 15 min) ### IMAP diff --git a/nerve/sources/gmail.py b/nerve/sources/gmail.py index f86d72b2..9a65acb3 100644 --- a/nerve/sources/gmail.py +++ b/nerve/sources/gmail.py @@ -8,8 +8,11 @@ - `gog gmail messages search ` → list of {id, subject, from, date, labels} - `gog gmail get ` → {body, headers, message} with full HTML body -Cursor semantics: epoch timestamp (seconds) from Gmail's internalDate -(the timestamp Gmail uses for `after:` search filtering). +Cursor semantics: JSON state containing the newest successfully fetched +Gmail `internalDate` epoch timestamp and any messages whose bodies still need +to be fetched. Failed message metadata is kept in the cursor so retries can +continue even after a message falls out of the search result window. Legacy +epoch-only cursors are still accepted. On first run (no cursor), uses `newer_than:1d`. Note: gog always returns HTML email bodies. We extract clean text for @@ -38,6 +41,122 @@ # Max concurrent message fetches to avoid overwhelming gog _MAX_CONCURRENT_GETS = 5 + +def _decode_cursor( + cursor: str | None, +) -> tuple[int | None, dict[str, dict[str, Any]]]: + """Decode a Gmail cursor, accepting the legacy epoch-only format.""" + if cursor is None: + return None, {} + + try: + raw_state = json.loads(cursor) + except (json.JSONDecodeError, TypeError): + try: + return int(cursor), {} + except (TypeError, ValueError): + logger.warning("Invalid Gmail cursor %r — starting from the lookback window", cursor) + return None, {} + + # ``json.loads("123")`` returns an int, which is the old cursor format. + if isinstance(raw_state, int): + return raw_state, {} + if not isinstance(raw_state, dict): + logger.warning("Invalid Gmail cursor %r — starting from the lookback window", cursor) + return None, {} + + raw_epoch = raw_state.get("epoch") + try: + epoch = int(raw_epoch) if raw_epoch is not None else None + except (TypeError, ValueError): + epoch = None + + pending: dict[str, dict[str, Any]] = {} + raw_pending = raw_state.get("pending", {}) + if isinstance(raw_pending, dict): + for message_id, raw_entry in raw_pending.items(): + if not isinstance(message_id, str) or not message_id: + continue + if not isinstance(raw_entry, dict): + continue + + message = raw_entry.get("message", {"id": message_id}) + if not isinstance(message, dict): + message = {"id": message_id} + else: + message = dict(message) + message.setdefault("id", message_id) + + try: + attempts = max(1, int(raw_entry.get("attempts", 1))) + except (TypeError, ValueError): + attempts = 1 + pending[message_id] = { + "status": raw_entry.get("status", "failed"), + "attempts": attempts, + "message": message, + } + + return epoch, pending + + +def _encode_cursor( + epoch: int | None, + pending: dict[str, dict[str, Any]], +) -> str: + """Encode the Gmail cursor and its durable per-message retry state.""" + return json.dumps({"epoch": epoch, "pending": pending}, sort_keys=True) + + +def _message_snapshot(message: dict[str, Any]) -> dict[str, Any]: + """Keep the search metadata needed to build a record on a later retry.""" + fields = ("id", "threadId", "subject", "from", "date", "labels") + return {field: message[field] for field in fields if field in message} + + +def _mark_pending( + pending: dict[str, dict[str, Any]], + message: dict[str, Any], +) -> None: + """Record a body-fetch failure without blocking newer messages.""" + message_id = str(message.get("id", "")) + if not message_id: + return + + previous = pending.get(message_id, {}) + try: + attempts = int(previous.get("attempts", 0)) + 1 + except (AttributeError, TypeError, ValueError): + attempts = 1 + pending[message_id] = { + "status": "failed", + "attempts": attempts, + "message": _message_snapshot(message), + } + + +def _candidate_messages( + messages: list[dict], + pending: dict[str, dict[str, Any]], +) -> list[dict]: + """Merge pending retries with fresh search results, deduplicated by ID.""" + candidates: dict[str, dict] = {} + for message_id, entry in pending.items(): + message = entry.get("message", {}) + if not isinstance(message, dict): + message = {} + message = dict(message) + message.setdefault("id", message_id) + candidates[message_id] = message + + for message in messages: + message_id = message.get("id") + if message_id: + # Fresh search metadata wins if a pending message is returned again. + candidates[str(message_id)] = message + + return list(candidates.values()) + # --------------------------------------------------------------------------- # Email boilerplate detection # --------------------------------------------------------------------------- @@ -86,21 +205,25 @@ def __init__(self, account: str, config: dict[str, Any]): self._config = config async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: - """Fetch new emails since cursor (epoch timestamp string). + """Fetch new emails since cursor. On first run (cursor=None): uses `newer_than:1d`. - On subsequent runs: uses `after:`. + On subsequent runs: uses `after:` and retries pending + message IDs independently of the search window. - The cursor stores Gmail's internalDate (second precision) which is the - same clock that `after:` filters on. Using +1 is sufficient because - both values are on the same timescale. + The cursor stores Gmail's internalDate (second precision), plus search + metadata for body fetches that need retrying. Using +1 is sufficient + because the cursor and Gmail's `after:` filter use the same timescale. """ + cursor_epoch, pending = _decode_cursor(cursor) + had_pending = bool(pending) + # Build query filter from cursor. # The cursor is derived from Gmail's internalDate (epoch seconds) — # the same timestamp that `after:` filters on. +1 ensures we skip # the message the cursor was set from. - if cursor: - after_epoch = int(cursor) + 1 + if cursor_epoch is not None: + after_epoch = cursor_epoch + 1 query_filter = f"after:{after_epoch} -in:spam -in:trash" else: query_filter = "newer_than:1d -in:spam -in:trash" @@ -111,25 +234,28 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: env["GOG_KEYRING_PASSWORD"] = keyring_password records: list[SourceRecord] = [] - newest_epoch: int | None = int(cursor) if cursor else None - body_fetch_failed = False + newest_epoch = cursor_epoch try: # Step 1: Search for message IDs + metadata messages = await self._search_messages(query_filter, limit, env) + candidates = _candidate_messages(messages, pending) - if not messages: + # Pending messages are retried by ID even if they no longer appear + # in the bounded search result window. + if not candidates: return FetchResult(records=[], next_cursor=cursor, has_more=False) # Step 2: Fetch body + internalDate for each message sem = asyncio.Semaphore(_MAX_CONCURRENT_GETS) tasks = [ self._fetch_message_body(msg["id"], env, sem) - for msg in messages + for msg in candidates ] bodies = await asyncio.gather(*tasks, return_exceptions=True) - for msg, body_result in zip(messages, bodies): + for msg, body_result in zip(candidates, bodies): + message_id = msg["id"] date_str = msg.get("date", "") # Extract body text, HTML, and internalDate from the fetch result @@ -139,14 +265,19 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: if isinstance(body_result, tuple): 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 + logger.warning("Failed to fetch body for %s: %s", message_id, body_result) + _mark_pending(pending, msg) continue else: - logger.warning("Failed to fetch body for %s", msg["id"]) - body_fetch_failed = True + logger.warning("Failed to fetch body for %s", message_id) + _mark_pending(pending, msg) continue + # A successful retry removes the message from the durable + # pending set before the record is returned. + was_pending = message_id in pending + pending.pop(message_id, None) + # Use internalDate for cursor tracking (matches Gmail's `after:` # filter). Fall back to the Date header only if internalDate # is unavailable. @@ -155,10 +286,15 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: # Client-side dedup: skip messages at or before the cursor. # Gmail's `after:` has a small tolerance window (~2s) so a # message right at the boundary can slip through. - if cursor and msg_epoch and msg_epoch <= int(cursor): + if ( + not was_pending + and cursor_epoch is not None + and msg_epoch + and msg_epoch <= cursor_epoch + ): logger.debug( "Gmail %s: skipping already-seen message %s (epoch=%d, cursor=%s)", - self.account, msg.get("id", "?"), msg_epoch, cursor, + self.account, message_id, msg_epoch, cursor_epoch, ) continue @@ -212,14 +348,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) - # 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 - ) + # The timestamp can move past a failed message because its search + # metadata is retained in ``pending`` and the body is retried by ID. + # This prevents one permanently failing message from pinning the whole + # Gmail stream or disappearing when it falls out of the search window. + if pending or newest_epoch != cursor_epoch or had_pending: + next_cursor = _encode_cursor(newest_epoch, pending) + else: + next_cursor = cursor return FetchResult(records=records, next_cursor=next_cursor, has_more=False) async def preprocess(self, records: list[SourceRecord]) -> list[SourceRecord]: diff --git a/tests/test_gmail_source.py b/tests/test_gmail_source.py index 21f6d67e..047107b0 100644 --- a/tests/test_gmail_source.py +++ b/tests/test_gmail_source.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import json from unittest.mock import AsyncMock, patch import pytest @@ -22,7 +23,7 @@ def _message(message_id: str, epoch: int) -> dict: @pytest.mark.asyncio -async def test_failed_body_fetch_keeps_cursor_until_message_recovers(): +async def test_failed_body_fetch_is_retried_until_message_recovers(): source = GmailSource("me@example.com", {}) messages = [_message("failed", 101), _message("healthy", 102)] failed_body_recovers = False @@ -44,16 +45,22 @@ async def fetch_body(message_id, env, sem): first = await source.fetch(cursor="100") assert [record.id for record in first.records] == ["healthy"] - assert first.next_cursor == "100" + first_cursor = json.loads(first.next_cursor) + assert first_cursor["epoch"] == 102 + assert first_cursor["pending"]["failed"]["status"] == "failed" + assert first_cursor["pending"]["failed"]["attempts"] == 1 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 [record.id for record in second.records] == ["failed"] assert "body for failed" in second.records[0].content - assert second.next_cursor == "102" - assert queries == ["after:101 -in:spam -in:trash"] * 2 + assert json.loads(second.next_cursor) == {"epoch": 102, "pending": {}} + assert queries == [ + "after:101 -in:spam -in:trash", + "after:103 -in:spam -in:trash", + ] @pytest.mark.asyncio @@ -72,17 +79,29 @@ async def fetch_body(message_id, env, sem): result = await source.fetch(cursor=None) assert result.records == [] - assert result.next_cursor is None + state = json.loads(result.next_cursor) + assert state["epoch"] is None + assert state["pending"]["failed"]["status"] == "failed" + assert state["pending"]["failed"]["attempts"] == 1 + + retry = await source.fetch(cursor=result.next_cursor) + retry_state = json.loads(retry.next_cursor) + assert retry_state["pending"]["failed"]["attempts"] == 2 @pytest.mark.asyncio -async def test_partial_first_sync_waits_to_establish_cursor_until_recovery(): +async def test_partial_first_sync_persists_failed_message_for_recovery(): 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): - assert query == "newer_than:1d -in:spam -in:trash" + queries.append(query) + assert query in { + "newer_than:1d -in:spam -in:trash", + "after:103 -in:spam -in:trash", + } return messages async def fetch_body(message_id, env, sem): @@ -97,13 +116,55 @@ async def fetch_body(message_id, env, sem): first = await source.fetch(cursor=None) assert [record.id for record in first.records] == ["healthy"] - assert first.next_cursor is None + first_cursor = json.loads(first.next_cursor) + assert first_cursor["epoch"] == 102 + assert "failed" in first_cursor["pending"] + + failed_body_recovers = True + second = await source.fetch(cursor=first.next_cursor) + + assert [record.id for record in second.records] == ["failed"] + assert json.loads(second.next_cursor) == {"epoch": 102, "pending": {}} + assert queries == [ + "newer_than:1d -in:spam -in:trash", + "after:103 -in:spam -in:trash", + ] + + +@pytest.mark.asyncio +async def test_pending_body_fetch_is_retried_after_message_leaves_search_window(): + source = GmailSource("me@example.com", {}) + failed = _message("failed", 101) + search_results = [[failed], []] + failed_body_recovers = False + queries: list[str] = [] + + async def search(query, limit, env): + queries.append(query) + return search_results.pop(0) + + async def fetch_body(message_id, env, sem): + if not failed_body_recovers: + return None + return "recovered body", None, 101 + + source._search_messages = search + source._fetch_message_body = fetch_body + + first = await source.fetch(cursor="100") + assert first.records == [] + assert json.loads(first.next_cursor)["pending"]["failed"]["attempts"] == 1 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" + assert [record.id for record in second.records] == ["failed"] + assert "recovered body" in second.records[0].content + assert json.loads(second.next_cursor) == {"epoch": 101, "pending": {}} + assert queries == [ + "after:101 -in:spam -in:trash", + "after:101 -in:spam -in:trash", + ] @pytest.mark.asyncio