fix(gmail-header-backfill): compat32 fallback + missing_file stat — 8-bit Date header crashed json.dumps mid-slice, losing 4999 rows
Diagnosis of the 5008 header-less envelopes after the full 225 030-row run
(read-only, on PIHA, 2026-07-14):
- 4999 = one contiguous block at ORDER BY id positions 70001-74999: the
--offset 70000 slice died mid-run. Root cause reproduced: a Date: header
with raw 8-bit bytes makes compat32 .get() return email.header.Header
(not str), and json.dumps([headers]) — which sat OUTSIDE the per-row
try/except — raised TypeError and killed the process, silently losing
the rest of the slice.
- 9 = genuine typed-parse failures: 7x RFC 2047 encoded-word decoding to
CR/LF inside a display name (ValueError in headerregistry), 1x RFC 5322
group syntax in To: ("unlisted-recipients:;"), 1x CPython
_header_value_parser bug on a malformed display name (fixed upstream,
present on PIHA's 3.11).
- 0 missing .eml files.
Fixes:
- date_raw: str() + surrogate sanitization on the compat32 value — the
crash cause, now also covered by a regression test.
- json.dumps moved inside the per-row try: a non-serializable value counts
as that row's parse_error instead of crashing the slice.
- parse_headers_fallback(): on typed-parse failure retry with a pure
compat32 parse — getaddresses over raw header text, raw-string values,
same §4.1 entity shape. Counted separately as parsed_fallback (labeled
subset of updated), logged per row with the original typed error.
- missing_file counter + skip.missing_file info log (id, expected path);
run_complete now balances: scanned = updated + already_has_headers +
parse_errors + read_errors + missing_file. Non-zero missing_file also
fails the exit code.
Verified: 43/43 pytest locally (3.13) and on PIHA (3.11); read-only dry
runs on PIHA — all 9 parse failures recover via fallback, the lost slice
completes scanned=5000 updated=4999 already_has_headers=1 with zero errors.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
oskar
parent
31e30b04d1
commit
f14ad410a0
|
|
@ -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?= <mailing@example.pl>`
|
||||
- 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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
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__":
|
||||
|
|
|
|||
|
|
@ -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?= <mailing@example.pl>\r\n"
|
||||
b"To: Jan Kowalski <jan@example.com>\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" <maria@example.pl>\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?= <mailing@example.pl>\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<EFBFBD>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 = {}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue