Skip to main content

Supply Chain Traceability and Recall Analysis

When a product defect is discovered, two questions need immediate answers:

  1. Upstream impact: which raw materials, batches, and suppliers contributed to the affected product?
  2. Downstream blast radius: which shipments, orders, and customers received products from this batch?

These questions cross multiple hops in a causal chain. A graph database handles them naturally. A set of JOIN-heavy relational tables does not — not at the speed required during a recall incident.

Graph shape​

LabelWhat it represents
SUPPLIERA vendor or raw material source
BATCHA specific lot of raw material or component
PRODUCTION_RUNA manufacturing run that consumed batches
PRODUCTA finished product unit or SKU
SHIPMENTA delivery fulfillment
CUSTOMERA receiving customer or distribution center
INCIDENTA quality or safety report

Step 1: Ingest supply chain records​

from rushdb import RushDB
import os

db = RushDB(os.environ["RUSHDB_API_KEY"], base_url="https://api.rushdb.com/api/v1")

db.records.import_json({
    "label": "BATCH",
    "data": [
        {"lotId": "LOT-2025-001", "material": "polymer-Z", "producedAt": "2025-01-10", "quantity": 5000},
        {"lotId": "LOT-2025-002", "material": "polymer-Z", "producedAt": "2025-01-17", "quantity": 4800},
        {"lotId": "LOT-2025-003", "material": "alloy-X",   "producedAt": "2025-01-20", "quantity": 2000}
    ]
})

db.records.import_json({
    "label": "PRODUCTION_RUN",
    "data": [
        {"runId": "RUN-A1", "startedAt": "2025-02-01", "completedAt": "2025-02-03", "facilityId": "FAC-Berlin"},
        {"runId": "RUN-A2", "startedAt": "2025-02-05", "completedAt": "2025-02-07", "facilityId": "FAC-Berlin"}
    ]
})

db.records.import_json({
    "label": "PRODUCT",
    "data": [
        {"sku": "PROD-001", "name": "Widget Alpha", "serialRange": "WA-10001:WA-11000"},
        {"sku": "PROD-002", "name": "Widget Beta",  "serialRange": "WB-20001:WB-20500"}
    ]
})

db.records.import_json({
    "label": "SHIPMENT",
    "data": [
        {"trackingId": "SHIP-4001", "shippedAt": "2025-02-15", "status": "delivered"},
        {"trackingId": "SHIP-4002", "shippedAt": "2025-02-18", "status": "in_transit"}
    ]
})
batches  = db.records.find({"labels": ["BATCH"]})
runs     = db.records.find({"labels": ["PRODUCTION_RUN"]})
products = db.records.find({"labels": ["PRODUCT"]})
shipments = db.records.find({"labels": ["SHIPMENT"]})

batch_map   = {b.data["lotId"]:  b for b in batches.data}
run_map     = {r.data["runId"]:  r for r in runs.data}
product_map = {p.data["sku"]:    p for p in products.data}

db.records.attach(batch_map["LOT-2025-001"].id, run_map["RUN-A1"].id,     {"type": "USED_IN",     "direction": "out"})
db.records.attach(batch_map["LOT-2025-002"].id, run_map["RUN-A2"].id,     {"type": "USED_IN",     "direction": "out"})
db.records.attach(run_map["RUN-A1"].id,         product_map["PROD-001"].id, {"type": "PRODUCED",  "direction": "out"})
db.records.attach(run_map["RUN-A2"].id,         product_map["PROD-002"].id, {"type": "PRODUCED",  "direction": "out"})
db.records.attach(product_map["PROD-001"].id,   shipments.data[0].id,       {"type": "INCLUDED_IN","direction": "out"})
db.records.attach(product_map["PROD-002"].id,   shipments.data[1].id,       {"type": "INCLUDED_IN","direction": "out"})

Step 3: Downstream blast radius — all shipments from a contaminated batch​

Given a defective batch, find every shipment that could contain affected product.

affected_shipments = db.records.find({
    "labels": ["SHIPMENT"],
    "where": {
        "PRODUCT": {
            "$relation": {"type": "INCLUDED_IN", "direction": "in"},
            "PRODUCTION_RUN": {
                "$relation": {"type": "PRODUCED", "direction": "in"},
                "BATCH": {
                    "$relation": {"type": "USED_IN", "direction": "in"},
                    "lotId": "LOT-2025-001"
                }
            }
        }
    }
})

print(f"Affected shipments: {affected_shipments.total}")
for s in affected_shipments.data:
    print(f"  {s.data.get('trackingId')} — {s.data.get('status')}")

Step 4: Upstream impact — which batches contributed to an affected product?​

Given a known-bad product SKU, trace back to every batch that could have contributed.

source_batches = db.records.find({
    "labels": ["BATCH"],
    "where": {
        "PRODUCTION_RUN": {
            "$relation": {"type": "USED_IN", "direction": "in"},
            "PRODUCT": {
                "$relation": {"type": "PRODUCED", "direction": "out"},
                "sku": "PROD-001"
            }
        }
    }
})

print(f"Source batches for PROD-001: {source_batches.total}")
for batch in source_batches.data:
    print(f"  {batch.data.get('lotId')} — {batch.data.get('material')}")

Step 5: Incident report — log a quality incident linked to a batch​

from datetime import datetime, timezone

incident = db.records.create("INCIDENT", {
    "title": "Polymer degradation detected in lot LOT-2025-001",
    "severity": "critical",
    "reportedAt": datetime.now(timezone.utc).isoformat(),
    "status": "open"
})

batch_result = db.records.find({"labels": ["BATCH"], "where": {"lotId": "LOT-2025-001"}})
db.records.attach(incident.id, batch_result.data[0].id, {"type": "CAUSED_BY", "direction": "out"})

Production caveat​

Real supply chains are many-to-many: a production run may consume dozens of batches; a shipment may contain hundreds of products. The traversal queries above traverse exactly three hops. Adding more hops increases execution cost proportionally. If your supply chain has more than three levels of indirection between primary inputs and customer delivery, benchmark traversal performance against your production data volume before deploying to incident response workflows.

Next steps​