Attachment & Parent-Child Mapping: Extracting Defensible Document Families at Scale

The attachment-to-parent mapping stage is the structural backbone of any defensible processing workflow, and it belongs to the same EDRM Processing stage covered by the parent Deduplication & Family Grouping pipeline. Its job is narrow but load-bearing: establish immutable parent-child relationships across nested document families before anything downstream touches the data. When custodial containers are ingested, binding every extracted child to its container of record — before review, threading, or production — is what prevents metadata fragmentation, preserves chain-of-custody integrity, and ensures that privilege logs and redaction layers propagate correctly across the family tree. Get this wrong and a suppressed parent orphans a privileged spreadsheet, a flattened archive loses its folder lineage, and a produced attachment can no longer be tied back to the message that carried it. This subsystem assumes prior completion of exact-match filtering so that only unique parent containers and their attachments enter the mapping pipeline, and it emits a family graph that the rest of the pipeline treats as authoritative.

Architecture Overview

Every parent container decomposes into a tree, and every node in that tree inherits a single family identifier so the relationship survives suppression, threading, and production. The diagram below shows a parent email decomposed into its immediate attachments, where a nested ZIP expands into its own child documents and an embedded OLE object while every node still shares one family:

Structurally, the subsystem is a hash-first, depth-bounded traversal. A normalized container arrives from native file ingestion, the extractor walks it under a strict recursion limit, each extracted child is hashed at the extraction boundary, and the result is either mapped, suppressed as a known duplicate, or routed to a failure lane. Nothing is held in memory longer than a single child payload, and no relationship is written until the child’s digest and safe path are both established.

Memory Constraints & the Cost of Naive Extraction

Processing nested containers at scale requires strict memory isolation. The naive approach — extract an entire archive to disk, load each child into memory, hash it, then walk the tree recursively on the call stack — fails in three predictable ways at ESI volume. A multi-gigabyte PST or a recursively compressed archive exhausts heap before the first family is written. A synchronous parser blocks the event loop, so one malformed OLE stream stalls every other container behind it. And a deeply nested archive overflows the recursion stack long before it trips any business limit.

The pipeline replaces all three with a generator-driven batching strategy that yields fixed-size chunks of container paths, processes them through an asyncio.Semaphore-controlled worker pool, and streams mapping records directly to a relational or document-store backend as they are produced. Because results are yielded rather than accumulated, peak memory scales with the batch size and the largest single child — not with corpus size. Decoupling I/O from CPU-bound parsing keeps the memory footprint predictable even when a batch mixes a 40 KB loose document with a 6 GB mail store.

Hash verification happens at the extraction boundary rather than after the whole tree is materialized, which is what prevents duplicate attachment inflation. When an extracted child matches an existing digest, the pipeline defers to the algorithm-selection rules in Hash-Based Deduplication Strategies to decide whether to suppress the child outright, store a logical reference pointer to the surviving master, or flag it for cross-family review. That decision keeps the review platform from ballooning with byte-identical attachments while still recording, in the audit trail, every place the duplicate appeared.

Concurrency Model & Depth-Bounded Traversal

The execution model is an async generator feeding a bounded worker pool. Containers are sliced into fixed batches; each batch is admitted through a semaphore whose size is set to the point where CPU-bound hashing saturates available cores without oversubscribing the thread pool that runs the blocking reads. Every blocking operation — the SHA-256 read loop, the zipfile.extract call, the post-hash cleanup — is pushed onto a worker thread via asyncio.to_thread, so the event loop stays responsive and backpressure is real rather than cosmetic: when the semaphore is full, new batches genuinely wait.

Recursion is bounded, not open-ended. An archive that contains an archive that contains an archive is legitimate custodial data, but it is also the exact shape of a decompression bomb, so traversal carries an explicit depth counter and halts at a configurable maximum. Crossing that boundary is not an error to retry — it is a mapping record in its own right, tagged DEPTH_LIMIT_EXCEEDED, so the auditor can see precisely where extraction was deliberately stopped and why. Sizing this limit is a matter policy decision, the same way the concurrency ceiling shares the bounded-concurrency contract described in Async Batch Processing Design; a limit of five levels comfortably covers real-world “ZIP inside an email inside a PST” nesting while cutting off adversarial recursion long before it can exhaust the host.

Resilience & Failure Routing

Defensible mapping means an extraction never halts the run and never silently drops a child. Zip bombs, symlink loops, and directory-traversal payloads are intercepted before extraction begins: the sanitizer resolves every candidate path and confirms it stays strictly inside the parent container’s boundary, rejecting anything that escapes with a PATH_TRAVERSAL_BLOCKED record rather than an exception. This is the same containment posture enforced at the infrastructure layer by Security Boundary Configuration — untrusted bytes are assumed hostile until proven otherwise.

Every non-success outcome becomes a typed status on a MappingRecord instead of a crash: EXTRACTION_FAILED for a child that could not be read or written, CORRUPT_ARCHIVE for a container that will not open, PATH_TRAVERSAL_BLOCKED for a hostile path, and DEPTH_LIMIT_EXCEEDED for the recursion ceiling. These records are the dead-letter manifest for the mapping stage. They carry the parent identifier, the offending relative path, the depth at which the failure occurred, and a human-readable detail, so a container that fails on one child still contributes every child that succeeded, and the failed child is queued for manual review with enough context to reproduce it. That deterministic routing is what keeps throughput stable across a collection seeded with encrypted, malformed, and password-protected containers.

For compound documents, specialized extraction routines handle embedded streams without corrupting parent metadata. Mapping an embedded OLE object back to its container ensures that a legacy Office attachment retains its original creation timestamp and authorship metadata, and mapping ZIP archive contents into review-platform hierarchies standardizes how a flattened archive is represented downstream — preserving the logical folder path while stripping the unsafe traversal sequences that path might otherwise smuggle in.

Production Python Implementation

The following implementation demonstrates a production-ready mapping engine. It enforces path-traversal sanitization, applies memory-aware chunking, routes failures through the typed status mechanism described above, and maintains strict audit compliance via structured logging. Every blocking call is offloaded to a worker thread, extracted children are hashed with a streaming SHA-256 and cleaned up in a finally block, and recursion is guarded by an explicit depth counter.

import asyncio
import hashlib
import logging
import os
import structlog
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import AsyncIterator, List, Optional
from uuid import uuid4

# Structured logging configuration for audit compliance
structlog.configure(
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True
)

logger = structlog.get_logger()

@dataclass
class MappingRecord:
    parent_id: str
    child_id: str
    relative_path: str
    extraction_depth: int
    content_hash: str
    status: str
    error_detail: Optional[str] = None

class AttachmentMappingPipeline:
    def __init__(self, max_depth: int = 5, batch_size: int = 50, concurrency: int = 8):
        self.max_depth = max_depth
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(concurrency)
        self.logger = logger.bind(pipeline="attachment_mapping")

    @staticmethod
    async def compute_sha256(file_path: Path) -> str:
        """Stream-based SHA-256 calculation to prevent memory exhaustion."""
        def _hash() -> str:
            sha256 = hashlib.sha256()
            with open(file_path, "rb") as f:
                while chunk := f.read(8192):
                    sha256.update(chunk)
            return sha256.hexdigest()

        # Offload the blocking read+hash loop to a worker thread so the event
        # loop stays responsive during large extractions.
        return await asyncio.to_thread(_hash)

    def sanitize_path(self, base: Path, target: Path) -> Optional[Path]:
        """Resolve and validate a path to prevent traversal attacks."""
        try:
            base_resolved = base.resolve()
            resolved = (base / target).resolve()
            if not resolved.is_relative_to(base_resolved):
                return None
            return resolved
        except (ValueError, RuntimeError, OSError):
            return None

    async def extract_and_map(
        self, parent_id: str, archive_path: Path, current_depth: int = 0
    ) -> AsyncIterator[MappingRecord]:
        """Recursively extract archives with depth limits and safe path validation."""
        if current_depth > self.max_depth:
            yield MappingRecord(
                parent_id=parent_id,
                child_id=str(uuid4()),
                relative_path=str(archive_path.name),
                extraction_depth=current_depth,
                content_hash="",
                status="DEPTH_LIMIT_EXCEEDED",
                error_detail="Extraction halted at recursion boundary"
            )
            return

        try:
            with zipfile.ZipFile(archive_path, "r") as zf:
                for info in zf.infolist():
                    if info.is_dir():
                        continue
                    
                    safe_path = self.sanitize_path(archive_path.parent, Path(info.filename))
                    if not safe_path:
                        yield MappingRecord(parent_id=parent_id, child_id=str(uuid4()),
                                            relative_path=info.filename, extraction_depth=current_depth,
                                            content_hash="", status="PATH_TRAVERSAL_BLOCKED")
                        continue

                    # Extract to the sanitized location, hash it, then clean up.
                    extracted_path = safe_path
                    try:
                        await asyncio.to_thread(zf.extract, info, archive_path.parent)
                        content_hash = await self.compute_sha256(extracted_path)
                        yield MappingRecord(parent_id=parent_id, child_id=str(uuid4()),
                                            relative_path=info.filename, extraction_depth=current_depth,
                                            content_hash=content_hash, status="MAPPED")
                    except Exception as e:
                        yield MappingRecord(parent_id=parent_id, child_id=str(uuid4()),
                                            relative_path=info.filename, extraction_depth=current_depth,
                                            content_hash="", status="EXTRACTION_FAILED",
                                            error_detail=str(e))
                    finally:
                        if extracted_path.exists():
                            await asyncio.to_thread(os.remove, extracted_path)
        except zipfile.BadZipFile:
            yield MappingRecord(parent_id=parent_id, child_id=str(uuid4()),
                                relative_path=str(archive_path.name), extraction_depth=current_depth,
                                content_hash="", status="CORRUPT_ARCHIVE")

    async def process_batch(self, parent_id: str, batch_paths: List[Path]) -> List[MappingRecord]:
        """Process a fixed-size chunk of containers under a concurrency bound."""
        async with self.semaphore:
            results: List[MappingRecord] = []
            for path in batch_paths:
                async for record in self.extract_and_map(parent_id, path):
                    results.append(record)
            return results

    async def run(self, parent_containers: List[Path]) -> AsyncIterator[MappingRecord]:
        """Execute the full pipeline with deterministic batch routing."""
        for i in range(0, len(parent_containers), self.batch_size):
            batch = parent_containers[i:i + self.batch_size]
            self.logger.info("processing_batch", batch_start=i, batch_size=len(batch))
            batch_results = await self.process_batch(str(uuid4()), batch)
            for record in batch_results:
                yield record

The engine’s reliance on official Python concurrency primitives and streaming hash verification aligns with the asyncio task documentation and the zipfile security guidance, which is what lets the mapping stage stand up under Daubert scrutiny: every extraction decision is a pure, reproducible function of the container bytes and the configured limits.

Observability & Compliance Metrics

Three KPIs tell you whether the mapping stage is healthy and defensible. Throughput confirms the pipeline is keeping pace with ingestion, family integrity rate confirms that children are actually being bound rather than dropped, and dead-letter velocity is the early-warning signal that a batch is full of malformed or hostile containers.

Because every outcome is already a typed status, instrumentation is a thin counter over the record stream rather than a second bookkeeping system:

from prometheus_client import Counter, Gauge

MAPPED = Counter(
    "attachment_children_mapped_total",
    "Children successfully bound to a parent family.",
)
DEAD_LETTERED = Counter(
    "attachment_children_dead_lettered_total",
    "Children diverted to the failure lane, labelled by status.",
    ["status"],
)
INTEGRITY_BPS = Gauge(
    "attachment_family_integrity_bps",
    "Mapped share of processed children, in basis points.",
)


def observe(records: list["MappingRecord"]) -> None:
    """Fold a completed batch's records into the shared registry."""
    total = len(records)
    mapped = sum(1 for r in records if r.status == "MAPPED")
    MAPPED.inc(mapped)
    for record in records:
        if record.status != "MAPPED":
            DEAD_LETTERED.labels(status=record.status).inc()
    if total:
        INTEGRITY_BPS.set(round(mapped / total * 10_000))

Conclusion

Attachment and parent-child mapping is not a convenience layer bolted onto ingestion; it is a bounded, depth-limited extraction subsystem that treats every container as untrusted and every extracted child as a record to bind or divert, never to guess about. By hashing at the extraction boundary, sanitizing every path before a byte is written, capping recursion at an explicit limit, and turning every failure into a typed dead-letter record rather than an exception, the engine guarantees that no attachment reaches production without a reproducible link back to its container of record — and that any challenge to that lineage can be answered from the audit trail. Its scaling limit is set by batch size and semaphore width, both tunable without weakening a single defensibility guarantee.

Frequently Asked Questions

Why hash each child at the extraction boundary instead of after the whole tree is built?

Hashing at the boundary is what keeps memory bounded and duplicates from inflating the family. If you materialize the entire tree first, peak memory scales with the largest container and every byte-identical attachment is written before you discover it is a duplicate. Boundary hashing lets the pipeline consult the deduplication registry the moment a child is extracted, so it can suppress or reference-pointer the duplicate immediately while still recording where it appeared.

What happens to a child the extractor cannot process — is it lost?

No. A child that cannot be extracted, opened, or safely placed is never dropped and never silently repaired. It becomes a MappingRecord with a typed status — EXTRACTION_FAILED, CORRUPT_ARCHIVE, PATH_TRAVERSAL_BLOCKED, or DEPTH_LIMIT_EXCEEDED — carrying the parent id, the offending path, and the depth, then routes to manual review. The rest of that container’s children still map successfully.

How should I size max_depth and the concurrency ceiling?

max_depth is a matter policy decision: five levels covers realistic “ZIP inside an email inside a PST” nesting while cutting off adversarial recursion, and crossing it is recorded rather than treated as an error. Concurrency should rise until CPU-bound hashing saturates your cores without oversubscribing the thread pool that runs the blocking reads; measure resident memory under a representative batch and leave headroom for the largest single child.

How does mapping stay intact when a parent is later suppressed as a duplicate?

The family identifier is content-anchored, not path-anchored, so a suppressed parent hands its family id to the surviving master rather than orphaning its children. Downstream stages such as Email Threading Algorithms read that identifier, which keeps attachments anchored to their originating message even after exact-match suppression collapses the duplicate copies.

Related

• Deduplication & Family Grouping — the parent pipeline whose exact-match filtering feeds unique containers into this mapping stage.
• Hash-Based Deduplication Strategies — the digest and collision rules that decide whether an extracted child is suppressed or kept.
• Email Threading Algorithms — consumes the family graph so attachments stay anchored to their originating message during threading.
• Similarity Threshold Configuration — near-duplicate routing for attachments that differ only in non-substantive bytes.
• Native File Ingestion Pipelines — the normalization step that hands containers to this subsystem.

Up: back to Deduplication & Family Grouping for how family mapping connects to hashing, threading, and defensible culling across the pipeline.