Debugging Partial Ingestion Failures in LadybugDB-Backed Graphiti

What we're building: A resilient ingestion pipeline that detects mid-chunk failures, captures full execution traces, and recovers gracefully instead of leaving partial documents.

Prerequisites

• Python 3.10+ with graphiti-core installed
• LadybugDB running locally or remotely
• A TracePilot API key (free at tracepilotai.com)
• Your existing ingestion script (we'll harden it)

The Problem

You're ingesting 760 docs. Chunk 47 fails with unordered_map::at: key not found. Chunks 1-46 are committed. Chunks 48+ never run. Your corpus is now silently incomplete.

The root cause is a C++ exception in LadybugDB's internal hash map — likely a race condition or corrupted state during batch writes. The Python wrapper doesn't cleanly translate this, so your try/except may miss it entirely.

Step 1: Instrument Your Ingestion Loop

First, let's wrap your existing ingestion with proper error capture. Here's the pattern:

from graphiti_core import Graphiti
from tracepilot_sdk import TracePilot
import traceback

tp = TracePilot("tp_live_YOUR_KEY")
graphiti = Graphiti(...)  # your existing config

async def ingest_document(doc_id: str, chunks: list[str]):
    await tp.start_trace(f"ingest-{doc_id}")

    ingested = 0
    failed_chunks = []

    for idx, chunk in enumerate(chunks):
        span_id = None
        try:
            # Wrap each chunk in its own span
            result, span_id = await tp.wrap_async(
                "chunk-ingest",
                lambda: graphiti.add_entity(chunk),
                parent_span_id=None,
                step_order=idx
            )
            ingested += 1

        except Exception as e:
            # Capture the full trace even on failure
            error_details = {
                "chunk_index": idx,
                "error_type": type(e).__name__,
                "error_msg": str(e),
                "native_trace": traceback.format_exc()
            }

            # Log this to TracePilot with the error context
            await tp.log_error(
                span_id=span_id,
                error=error_details,
                severity="critical"
            )

            failed_chunks.append({
                "index": idx,
                "error": error_details
            })

            # CRITICAL: Don't continue silently
            # Mark this trace as failed
            await tp.mark_trace_failed(
                reason=f"chunk {idx} failed",
                failed_chunks=failed_chunks
            )

            # Option: break or continue based on your strategy
            # break  # stop entirely
            continue  # skip and move on

    # Record final state
    await tp.log_metric("chunks_ingested", ingested)
    await tp.log_metric("chunks_failed", len(failed_chunks))

    return {
        "doc_id": doc_id,
        "ingested": ingested,
        "failed": failed_chunks,
        "partial": len(failed_chunks) > 0
    }

Step 2: Detect the Native Exception

The tricky part: unordered_map::at might not raise a standard Python exception. It can surface as a SystemError or even a segfault. Here's how to catch it robustly:

import faulthandler
import signal

# Enable fault handler to catch segfaults
faulthandler.enable()

class LadybugErrorHandler:
    def __init__(self, tp: TracePilot):
        self.tp = tp
        self.segv_count = 0

    def handle_native_error(self, chunk_idx: int, doc_id: str):
        """Called when we detect the native error pattern"""
        self.segv_count += 1

        # Log to TracePilot with full context
        self.tp.log_event(
            "native_error_detected",
            {
                "doc_id": doc_id,
                "chunk_idx": chunk_idx,
                "segv_count": self.segv_count,
                "pattern": "unordered_map::at"
            }
        )

        # If we see this repeatedly, suggest a restart
        if self.segv_count > 3:
            self.tp.log_alert(
                "ladybugdb_state_corrupted",
                "Multiple native errors — consider restarting LadybugDB"
            )

# Use it in your loop
error_handler = LadybugErrorHandler(tp)

# Wrap the loop with signal handling for segfaults
def safe_ingest(doc_id, chunks):
    try:
        return asyncio.run(ingest_document(doc_id, chunks))
    except SystemError as e:
        if "unordered_map" in str(e):
            error_handler.handle_native_error(-1, doc_id)
            return {"doc_id": doc_id, "fatal": True, "error": str(e)}
        raise

Step 3: Add Checkpointing for Resume

Don't re-ingest 46 good chunks when chunk 47 fails. Persist progress:

import json
from pathlib import Path

class IngestionCheckpoint:
    def __init__(self, checkpoint_file: str):
        self.file = Path(checkpoint_file)
        self.state = self._load()

    def _load(self) -> dict:
        if self.file.exists():
            return json.loads(self.file.read_text())
        return {"completed_docs": {}, "failed_chunks": {}}

    def save(self):
        self.file.write_text(json.dumps(self.state, indent=2))

    def mark_chunk_done(self, doc_id: str, chunk_idx: int):
        self.state["completed_docs"].setdefault(doc_id, [])
        if chunk_idx not in self.state["completed_docs"][doc_id]:
            self.state["completed_docs"][doc_id].append(chunk_idx)
        self.save()

    def mark_chunk_failed(self, doc_id: str, chunk_idx: int, error: dict):
        self.state["failed_chunks"].setdefault(doc_id, {})
        self.state["failed_chunks"][doc_id][str(chunk_idx)] = error
        self.save()

    def get_resume_point(self, doc_id: str) -> int:
        """Returns the next chunk index to process"""
        done = self.state["completed_docs"].get(doc_id, [])
        return max(done) + 1 if done else 0

# Integrate into your loop
checkpoint = IngestionCheckpoint("ingestion_state.json")

# In ingest_document, before processing:
start_idx = checkpoint.get_resume_point(doc_id)
for idx in range(start_idx, len(chunks)):
    # ... existing logic ...

    # After successful ingest:
    checkpoint.mark_chunk_done(doc_id, idx)

    # On failure:
    checkpoint.mark_chunk_failed(doc_id, idx, error_details)

Adding Observability

You've already seen TracePilot integrated above. The key insight: you get the exact state of the failed chunk — input, output, error, and the native trace — without redeploying.

Install it:

pip install tracepilot-sdk

That's it. Every chunk ingest is now a span in your dashboard. When chunk 47 fails, you see:

• The exact chunk content that triggered it
• The full error trace including the C++ exception
• Token counts and timing for all previous chunks
• A Fork & Rerun button to test a modified chunk immediately

Next Steps

1. Run a batch test — ingest 10 docs and verify checkpoints work
2. Simulate a failure — temporarily break a chunk to test recovery
3. Set up alerts — configure TracePilot to notify you when native_error_detected fires more than twice
4. Consider batching — if LadybugDB has race conditions, try ingesting chunks with small delays (e.g., await asyncio.sleep(0.1) between chunks)

The combination of

---

Debugging AI agents shouldn't feel like reading The Matrix.
Join other engineers who are building reliable autonomous workflows in our community: TracePilot Discord