How to Simulate an FDA 24-Hour Traceability Record Request in Python

FSMA 204 gives a covered facility 24 hours to hand the FDA electronic, sortable traceability records once an outbreak request lands. The only way to know whether your system can actually meet that deadline is to rehearse it before the phone rings, and rehearsing it means a harness that runs the whole request end to end: pick a seed lot, resolve its affected set, reconstruct the chain, export the spreadsheet the agency expects, and — the part teams skip — assert the whole run finishes inside a wall-clock budget. This guide builds that harness on top of the Lot-Level Recall Scoping engine, so the drill exercises the exact scoping path a real traceback would take rather than a toy stand-in.

The trap is that a harness written the obvious way passes on a five-lot test fixture and then blows the budget on a real recall of several thousand lots. The failure is not in the compliance logic; it is in how the harness reaches the ledger. This walkthrough shows the naive version that hides the problem, the fix that makes the run scale, and the timing and row-count assertions that turn the harness into a regression test you can run every night against the FDA 24-hour response automation pipeline.

Root Cause: The Harness Round-Trips the Ledger Per Lot

A traceback touches every lot in the affected set, and the obvious harness fetches each lot’s events the moment the walk reaches it — one query per node. On a fixture that is invisible: five lots, five fast queries, done in milliseconds. On a real recall the affected set is thousands of lots deep across several supply tiers, and one network round trip per lot turns a few-second job into a multi-minute one. The wall-clock cost is dominated entirely by round-trip latency, not by the graph walk, and it grows linearly with the size of the recall — exactly the dimension a drill on a small fixture cannot see.

This is the classic N+1 access pattern, and it is dangerous here precisely because it is silent. The compliance output is correct; the harness produces the right lot set and the right chain. It just produces them too slowly to matter, and the only signal that anything is wrong is the wall clock. Because the FDA budget is measured in hours, a run that takes eight minutes still technically passes on any single lot — which is why teams discover the problem during a real recall, when the affected set is large and the eight minutes becomes ninety. The harness has to load the ledger slice once, up front, and walk it in memory, and it has to assert against a wall-clock budget scaled well below the 24-hour ceiling so a regression trips the drill instead of the FDA.

Figure — the simulated request inside its wall-clock budget:

Minimal Reproducible Example: The Naive Harness

Here is the harness written the obvious way. The ledger is stubbed with realistic KDE payloads, and fetch_events_for carries a small per-call latency to stand in for a real database round trip. The harness walks the affected set and fetches each lot’s events as it reaches them, then measures the wall clock against an internal budget set well below the 24-hour ceiling.

import time
from decimal import Decimal

# A tiny in-memory ledger keyed by (assigning_location_gln, traceability_lot_code).
LEDGER = {
    ("0086000000012", "FARM-RC-0512"): [
        {"cte_type": "Shipping", "counterparty_gln": "0086000000029",
         "reference_document": "BOL-4471", "quantity": Decimal("40"), "unit_of_measure": "case"},
    ],
    ("0086000000029", "FARM-RC-0512"): [
        {"cte_type": "Receiving", "counterparty_gln": "0086000000012",
         "reference_document": "BOL-4471", "quantity": Decimal("40"), "unit_of_measure": "case"},
        {"cte_type": "Shipping", "counterparty_gln": "0086000000036",
         "reference_document": "BOL-4490", "quantity": Decimal("22"), "unit_of_measure": "case"},
    ],
    ("0086000000036", "FARM-RC-0512"): [
        {"cte_type": "Receiving", "counterparty_gln": "0086000000029",
         "reference_document": "BOL-4490", "quantity": Decimal("22"), "unit_of_measure": "case"},
    ],
}

# Reference-document index so a shipment can find its matching receipt node.
REF_INDEX: dict[str, list[tuple[str, str]]] = {}
for node_key, events in LEDGER.items():
    for ev in events:
        ref = ev.get("reference_document")
        if ref:
            REF_INDEX.setdefault(ref, []).append(node_key)

RECORD_REQUEST_BUDGET_SECONDS = 2.0  # internal SLA proxy; the FDA ceiling is 86_400s


def fetch_events_for(node_key: tuple[str, str]) -> list[dict]:
    time.sleep(0.25)  # one network round trip per lot — the hidden cost
    return LEDGER.get(node_key, [])


def naive_record_request(seed_key: tuple[str, str]) -> tuple[list[tuple[str, str]], float]:
    start = time.perf_counter()
    seen: set[tuple[str, str]] = {seed_key}
    frontier = [seed_key]
    chain: list[tuple[str, str]] = []

    while frontier:
        node_key = frontier.pop()
        events = fetch_events_for(node_key)  # N+1: one round trip per node
        chain.append(node_key)
        for ev in events:
            ref = ev.get("reference_document")
            for peer in REF_INDEX.get(ref, []):
                if peer not in seen:
                    seen.add(peer)
                    frontier.append(peer)

    elapsed = time.perf_counter() - start
    return chain, elapsed


scoped_chain, wall_clock = naive_record_request(("0086000000012", "FARM-RC-0512"))
print(f"lots reconstructed: {len(scoped_chain)} in {wall_clock:.2f}s")
assert wall_clock < RECORD_REQUEST_BUDGET_SECONDS, "over budget"

On this three-lot fixture the harness makes three round trips and finishes in about 0.75s — comfortably under the budget, so the drill goes green. Now imagine the real recall: an affected set of 3,000 lots at the same 0.25s per round trip is 750 seconds, more than twelve minutes, and that is before the export step runs. The assertion that passed on the fixture would fail catastrophically in production, and nothing in the small test revealed it. The harness is measuring the wrong thing on the wrong scale.

Fix: Bulk-Load Once, Walk In Memory, Export Sortable

The fix removes the per-lot round trip entirely. Load the ledger slice for the traceback window in a single bulk read, build the reference-document index once, and run the whole walk against memory. Wall-clock cost then tracks the number of lots in Python-loop time — microseconds each — rather than in network round trips, so a 3,000-lot recall runs in well under a second. The corrected harness also completes the two stages the naive version omitted: reconstructing an ordered, row-per-event chain and exporting it as a sortable spreadsheet with stable column headers.

import csv
import io
import time
from decimal import Decimal


def bulk_load_ledger(window_keys: list[tuple[str, str]]) -> dict[tuple[str, str], list[dict]]:
    """One round trip for the whole traceback window instead of one per lot."""
    time.sleep(0.25)  # a single bulk read, regardless of affected-set size
    return {k: LEDGER[k] for k in window_keys if k in LEDGER}


def resolve_and_reconstruct(
    seed_key: tuple[str, str], ledger: dict[tuple[str, str], list[dict]]
) -> list[dict]:
    """Breadth-first walk over the preloaded slice; returns ordered CTE rows."""
    seen: set[tuple[str, str]] = {seed_key}
    frontier = [seed_key]
    rows: list[dict] = []

    while frontier:
        node_key = frontier.pop(0)
        gln, tlc = node_key
        for ev in ledger.get(node_key, []):
            rows.append({
                "assigning_location_gln": gln,
                "traceability_lot_code": tlc,
                "cte_type": ev["cte_type"],
                "counterparty_gln": ev["counterparty_gln"],
                "reference_document": ev["reference_document"],
                "quantity": str(ev["quantity"]),
                "unit_of_measure": ev["unit_of_measure"],
            })
            for peer in REF_INDEX.get(ev["reference_document"], []):
                if peer not in seen:
                    seen.add(peer)
                    frontier.append(peer)

    # Deterministic sort: lot, then event type, then reference document.
    rows.sort(key=lambda r: (r["traceability_lot_code"], r["cte_type"], r["reference_document"]))
    return rows


def export_sortable_sheet(rows: list[dict]) -> str:
    """Emit a sortable, column-stable CSV the FDA can filter and sort in place."""
    header = [
        "traceability_lot_code", "assigning_location_gln", "cte_type",
        "counterparty_gln", "reference_document", "quantity", "unit_of_measure",
    ]
    buffer = io.StringIO()
    writer = csv.DictWriter(buffer, fieldnames=header)
    writer.writeheader()
    writer.writerows(rows)
    return buffer.getvalue()


def simulate_record_request(seed_key: tuple[str, str], budget_s: float) -> dict:
    start = time.perf_counter()
    window = list(LEDGER.keys())          # in production, the ledger slice for the window
    ledger = bulk_load_ledger(window)     # single round trip
    rows = resolve_and_reconstruct(seed_key, ledger)
    sheet = export_sortable_sheet(rows)
    elapsed = time.perf_counter() - start

    result = {"row_count": len(rows), "elapsed_s": elapsed, "sheet": sheet}
    assert elapsed < budget_s, f"record request took {elapsed:.2f}s, over {budget_s}s budget"
    return result

The single bulk_load_ledger call replaces N round trips with one, so the harness scales with the size of the affected set in memory-loop time rather than network time. The resolve_and_reconstruct step returns one row per CTE and sorts deterministically, and export_sortable_sheet writes a header-stable CSV — the sortable format the agency asks for, and the same contract the generating FDA sortable spreadsheet exports guide hardens for production.

Verification: Assert Timing and Row Counts

A drill is only a regression test if it fails when it should. Verify three things: the run finishes inside the budget, the reconstructed chain has the expected row count, and the export is sortable with the correct header. The timing assertion is the one that catches an N+1 regression creeping back in.

def test_simulated_record_request_meets_budget():
    result = simulate_record_request(("0086000000012", "FARM-RC-0512"), budget_s=2.0)

    # Timing: the whole request finished well inside the internal budget.
    assert result["elapsed_s"] < 2.0

    # Row count: three lots, four CTE rows (one ship + one recv/ship + one recv).
    assert result["row_count"] == 4

    # Sortable export: header present and lot codes appear in sorted order.
    lines = result["sheet"].splitlines()
    assert lines[0].startswith("traceability_lot_code,")
    lot_column = [row.split(",")[0] for row in lines[1:]]
    assert lot_column == sorted(lot_column)
    print(f"drill passed: {result['row_count']} rows in {result['elapsed_s']:.3f}s")


test_simulated_record_request_meets_budget()

Run this in continuous integration with the budget pinned to a value you can defend to an auditor — a few seconds for a representative slice, scaling linearly to stay far under 86,400 seconds for a full recall. When someone reintroduces a per-lot fetch, the elapsed time jumps and the timing assertion fails in the pipeline instead of during a real FDA request. Pair the row-count assertion with a known-answer fixture so a change in scoping logic that adds or drops a lot also trips the drill, the same discipline the automating mock recall drills in Python guide builds into a scheduled job.

Related Edge Cases to Check Next

• Export serialization dominates on wide chains. Once the round trips are gone, the next bottleneck is writing the spreadsheet. For a chain with hundreds of thousands of rows, csv streaming beats building an openpyxl workbook in memory; measure the export step separately so a slow serializer does not hide behind a fast walk.
• Budget set on a fixture, not a representative slice. A two-second budget calibrated on three lots tells you nothing about a 3,000-lot recall. Calibrate the drill budget against a slice sized like a real outbreak, then scale it linearly and leave generous headroom under the regulatory ceiling.
• The bulk load itself times out on a huge window. A single query over a very wide traceback window can exceed the store’s statement timeout. Page the bulk read in bounded chunks and still assemble the index once, so you keep the single-pass walk without asking the ledger for everything in one statement — the paging concern the one-up, one-back reconstruction guide handles for very deep chains.

Frequently Asked Questions

Related

• Lot-Level Recall Scoping — the scoping engine this harness drives to resolve the affected set.
• Scoping a recall by traceability lot code — why the seed lot must be keyed by assigning GLN, not the lot code alone.
• Generating FDA sortable spreadsheet exports — hardens the export step this harness exercises.
• One-Up, One-Back Reconstruction — the chain-expansion logic the harness calls per lot.
• Automating mock recall drills in Python — schedules this harness as a recurring known-answer drill.

Up: Lot-Level Recall Scoping — this harness times the scoping engine end to end against a wall-clock budget.