diff --git a/jobs/gmail-header-backfill/README.md b/jobs/gmail-header-backfill/README.md index 31827a8..04301d5 100644 --- a/jobs/gmail-header-backfill/README.md +++ b/jobs/gmail-header-backfill/README.md @@ -91,6 +91,49 @@ parts (plan §5.1); that keeps per-message cost low relative to text or are skipped, logged, and counted; the row is left for a future run rather than half-updated. +## Fallback parse (`parse_headers_fallback`) + +The full 225 030-row run (2026-07) left 9 envelopes whose headers the typed +`policy.default` parse rejects outright (diagnosis 2026-07-14): + +- 7× `ValueError: address parts cannot contain CR or LF` — an RFC 2047 + encoded-word decoding to text with a newline (`=0A`) inside a display + name, e.g. `Rekrutacja z =?utf-8?Q?ExampleCorp=0A?= ` +- 1× `AttributeError: 'Group' object has no attribute 'local_part'` — RFC + 5322 group syntax: `To: unlisted-recipients:; (no To-header on input)` +- 1× `AttributeError: 'str' object has no attribute 'token_type'` — a + malformed display name (CPython parser bug, fixed in newer versions but + present on PIHA's 3.11) + +When the typed parse raises, the job retries with `parse_headers_fallback`: +a pure `policy.compat32` parse where address fields are split into +`{name, address}` by `email.utils.getaddresses` over the **raw** header text +(no RFC 2047 decoding — names may keep literal `=?...?=` encoded-words), and +subject/`delivered_to`/`date_raw` stay raw strings. The written entity has +exactly the same §4.1 shape — only parse **quality** degrades, never the +schema. String values are sanitized so no lone surrogates (which postgres +jsonb rejects) can reach the DB. + +Fallback successes are counted separately as `parsed_fallback` (a labeled +subset of `updated`) and logged per row (`headers.parsed_fallback` with the +original typed error) — never silently mixed into ordinary successes. Only +when the fallback **also** fails does the row count as `parse_errors`. + +## Stats must balance — no silent skips + +Every scanned row lands in exactly one bucket, and `run_complete` / +`summary` must satisfy: + +``` +scanned = updated + already_has_headers + parse_errors + read_errors + + missing_file +``` + +A missing `.eml` is its own counter (`missing_file`) with a per-row +`skip.missing_file` info log carrying the envelope id and the expected +path — a full run can no longer lose rows without a trace. Any non-zero +`parse_errors`/`read_errors`/`missing_file` makes the CLI exit 1. + ## Idempotency and resumability (plan §5.2) ```sql diff --git a/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py index 11970f4..e7ce14b 100644 --- a/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py +++ b/jobs/gmail-header-backfill/src/gmail_header_backfill/backfill.py @@ -25,6 +25,10 @@ DSN can also come from the KB_DSN env var instead of --dsn. Idempotency: the UPDATE only touches rows that don't already carry a `headers` entity (§5.2 `WHERE NOT EXISTS`), so re-running any slice (including after a crash) is safe — already-backfilled rows are skipped, not double-appended. + +Messages the typed (policy.default) parse rejects fall back to a degraded +compat32/raw-string parse (parse_headers_fallback) — same §4.1 entity shape, counted +separately as `parsed_fallback`, never silently mixed into ordinary successes. """ from __future__ import annotations @@ -104,7 +108,86 @@ def parse_headers(raw: bytes) -> dict: subject_header = msg.get("Subject") subject = str(subject_header) if subject_header is not None else None - date_raw = msg_compat.get("Date") + # compat32 .get() returns an email.header.Header (not str) when the raw + # value contains 8-bit bytes — str() + _sanitize keeps it JSON/jsonb-safe. + # An unguarded Header here crashed the original full run mid-slice + # (TypeError at json.dumps), silently losing the rest of the slice. + date_header = msg_compat.get("Date") + date_raw = _sanitize(str(date_header)) if date_header is not None else None + + return { + "type": "headers", + "from": from_obj, + "to": to_list, + "cc": cc_list, + "delivered_to": delivered_to, + "subject": subject, + "date_raw": date_raw, + } + + +def _sanitize(value: Optional[str]) -> Optional[str]: + """Strip surrogates a compat32 bytes-parse can leave in header text. + + Lone surrogates are not valid UTF-8 and postgres jsonb rejects them; each + undecodable byte degrades to '?' instead. Real Unicode passes through. + """ + if value is None: + return None + return value.encode("utf-8", errors="replace").decode("utf-8") + + +def parse_headers_fallback(raw: bytes) -> dict: + """Degraded parse for messages the typed (policy.default) path rejects. + + Real-world triggers found in the archive (diagnosis 2026-07-14, 9 of + 225 030): RFC 2047 encoded-words that decode to text with CR/LF in a + display name (ValueError in headerregistry), group syntax in To: + ("unlisted-recipients:;"), and a _header_value_parser bug on malformed + display names (fixed in newer CPython, present on PIHA's 3.11). + + Everything comes from a compat32 parse: address fields are split into + {name, address} by email.utils.getaddresses over the RAW header text — no + RFC 2047 decoding, so names may keep literal =?...?= encoded-words. + Subject and delivered_to stay raw strings too. The returned shape is + exactly parse_headers()'s §4.1 shape — only parse QUALITY degrades, never + the schema. Callers count these separately (stats["parsed_fallback"]). + """ + msg = email.message_from_bytes(raw, policy=email.policy.compat32) + + from_headers = [str(h) for h in (msg.get_all("From") or [])] + if len(from_headers) > 1: + _log.warning("headers.multiple_from", count=len(from_headers)) + + from_obj: Optional[dict] = None + if from_headers: + from_addrs = email.utils.getaddresses([from_headers[0]]) + if from_addrs and from_addrs[0][1]: + name, address = from_addrs[0] + from_obj = {"name": _sanitize(name) or None, "address": _sanitize(address)} + + to_list = [ + {"name": _sanitize(name) or None, "address": _sanitize(address)} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("To") or [])] + ) + if address + ] + + cc_list = [ + {"name": _sanitize(name) or None, "address": _sanitize(address)} + for name, address in email.utils.getaddresses( + [str(h) for h in (msg.get_all("Cc") or [])] + ) + if address + ] + + delivered_to = [_sanitize(str(h)) for h in (msg.get_all("Delivered-To") or [])] + + subject_header = msg.get("Subject") + subject = _sanitize(str(subject_header)) if subject_header is not None else None + + date_raw = _sanitize(msg.get("Date")) return { "type": "headers", @@ -169,16 +252,25 @@ async def run( """Backfill headers for one --limit/--offset slice of `source='gmail'` envelopes. dry-run (apply=False) parses and counts everything without writing to the DB. - Returns stats: scanned, already_has_headers, updated, read_errors, parse_errors. - Anomalies that aren't errors (e.g. more than one From: header) are logged - (`headers.multiple_from`) rather than counted here — see parse_headers(). + Returns stats that must always balance: + + scanned = updated + already_has_headers + parse_errors + read_errors + + missing_file + + `parsed_fallback` is a labeled SUBSET of `updated` (rows recovered by + parse_headers_fallback() after the typed parse raised), not a separate + outcome — it exists so degraded parses are never silently mixed into + ordinary successes. Anomalies that aren't errors (e.g. more than one + From: header) are logged (`headers.multiple_from`) rather than counted. """ stats = { "scanned": 0, "already_has_headers": 0, "updated": 0, + "parsed_fallback": 0, "read_errors": 0, "parse_errors": 0, + "missing_file": 0, } conn = await asyncpg.connect(dsn) @@ -204,20 +296,34 @@ async def run( eml_path = archive_root / row["raw_ref"] try: raw = eml_path.read_bytes() + except FileNotFoundError: + _log.info("skip.missing_file", envelope_id=envelope_id, + expected_path=str(eml_path)) + stats["missing_file"] += 1 + continue except OSError: _log.warning("skip.read_error", envelope_id=envelope_id, path=str(eml_path)) stats["read_errors"] += 1 continue + # json.dumps stays INSIDE the try: a non-serializable header value + # must count as this row's parse_error, not crash the whole slice + # (that's exactly how the original full run lost 4999 rows). try: - headers = parse_headers(raw) - except Exception: - _log.warning("skip.parse_error", envelope_id=envelope_id, exc_info=True) - stats["parse_errors"] += 1 - continue + patch = json.dumps([parse_headers(raw)]) + except Exception as typed_exc: + try: + patch = json.dumps([parse_headers_fallback(raw)]) + except Exception: + _log.warning("skip.parse_error", envelope_id=envelope_id, exc_info=True) + stats["parse_errors"] += 1 + continue + _log.info("headers.parsed_fallback", envelope_id=envelope_id, + typed_error=repr(typed_exc)) + stats["parsed_fallback"] += 1 stats["updated"] += 1 - pending.append((envelope_id, json.dumps([headers]))) + pending.append((envelope_id, patch)) if len(pending) >= UPDATE_BATCH_SIZE: await _flush() @@ -265,7 +371,8 @@ def main() -> None: mode = "APPLY" if args.apply else "DRY-RUN" _log.info("summary", mode=mode, **stats) - sys.exit(1 if stats["read_errors"] + stats["parse_errors"] > 0 else 0) + failures = stats["read_errors"] + stats["parse_errors"] + stats["missing_file"] + sys.exit(1 if failures > 0 else 0) if __name__ == "__main__": diff --git a/jobs/gmail-header-backfill/tests/test_backfill.py b/jobs/gmail-header-backfill/tests/test_backfill.py index a881694..255753a 100644 --- a/jobs/gmail-header-backfill/tests/test_backfill.py +++ b/jobs/gmail-header-backfill/tests/test_backfill.py @@ -11,6 +11,7 @@ from gmail_header_backfill.backfill import ( _has_headers, fetch_batch, parse_headers, + parse_headers_fallback, run, ) @@ -170,6 +171,95 @@ class TestParseHeadersDateRaw: raw = _eml({"From": "a@b.com", "Date": "not-a-date-at-all"}) assert parse_headers(raw)["date_raw"] == "not-a-date-at-all" + def test_date_raw_with_8bit_bytes_is_json_serializable_str(self): + # compat32 .get() returns email.header.Header (not str) when the raw + # value has 8-bit bytes; unguarded, json.dumps then raises TypeError — + # this exact case crashed the original full run and lost 4999 rows. + raw = ( + b"From: a@b.com\r\n" + b"Date: Tue, 19 May 2009 10:27:09 +0200 (Ho\xe9ra)\r\n" + b"\r\nbody" + ) + headers = parse_headers(raw) + assert isinstance(headers["date_raw"], str) + json.dumps([headers], ensure_ascii=False).encode("utf-8") # must not raise + + +# Anonymized real patterns from the 2026-07-14 diagnosis of the 9 parse_errors +# left by the full 225 030-row run (see the job README, "Fallback parse"). +# +# Pattern 1 (7 of 9): RFC 2047 encoded-word whose decoded text contains a +# newline (=0A) inside a display name — policy.default raises +# "ValueError: invalid arguments; address parts cannot contain CR or LF". +_CRLF_ENCODED_WORD_EML = ( + b"From: Rekrutacja z =?utf-8?Q?ExampleCorp=0A?= \r\n" + b"To: Jan Kowalski \r\n" + b"Subject: Nowe oferty\r\n" + b"Date: Mon, 15 Sep 2014 13:24:18 +0000\r\n" + b"\r\nbody" +) + +# Pattern 2 (1 of 9): RFC 5322 group syntax in To: — policy.default raises +# "AttributeError: 'Group' object has no attribute 'local_part'". +_GROUP_TO_EML = ( + b'From: "Maria Example" \r\n' + b"To: unlisted-recipients:; (no To-header on input)\r\n" + b"Subject: Re: [List] hello\r\n" + b"Date: Tue, 16 Oct 2007 20:12:41 +0200\r\n" + b"\r\nbody" +) + + +class TestParseHeadersFallback: + def test_typed_parse_raises_on_crlf_encoded_word_from(self): + with pytest.raises(ValueError): + parse_headers(_CRLF_ENCODED_WORD_EML) + + def test_fallback_recovers_crlf_encoded_word_from(self): + headers = parse_headers_fallback(_CRLF_ENCODED_WORD_EML) + assert headers["type"] == "headers" + # address is extracted; display name stays a RAW undecoded string + assert headers["from"]["address"] == "mailing@example.pl" + assert "=?utf-8?Q?ExampleCorp" in headers["from"]["name"] + assert headers["to"] == [{"name": "Jan Kowalski", "address": "jan@example.com"}] + assert headers["subject"] == "Nowe oferty" + assert headers["date_raw"] == "Mon, 15 Sep 2014 13:24:18 +0000" + + def test_typed_parse_raises_on_group_to(self): + with pytest.raises(AttributeError): + parse_headers(_GROUP_TO_EML) + + def test_fallback_recovers_group_to(self): + headers = parse_headers_fallback(_GROUP_TO_EML) + assert headers["from"] == {"name": "Maria Example", "address": "maria@example.pl"} + assert headers["to"] == [] # empty group has no real addresses + + def test_fallback_shape_matches_normal_parse_shape(self): + # Degradation is parse quality only, never the schema (§4.1). + normal = parse_headers(_eml({ + "From": "a@b.com", "To": "c@d.com", + "Subject": "x", "Date": "Tue, 10 Jun 2025 12:00:00 +0000", + })) + fallback = parse_headers_fallback(_CRLF_ENCODED_WORD_EML) + assert set(fallback.keys()) == set(normal.keys()) + assert isinstance(fallback["to"], list) + assert isinstance(fallback["delivered_to"], list) + + def test_fallback_output_is_json_and_utf8_safe(self): + # compat32 over raw 8-bit bytes leaves surrogates in header text; + # postgres jsonb rejects them, so the fallback must sanitize. + raw = ( + b"From: Rekrutacja z =?utf-8?Q?ExampleCorp=0A?= \r\n" + b"Subject: Pr\xe9sent\r\n" + b"Date: Tue, 19 May 2009 10:27:09 +0200\r\n" + b"\r\nbody" + ) + headers = parse_headers_fallback(raw) + # the undecodable byte degrades to U+FFFD (via compat32's Header) or + # '?' (via _sanitize) — never a lone surrogate, which jsonb rejects + assert headers["subject"] in ("Pr�sent", "Pr?sent") + json.dumps(headers, ensure_ascii=False).encode("utf-8") # must not raise + class TestDecodeJsonb: def test_passes_through_object(self): @@ -297,7 +387,7 @@ class TestRun: assert stats["updated"] == 0 assert conn.executemany_calls == [] - async def test_missing_eml_file_counted_as_read_error(self, tmp_path, monkeypatch): + async def test_missing_eml_file_counted_as_missing_file(self, tmp_path, monkeypatch): archive_root = tmp_path / "archive" archive_root.mkdir() conn = _FakeConn([_row("env1", "gmail/2025/06/missing.eml", [_attachment_entity()])]) @@ -305,9 +395,75 @@ class TestRun: stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) - assert stats["read_errors"] == 1 + assert stats["missing_file"] == 1 + assert stats["read_errors"] == 0 assert stats["updated"] == 0 + async def test_fallback_row_updated_and_counted_separately(self, tmp_path, monkeypatch): + # A message the typed parse rejects (real anonymized pattern from the + # 2026-07-14 diagnosis) must be recovered by the fallback, written with + # the same schema, and labeled parsed_fallback — not a silent success. + archive_root = self._setup_archive( + tmp_path, "gmail/2014/09/m1.eml", _CRLF_ENCODED_WORD_EML) + conn = _FakeConn([_row("env1", "gmail/2014/09/m1.eml", [_attachment_entity()])]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["updated"] == 1 + assert stats["parsed_fallback"] == 1 + assert stats["parse_errors"] == 0 + _, rows = conn.executemany_calls[0] + patch = json.loads(rows[0][1]) + assert patch[0]["type"] == "headers" + assert patch[0]["from"]["address"] == "mailing@example.pl" + + async def test_parse_error_when_fallback_also_fails(self, tmp_path, monkeypatch): + archive_root = self._setup_archive( + tmp_path, "gmail/2014/09/m1.eml", _CRLF_ENCODED_WORD_EML) + conn = _FakeConn([_row("env1", "gmail/2014/09/m1.eml", [_attachment_entity()])]) + self._patch_connect(monkeypatch, conn) + + def _boom(raw): + raise RuntimeError("fallback failed too") + monkeypatch.setattr( + "gmail_header_backfill.backfill.parse_headers_fallback", _boom) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["parse_errors"] == 1 + assert stats["parsed_fallback"] == 0 + assert stats["updated"] == 0 + + async def test_stats_balance_across_all_outcomes(self, tmp_path, monkeypatch): + # scanned = updated + already_has_headers + parse_errors + read_errors + # + missing_file (parsed_fallback is a subset of updated) + ok_raw = _eml({"From": "a@b.com", "Date": "Tue, 10 Jun 2025 12:00:00 +0000"}) + archive_root = self._setup_archive(tmp_path, "gmail/ok.eml", ok_raw) + (archive_root / "gmail/fb.eml").write_bytes(_CRLF_ENCODED_WORD_EML) + + already = [{"type": "headers", "from": None, "to": [], "cc": [], + "delivered_to": [], "subject": None, "date_raw": None}] + conn = _FakeConn([ + _row("env-ok", "gmail/ok.eml", [_attachment_entity()]), + _row("env-already", "gmail/ok.eml", already), + _row("env-fallback", "gmail/fb.eml", [_attachment_entity()]), + _row("env-missing", "gmail/nope.eml", [_attachment_entity()]), + ]) + self._patch_connect(monkeypatch, conn) + + stats = await run(dsn="postgresql://fake", archive_root=archive_root, apply=True) + + assert stats["scanned"] == 4 + assert stats["scanned"] == ( + stats["updated"] + stats["already_has_headers"] + + stats["parse_errors"] + stats["read_errors"] + stats["missing_file"] + ) + assert stats["updated"] == 2 + assert stats["parsed_fallback"] == 1 + assert stats["already_has_headers"] == 1 + assert stats["missing_file"] == 1 + async def test_limit_and_offset_are_passed_to_query(self, tmp_path, monkeypatch): captured = {}