Debugging Memory Exhaustion and Hash Mismatch Failures in Python Email Threading Pipelines

[INFO] Building adjacency matrix... nodes=412891, edges=389201
[ERROR] MemoryError: Unable to allocate 2.1 GiB for an array with shape (412891, 412891) and data type float64
[WARN] Hash mismatch for MSG-8842A: expected=sha256:a1b2c3..., computed=sha256:d4e5f6...
[CRITICAL] Process killed by OOM killer (exit code 137). Thread state not persisted.

import sqlite3
import hashlib
import logging
from collections import deque
from email import message_from_bytes
from email.policy import SMTP
from typing import Iterator, Optional, Tuple

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("threading_audit.log"), logging.StreamHandler()]
)

VOLATILE_HEADERS = {
    "Received", "X-Spam-Status", "X-Spam-Score", "ARC-Seal", "ARC-Message-Signature",
    "DKIM-Signature", "Authentication-Results", "X-MS-Exchange-Organization-SCL",
    "X-MS-Exchange-Organization-AuthSource"
}

def canonicalize_and_hash(raw_bytes: bytes) -> Tuple[str, bytes]:
    """RFC 5322 compliant canonicalization with SHA-256 digest generation."""
    try:
        msg = message_from_bytes(raw_bytes, policy=SMTP)
    except Exception as exc:
        raise ValueError(f"MIME parse failure: {exc}") from exc

    # Strip volatile transit headers
    for header in VOLATILE_HEADERS:
        del msg[header]

    # Reconstruct with normalized line endings
    canonical_bytes = msg.as_bytes().replace(b"\n", b"\r\n")
    
    digest = hashlib.sha256(canonical_bytes).hexdigest()
    return digest, canonical_bytes

class ThreadGraphBuilder:
    def __init__(self, db_path: str = "thread_graph.db"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path, timeout=30)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute("PRAGMA synchronous=NORMAL")
        self._init_schema()

    def _init_schema(self):
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS nodes (
                msg_id TEXT PRIMARY KEY,
                sha256 TEXT NOT NULL,
                subject TEXT,
                date TEXT
            );
            CREATE TABLE IF NOT EXISTS edges (
                parent_id TEXT NOT NULL,
                child_id TEXT NOT NULL,
                PRIMARY KEY (parent_id, child_id)
            );
            CREATE INDEX IF NOT EXISTS idx_edges_parent ON edges(parent_id);
            CREATE INDEX IF NOT EXISTS idx_edges_child ON edges(child_id);
        """)
        self.conn.commit()

    def ingest_node(self, msg_id: str, in_reply_to: Optional[str], 
                    references: Optional[str], sha256: str, 
                    subject: str, date: str) -> bool:
        """Ingest header metadata into disk-backed graph."""
        if not msg_id or not sha256:
            logging.error("Validation failed: msg_id and sha256 are required.")
            return False

        try:
            self.conn.execute(
                "INSERT OR IGNORE INTO nodes VALUES (?, ?, ?, ?)",
                (msg_id, sha256, subject, date)
            )
            # Parse References header into individual Message-IDs
            refs = [r.strip("<>") for r in (references or "").split() if r.strip("<>")]
            edges = [(r, msg_id) for r in refs]
            if in_reply_to:
                edges.append((in_reply_to.strip("<>"), msg_id))
            
            self.conn.executemany("INSERT OR IGNORE INTO edges VALUES (?, ?)", edges)
            self.conn.commit()
            return True
        except sqlite3.Error as exc:
            logging.error(f"DB ingestion failed for {msg_id}: {exc}")
            return False

    def resolve_thread(self, root_msg_id: str) -> Iterator[str]:
        """Iterative BFS traversal to resolve a family group without recursion."""
        queue = deque([root_msg_id])
        visited = {root_msg_id}
        yield root_msg_id

        while queue:
            current = queue.popleft()
            cursor = self.conn.execute(
                "SELECT child_id FROM edges WHERE parent_id = ?", (current,)
            )
            for (child_id,) in cursor.fetchall():
                if child_id not in visited:
                    visited.add(child_id)
                    yield child_id
                    queue.append(child_id)

    def close(self):
        self.conn.close()