Capsule 04: Backup and Disaster Recovery

Capsule description

A RAG system without a tested backup is not production-ready. This capsule guides you to define backup, retention, and recovery with clear time objectives, automation scripts for ChromaDB, and step-by-step procedures you can execute in a real incident.

Estimated time: 45-60 minutes

RTO and RPO for RAG systems

What do they mean?

• RPO (Recovery Point Objective): the maximum amount of data you can lose without compromising the business's integrity. In RAG it means: documents, embeddings, and metadata that wouldn't be in the last backup.
• RTO (Recovery Time Objective): the maximum time you accept the system being down or degraded until it recovers.

Suggested objectives by criticality

A reasonable initial example

• RPO: 24 hours (daily backup)
• RTO: 2 hours (documented and tested restore)

ChromaDB data structure

Before backing up, it helps to know what you're copying:

./chroma_db/
├── chroma.sqlite3       # Metadata, IDs, collection config
├── index/               # HNSW indexes (fast search)
│   └── <collection_id>.bin
└── data/                # Vectors and documents
    └── <collection_id>.parquet

Important: ChromaDB requires the client to be closed for consistent file-copy backups. Otherwise, you may get a corrupt backup.

Backup and export scripts for ChromaDB

1. Copy backup (offline)

Ideal for a complete and fast restore. The client must be closed.

# scripts/backup_chromadb.py
import os
import shutil
from datetime import datetime
from pathlib import Path


def backup_chromadb(
    source_path: str,
    backup_dir: str,
    prefix: str = "chroma_backup",
) -> str:
    """
    Copy the entire ChromaDB folder. Use when the process
    that uses ChromaDB is stopped.
    """
    source = Path(source_path)
    if not source.exists():
        raise FileNotFoundError(f"Does not exist: {source_path}")

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"{prefix}_{timestamp}"
    backup_path = Path(backup_dir) / backup_name

    Path(backup_dir).mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, backup_path)

    print(f"Backup created: {backup_path}")
    return str(backup_path)


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--source", default="./chroma_db")
    parser.add_argument("--backup-dir", default="./backups")
    args = parser.parse_args()
    backup_chromadb(args.source, args.backup_dir)

2. Per-collection export (online)

Useful for migrating a collection or backing up without stopping the service (the data is read via the API).

# scripts/export_collection.py
import json
import chromadb
from pathlib import Path


def export_collection(
    chroma_path: str,
    collection_name: str,
    output_file: str,
    include_embeddings: bool = True,
) -> int:
    """
    Export a collection to JSON. Works with the service running.
    Returns the number of documents exported.
    """
    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_collection(collection_name)

    include = ["documents", "metadatas"]
    if include_embeddings:
        include.append("embeddings")

    data = collection.get(include=include)

    payload = {
        "collection": collection_name,
        "count": len(data["ids"]),
        "ids": data["ids"],
        "documents": data.get("documents", []),
        "metadatas": data.get("metadatas", []),
    }
    if include_embeddings and data.get("embeddings"):
        payload["embeddings"] = data["embeddings"]

    Path(output_file).parent.mkdir(parents=True, exist_ok=True)
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)

    print(f"Exported {len(data['ids'])} docs to {output_file}")
    return len(data["ids"])


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", default="./chroma_db")
    parser.add_argument("--collection", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    export_collection(args.chroma_path, args.collection, args.output)

3. Import from JSON

# scripts/import_collection.py
import json
import chromadb
from pathlib import Path


def import_collection(
    chroma_path: str,
    collection_name: str,
    json_file: str,
    batch_size: int = 500,
) -> int:
    """
    Import a collection from a JSON backup.
    Returns the number of documents imported.
    """
    with open(json_file, "r", encoding="utf-8") as f:
        data = json.load(f)

    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_or_create_collection(collection_name)

    ids = data["ids"]
    documents = data.get("documents", [])
    metadatas = data.get("metadatas", [])
    embeddings = data.get("embeddings")

    n = 0
    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i : i + batch_size]
        batch_docs = documents[i : i + batch_size] if documents else None
        batch_meta = metadatas[i : i + batch_size] if metadatas else None
        batch_emb = embeddings[i : i + batch_size] if embeddings else None

        kwargs = {"ids": batch_ids}
        if batch_docs:
            kwargs["documents"] = batch_docs
        if batch_meta:
            kwargs["metadatas"] = batch_meta
        if batch_emb:
            kwargs["embeddings"] = batch_emb

        collection.add(**kwargs)
        n += len(batch_ids)

    print(f"Imported {n} docs to {collection_name}")
    return n

Automatic daily backup with 30-day retention

Complete automation script

# scripts/backup_automation.py
"""
Daily ChromaDB backup with a retention policy.
Run via cron or a systemd timer.
"""
import argparse
import logging
import os
import shutil
import sys
from datetime import datetime, timedelta
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
    ],
)
logger = logging.getLogger(__name__)

# Default configuration
DEFAULT_SOURCE = os.environ.get("CHROMA_DB_PATH", "./chroma_db")
DEFAULT_BACKUP_DIR = os.environ.get("CHROMA_BACKUP_DIR", "./backups")
DEFAULT_RETENTION_DAYS = int(os.environ.get("CHROMA_BACKUP_RETENTION_DAYS", "30"))


def run_backup(source: str, backup_dir: str) -> str:
    """Run a directory-copy backup."""
    source_path = Path(source)
    if not source_path.exists():
        raise FileNotFoundError(f"ChromaDB directory does not exist: {source}")

    backup_root = Path(backup_dir)
    backup_root.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"chroma_backup_{timestamp}"
    backup_path = backup_root / backup_name

    shutil.copytree(source_path, backup_path)
    logger.info("Backup created: %s", backup_path)
    return str(backup_path)


def apply_retention(backup_dir: str, keep_days: int) -> int:
    """
    Delete backups older than keep_days.
    Returns the number of backups deleted.
    """
    backup_root = Path(backup_dir)
    if not backup_root.exists():
        return 0

    cutoff = datetime.now() - timedelta(days=keep_days)
    cutoff_ts = cutoff.timestamp()
    deleted = 0

    for item in backup_root.iterdir():
        if item.is_dir() and item.name.startswith("chroma_backup_"):
            if item.stat().st_mtime < cutoff_ts:
                shutil.rmtree(item)
                logger.info("Old backup deleted: %s", item.name)
                deleted += 1

    return deleted


def main():
    parser = argparse.ArgumentParser(description="ChromaDB backup with retention")
    parser.add_argument("--source", default=DEFAULT_SOURCE)
    parser.add_argument("--backup-dir", default=DEFAULT_BACKUP_DIR)
    parser.add_argument("--retention-days", type=int, default=DEFAULT_RETENTION_DAYS)
    parser.add_argument("--retention-only", action="store_true", help="Only clean up, don't back up")
    args = parser.parse_args()

    try:
        if not args.retention_only:
            run_backup(args.source, args.backup_dir)
        deleted = apply_retention(args.backup_dir, args.retention_days)
        logger.info("Retention: %d backups deleted", deleted)
    except Exception as e:
        logger.exception("Backup error: %s", e)
        sys.exit(1)


if __name__ == "__main__":
    main()

Cron configuration

Run a daily backup at 02:00 and apply retention:

# /etc/cron.d/chromadb-backup
# Daily ChromaDB backup at 02:00, 30-day retention
0 2 * * * cd /opt/rag-app && /usr/bin/python3 scripts/backup_automation.py --source /opt/rag-app/chroma_db --backup-dir /opt/backups/chroma --retention-days 30 >> /var/log/chromadb-backup.log 2>&1

With environment variables (recommended):

# In /opt/rag-app/.env or systemd
export CHROMA_DB_PATH=/opt/rag-app/chroma_db
export CHROMA_BACKUP_DIR=/opt/backups/chroma
export CHROMA_BACKUP_RETENTION_DAYS=30

0 2 * * * source /opt/rag-app/.env && /usr/bin/python3 /opt/rag-app/scripts/backup_automation.py >> /var/log/chromadb-backup.log 2>&1

systemd timer (alternative to cron)

# /etc/systemd/system/chromadb-backup.service
[Unit]
Description=ChromaDB daily backup
After=network.target

[Service]
Type=oneshot
User=rag-app
WorkingDirectory=/opt/rag-app
Environment="CHROMA_DB_PATH=/opt/rag-app/chroma_db"
Environment="CHROMA_BACKUP_DIR=/opt/backups/chroma"
Environment="CHROMA_BACKUP_RETENTION_DAYS=30"
ExecStart=/usr/bin/python3 /opt/rag-app/scripts/backup_automation.py

# /etc/systemd/system/chromadb-backup.timer
[Unit]
Description=ChromaDB daily backup
Requires=chromadb-backup.service

[Timer]
OnCalendar=*-*-* 02:00:00
Persistent=true

[Install]
WantedBy=timers.target

sudo systemctl enable chromadb-backup.timer
sudo systemctl start chromadb-backup.timer

Backup validation

A backup without validation can fail when you need it most. This script verifies basic integrity:

# scripts/validate_backup.py
"""
Validate that a ChromaDB backup is restorable.
"""
import argparse
import sys
import chromadb
from pathlib import Path


def validate_backup(backup_path: str) -> bool:
    """
    Try to load the backup as a ChromaDB client.
    Returns True if it's valid.
    """
    path = Path(backup_path)
    if not path.exists():
        print(f"❌ Does not exist: {backup_path}")
        return False

    # Minimum expected files
    sqlite = path / "chroma.sqlite3"
    if not sqlite.exists():
        print(f"❌ Missing chroma.sqlite3")
        return False

    try:
        client = chromadb.PersistentClient(path=str(backup_path))
        collections = client.list_collections()
        total_docs = 0
        for col in collections:
            count = col.count()
            total_docs += count
            print(f"  - {col.name}: {count} documents")
        print(f"✅ Valid backup: {len(collections)} collections, {total_docs} documents")
        return True
    except Exception as e:
        print(f"❌ Corrupt backup: {e}")
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("backup_path")
    args = parser.parse_args()
    ok = validate_backup(args.backup_path)
    sys.exit(0 if ok else 1)

Integrate validation into the backup flow:

# Add at the end of run_backup() in backup_automation.py
def run_backup_with_validation(source: str, backup_dir: str) -> str:
    backup_path = run_backup(source, backup_dir)
    if not validate_backup(backup_path):
        Path(backup_path).rmdir()
        raise RuntimeError("Backup failed validation, deleted")
    return backup_path

Step-by-step recovery procedures

Scenario 1: Total loss of the ChromaDB directory

Estimated time: 15-30 min (depends on the backup size)

1. Stop the RAG service (2 min)

   sudo systemctl stop rag-api
   # or pm2 stop rag-api, or docker stop rag-container

2. Identify the most recent backup (1 min)

   ls -lt /opt/backups/chroma/ | head -5

3. Make a backup of the current state (1 min)

   mv /opt/rag-app/chroma_db /opt/rag-app/chroma_db.corrupt.$(date +%Y%m%d)

4. Restore the backup (5-20 min depending on size)

   cp -r /opt/backups/chroma/chroma_backup_20240313_020000 /opt/rag-app/chroma_db

5. Validate the restore (2 min)

   python scripts/validate_backup.py /opt/rag-app/chroma_db

6. Restart the service (1 min)

   sudo systemctl start rag-api

7. Smoke test (3 min)

   • Run 3-5 test queries against the API
   • Verify that the answers are coherent

Scenario 2: Restore in a staging/test environment

Estimated time: 10-20 min

1. Create a test directory
2. Copy the backup to that directory
3. Validate with validate_backup.py
4. Point the staging service to the restored directory
5. Run automated smoke tests

Scenario 3: Restore a collection from JSON

Estimated time: 5-15 min (depending on the collection size)

1. Identify the collection's JSON
2. Create or empty the target collection
3. Run import_collection.py
4. Re-index if you use on-the-fly embeddings (less frequent case)

Game day: disaster simulation

A game day is an exercise where you simulate a failure and execute the real runbook.

Simulation script

# scripts/game_day_simulation.py
"""
Simulate index loss and recovery.
RUN ONLY IN A TEST ENVIRONMENT.
"""
import os
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path


def game_day_simulation(
    chroma_path: str,
    backup_dir: str,
    validate_script: str = "scripts/validate_backup.py",
) -> bool:
    """
    1. Back up the current state
    2. Delete chroma_path (simulate loss)
    3. Restore from the latest backup
    4. Validate
    Returns True if everything is ok.
    """
    chroma = Path(chroma_path)
    if not chroma.exists():
        print("❌ chroma_path does not exist, nothing to simulate")
        return False

    # 1. Additional safety backup
    safe_backup = Path(backup_dir) / f"game_day_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    shutil.copytree(chroma, safe_backup)
    print(f"✓ Safety backup: {safe_backup}")

    # 2. Simulate loss
    shutil.rmtree(chroma)
    chroma.mkdir(parents=True)
    print("✓ Directory emptied (loss simulation)")

    # 3. Find the latest backup
    backups = sorted(
        Path(backup_dir).glob("chroma_backup_*"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    if not backups:
        print("❌ No backups to restore")
        return False

    latest = backups[0]
    print(f"✓ Restoring from: {latest}")

    # 4. Restore
    for item in latest.iterdir():
        dest = chroma / item.name
        if item.is_dir():
            shutil.copytree(item, dest)
        else:
            shutil.copy2(item, dest)

    # 5. Validate
    result = subprocess.run(
        [sys.executable, validate_script, str(chroma)],
        capture_output=True,
        text=True,
    )
    print(result.stdout)
    if result.returncode != 0:
        print(result.stderr)
        return False

    print("✅ Game day completed: successful restoration")
    return True


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", default="./chroma_db")
    parser.add_argument("--backup-dir", default="./backups")
    args = parser.parse_args()
    ok = game_day_simulation(args.chroma_path, args.backup_dir)
    sys.exit(0 if ok else 1)

Post-recovery smoke tests

# scripts/smoke_test_restore.py
"""
Smoke test queries after a restore.
"""
import chromadb
import sys


def smoke_test(chroma_path: str, collection_name: str = None) -> bool:
    client = chromadb.PersistentClient(path=chroma_path)
    collections = client.list_collections()

    if not collections:
        print("❌ No collections")
        return False

    col_name = collection_name or collections[0].name
    col = client.get_collection(col_name)

    # Simple query
    results = col.query(query_texts=["test"], n_results=min(3, col.count()))
    print(f"✓ Smoke query: {len(results['ids'][0])} results")

    # Verify there are documents
    if col.count() == 0:
        print("❌ Empty collection")
        return False

    print("✅ Smoke test passed")
    return True


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else "./chroma_db"
    ok = smoke_test(path)
    sys.exit(0 if ok else 1)

Suggested criticality levels

DR troubleshooting

1. "We have a backup, but we never restore"

Without a restore test there's no guarantee. Backups can be corrupt, in the wrong paths, or with wrong permissions. Action: schedule a monthly test restore at minimum and document the real time.

2. "The restore takes too long"

Possible causes: a very large backup, a slow disk, a slow network (if the backup is in remote storage). Action: measure times per stage (copy, validate, bring up the service) and optimize the bottleneck. Consider incremental or per-collection backups if only one fails.

3. "The team doesn't know how to operate the runbook"

Runbooks rust if they aren't used. Action: run a guided game day quarterly, rotate who leads the exercise, and update the runbook with the findings.

4. "The backup fails because the disk is full"

Retention must be tied to available space. Action: monitor space in /opt/backups, alert when less than 20% is free, and adjust retention_days or add more aggressive cleanup.

5. "Corrupt backup after copying with the service active"

ChromaDB writes to SQLite and files; a copy with the process active can leave inconsistent data. Action: for a copy backup, stop the service or use per-collection export (which reads via the API consistently).

Validation checklist

• I can list available backups with clear dates.
• I can restore in a test environment in less than the defined RTO.
• The recovered system answers smoke-test queries correctly.
• The team knows the runbook and has practiced at least once.
• The backups are validated automatically (or manually on a periodic basis).
• Retention is applied correctly (30 days or as defined).

Exercises

Exercise 1: First manual backup

Goal: Make your first backup of a ChromaDB instance.

1. Create a collection with 5 test documents.
2. Stop any process that uses ChromaDB.
3. Run the backup_chromadb.py script with --source and --backup-dir.
4. Verify that the backup directory exists with chroma.sqlite3 and index/.

# Create test data
import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
col = client.get_or_create_collection("test_backup")
col.add(
    documents=["Doc A", "Doc B", "Doc C", "Doc D", "Doc E"],
    ids=["1", "2", "3", "4", "5"]
)
# Close Python to release the lock

# In the terminal:
# python scripts/backup_chromadb.py --source ./chroma_db --backup-dir ./backups
# ls -la ./backups/chroma_backup_*/

Exercise 2: Restore in a new directory

Goal: Restore a backup in a different directory and validate.

1. Copy a backup to ./chroma_restored.
2. Run validate_backup.py on that directory.
3. Open a ChromaDB client pointing to ./chroma_restored and run a query.

cp -r ./backups/chroma_backup_20240313_020000 ./chroma_restored
python scripts/validate_backup.py ./chroma_restored

import chromadb
client = chromadb.PersistentClient(path="./chroma_restored")
col = client.get_collection("test_backup")
print(col.query(query_texts=["Doc"], n_results=3))

Exercise 3: 7-day retention

Goal: Modify the retention script to keep only 7 days and test it.

1. Manually create several chroma_backup_* directories with old dates (use touch -d to simulate).
2. Run backup_automation.py --retention-only --retention-days 7.
3. Check that the old backups were deleted.

# Create fake backups with old dates
mkdir -p backups/chroma_backup_20240301_020000
mkdir -p backups/chroma_backup_20240305_020000
touch -d "2024-02-28" backups/chroma_backup_20240301_020000
touch -d "2024-03-02" backups/chroma_backup_20240305_020000

# Run retention only
python scripts/backup_automation.py --backup-dir ./backups --retention-days 7 --retention-only

# Verify: the old February/March ones should have been deleted
ls -la backups/

Exercise 4: Export and import a collection

Goal: Export a collection to JSON and import it into a new base.

1. Export the test_backup collection to ./exports/test_backup.json.
2. Create a client pointing to ./chroma_new.
3. Import the JSON into a collection with the same name.
4. Compare the count() of the original collection and the new one.

python scripts/export_collection.py --chroma-path ./chroma_db --collection test_backup --output ./exports/test_backup.json

# import_collection uses the script or:
import scripts.import_collection as imp
imp.import_collection("./chroma_new", "test_backup", "./exports/test_backup.json")

# Verification
import chromadb
c_orig = chromadb.PersistentClient(path="./chroma_db").get_collection("test_backup")
c_new = chromadb.PersistentClient(path="./chroma_new").get_collection("test_backup")
assert c_orig.count() == c_new.count()

Exercise 5: Game day in a local environment

Goal: Run the game day without touching production.

1. Use a ./chroma_test directory with test data.
2. Make a prior manual backup.
3. Run game_day_simulation.py with --chroma-path ./chroma_test.
4. Run smoke_test_restore.py on the restored directory.
5. Record the total time and compare it with your RTO.

# Prepare the environment
cp -r ./chroma_db ./chroma_test

# Run the simulation
time python scripts/game_day_simulation.py --chroma-path ./chroma_test --backup-dir ./backups

# Smoke test
python scripts/smoke_test_restore.py ./chroma_test

Exercise 6: Local cron (Linux/Mac only)

Goal: Configure a cron that runs the daily backup.

1. Edit your crontab: crontab -e.
2. Add the backup line at 02:00 (or at 5 minutes for a test).
3. For a quick test, use */5 * * * * (every 5 min) and check that backups are created.
4. Change to 0 2 * * * once you confirm it works.

# Test every 5 minutes (change later)
*/5 * * * * cd /Users/you/project && python scripts/backup_automation.py >> /tmp/chroma-backup.log 2>&1

# Production
0 2 * * * cd /opt/rag-app && python scripts/backup_automation.py >> /var/log/chromadb-backup.log 2>&1

Summary

• RPO defines how much data you can lose; RTO, how long you can take to recover. For RAG, an RPO of 24h and RTO of 2h is a reasonable starting point.
• ChromaDB can be backed up by directory copy (offline) or by JSON export/import per collection (online).
• Implement automated daily backup with 30-day retention using a Python script and cron or a systemd timer.
• Validate the backups after creating them or periodically; an unvalidated backup can fail at the worst moment.
• Document step-by-step recovery procedures with estimated times and run game days to practice.
• The runbook must be accessible and the team must have practiced it at least once.
• Without a restore test there's no guarantee the plan works; schedule it and adjust it based on the results.

Additional resources

• Disaster recovery planning - Google Cloud
• ChromaDB persistence docs
• AWS Backup best practices
• SQLite backup (ChromaDB internals)
• Cron tutorial
• systemd timers
• RTO/RPO in practice - Backblaze
• Chaos engineering - principles

Estimated time: 45-60 minutes
Next: 05-rag-security.md