04 — Stale Index Problem

Answer:

The stale index problem occurs when the knowledge base (vector index, documents, metadata) is not synchronized with the source of truth. The system returns correct from the index's perspective but incorrect in reality.

Example:

Source of truth (Real-world):
  Apple CEO: Tim Cook (as of 2024)

Stale index (Last updated: 2021):
  Apple CEO: Tim Cook... wait, was it Steve Jobs?
  
User query: "Who is the CEO of Apple?"
System returns: (stale data from index)
User accepts answer as factual

Why it's critical:

1. User trust erosion: Incorrect information damages credibility
2. Downstream impact: Decisions based on stale data lead to failures
3. Compliance risk: In regulated domains (finance, healthcare), stale data can violate SLAs
4. Silent failure: Unlike a crash, stale data doesn't trigger alarms—it's just wrong
5. Compounding errors: Stale index + hallucination = high confidence in false claims

This is distinct from retrieval failure (missing data) because the data exists and is findable, but it's just old.

Answer:

Stale index symptoms manifest through multiple detection signals:

Production signals:

def detect_stale_index_signals(query, retrieved_doc, real_time_source):
    """Flag potential staleness."""
    
    # Signal 1: Doc version mismatch
    doc_timestamp = retrieved_doc.get('updated_at')
    current_time = datetime.now()
    days_old = (current_time - doc_timestamp).days
    
    if days_old > 30:  # Threshold
        log_staleness_alert(f"Doc is {days_old} days old", severity='medium')
    
    # Signal 2: Answer contradicts real-time source
    real_value = real_time_source.query(query)
    answer_value = extract_value_from_answer(retrieved_doc)
    
    if real_value != answer_value:
        log_staleness_alert("Answer contradicts source of truth", severity='high')
    
    # Signal 3: Entity list mismatch
    entities_in_doc = extract_entities(retrieved_doc)
    entities_in_real_time = real_time_source.get_entities()
    
    if len(entities_in_doc) < len(entities_in_real_time) * 0.9:
        log_staleness_alert("Index is missing entities", severity='high')

Answer:

Staleness has multiple independent causes across indexing frequency, update lag, and deletion handling:

1. Infrequent Indexing

Index is built once, never updated:

Timeline:
  Jan 1: Index built with 10,000 products
  Jan 2-Dec 31: Products change in source (prices, descriptions, deletions)
  Dec 31: Index still contains original 10,000 products
  
Problem: 12 months of drift, no synchronization

Root cause: Full re-indexing is expensive:

# Full reindex of 1M documents
start_time = time.time()
index.clear()  # Wipe old index
for doc in corpus:
    doc_embedding = embedding_model.encode(doc['text'])
    index.add(doc_embedding)
elapsed = time.time() - start_time
# Duration: 1-2 hours for large corpus

During re-index, old index is offline or stale.

2. Incremental Update Lag

Updates are queued but not immediately reflected:

Event timeline:
  12:00 PM: Product A's price changes ($50 → $45)
  12:00:05 PM: Database updated
  12:00:10 PM: Update event published to message queue
  12:00:15 PM: Worker picks up update (queue lag)
  12:00:20 PM: Embedding computed
  12:00:30 PM: Index updated
  
User query at 12:00:25 PM:
  → Index hasn't been updated yet (5-30 seconds stale)
  
Worse case: Multiple updates queued, worker falls behind
  → Index becomes hours/days stale

3. Deletion Handling Problems

Documents are deleted from source but remain in index:

Scenario: E-commerce product catalog
  
Database:
  - Product A: Active
  - Product B: Deleted
  - Product C: Active

Stale index:
  - Product A embedding
  - Product B embedding (still present, should be gone)
  - Product C embedding

User query: "Show all products"
Retrieval returns Product B (now unavailable)

4. Metadata Staleness

Metadata (prices, availability, ratings) updated separately from embeddings:

Document structure:
  {
    'id': 'product_123',
    'text': 'iPhone 15 Pro...',
    'embedding': [...],
    'metadata': {
      'price': 999,          # Last updated: 3 months ago
      'stock': 50,
      'rating': 4.5
    }
  }

Real-world price: $799 (sale started yesterday)
Index price: $999 (outdated)

Embedding (based on text) is current, but metadata is stale.

5. Distributed System Lag

In distributed index (shards), updates reach shards at different times:

Shard 1: Updated to v42 at 12:00:00
Shard 2: Updated to v42 at 12:00:30 (network lag)
Shard 3: Still on v41 at 12:00:45

Query hits shards at 12:00:35:
  → Shard 1: Returns latest (v42)
  → Shard 2: Returns v42
  → Shard 3: Returns old data (v41)
  
Results are inconsistent.

6. Batch vs. Real-Time Mismatch

Index built from batch ETL (overnight), but source updates continuously:

Typical ETL pipeline:
  11 PM: Extract all documents from source
  11:30 PM: Transform + embed
  12:30 AM: Load into index
  
Updates to source during ETL:
  11:15 PM: New product added to source
  11:45 PM: Existing product's price changes
  
ETL doesn't capture these because extraction already happened
→ New product missing from index
→ Price change not reflected

7. Cache Staleness

Cached embeddings/results served instead of fresh data:

# Caching without TTL
@cache  # Default: cache forever
def get_product_embedding(product_id):
    doc = fetch_document(product_id)
    return embed(doc)

# First call: 12:00 PM
embedding_1 = get_product_embedding('prod_123')  # Cached

# Product description updated: 12:30 PM (but cache not invalidated)

# Second call: 12:31 PM
embedding_2 = get_product_embedding('prod_123')  # Returns old cached version
# Same embedding, even though document changed

Answer:

Detection Method 1: Timestamp-Based Monitoring

def check_index_freshness(index):
    """Monitor how old the index is."""
    
    # Get index metadata
    index_built_time = index.get_metadata()['built_at']
    current_time = datetime.now()
    
    age_hours = (current_time - index_built_time).total_seconds() / 3600
    
    # Alert based on age
    freshness_slo = 24  # Must be reindexed within 24 hours
    
    if age_hours > freshness_slo:
        log_alert(f"Index is {age_hours:.1f} hours old (SLO: {freshness_slo}h)")
        return False  # Stale
    
    return True  # Fresh

def monitor_document_lag(index, source_db):
    """Check if indexed documents match source."""
    
    # Sample: compare metadata in index vs source
    sample_doc_ids = random.sample(list(index.all_doc_ids()), 100)
    
    mismatches = 0
    
    for doc_id in sample_doc_ids:
        indexed_doc = index.get(doc_id)
        source_doc = source_db.get(doc_id)
        
        if indexed_doc is None and source_doc is not None:
            # Document in source but not in index (missing)
            mismatches += 1
        elif indexed_doc is not None and source_doc is None:
            # Document in index but not in source (stale)
            mismatches += 1
        elif indexed_doc['updated_at'] < source_doc['updated_at']:
            # Index version is older
            mismatches += 1
    
    staleness_ratio = mismatches / len(sample_doc_ids)
    
    if staleness_ratio > 0.05:  # Alert if > 5% stale
        log_alert(f"Staleness ratio: {staleness_ratio:.1%}")
    
    return staleness_ratio

Detection Method 2: A/B Testing Against Real-Time Source

def compare_index_vs_real_time(query):
    """Query both index and real-time source, compare results."""
    
    # Index-based retrieval
    indexed_results = index.search(query, k=5)
    indexed_docs = [index.fetch(rid) for rid in indexed_results]
    
    # Real-time query
    real_time_results = source_db.query(query)
    
    # Compare
    mismatch_count = 0
    
    for doc in indexed_docs:
        real_time_doc = source_db.get(doc['id'])
        
        if real_time_doc is None:
            # Indexed doc doesn't exist in source
            mismatch_count += 1
        elif doc['updated_at'] < real_time_doc['updated_at']:
            # Indexed doc is older
            mismatch_count += 1
    
    freshness_score = 1 - (mismatch_count / len(indexed_docs))
    
    return {
        'query': query,
        'freshness_score': freshness_score,
        'stale_count': mismatch_count,
    }

# Run continuously
def continuous_freshness_check(sample_rate=0.1):
    """Spot-check 10% of queries against real-time source."""
    
    for query in stream_queries():
        if random.random() < sample_rate:
            freshness = compare_index_vs_real_time(query)
            log_metric('index_freshness_score', freshness['freshness_score'])
            
            if freshness['freshness_score'] < 0.95:
                log_alert(f"Low freshness for query: {query}")

Detection Method 3: Source-of-Truth Comparison

def detect_contradictions(retrieved_docs, user_query):
    """Check if indexed answers contradict known facts."""
    
    # Extract claim from retrieved document
    claim = extract_claim(retrieved_docs[0], user_query)
    
    # Verify against external source of truth
    truth_source = get_truth_source(user_query)  # e.g., API, curated KB
    true_claim = truth_source.lookup(claim)
    
    if claim != true_claim:
        # Mismatch: indexed data contradicts source of truth
        log_staleness_alert(
            f"Contradiction: Index says '{claim}', Truth source says '{true_claim}'"
        )
        return False
    
    return True

Detection Method 4: Change Feed Monitoring

class UpdateLagMonitor:
    def __init__(self, source_change_feed):
        self.change_feed = source_change_feed
        self.last_processed_change_id = None
    
    def check_update_lag(self):
        """Monitor lag between source changes and index updates."""
        
        # Get latest change in source
        latest_change = self.change_feed.get_latest()
        
        # Check if index has processed this change
        if self.index.has_processed(latest_change['id']):
            lag = 0
        else:
            # Get timestamp when change should have been processed
            expected_process_time = latest_change['timestamp'] + timedelta(seconds=30)
            lag = (datetime.now() - expected_process_time).total_seconds()
        
        if lag > 300:  # 5 minutes
            log_alert(f"Index update lag: {lag:.0f}s")
        
        return lag

Production SLOs for Staleness

staleness_slos = {
    'max_index_age_hours': 24,           # Reindex at least daily
    'max_document_lag_seconds': 300,     # Updates reflected within 5 min
    'max_metadata_lag_seconds': 60,      # Prices/availability updated within 1 min
    'max_deletions_lag_seconds': 600,    # Deleted items removed within 10 min
    'acceptable_staleness_ratio': 0.02,  # <2% of docs out of sync
    'freshness_check_frequency': '1h',   # Verify freshness hourly
}

Answer:

Strategy 1: Incremental Upsert (Real-Time Updates)

Update individual documents in-place:

class IncrementalIndexUpdater:
    def __init__(self, index):
        self.index = index
        self.embedding_model = SentenceTransformer('all-mpnet-base-v2')
    
    def upsert_document(self, doc_id, doc_text, metadata=None):
        """Add or update single document in index."""
        
        # Compute embedding for new/updated document
        embedding = self.embedding_model.encode(doc_text)
        
        # Upsert to index (replace if exists, add if not)
        self.index.upsert(
            id=doc_id,
            embedding=embedding,
            metadata=metadata or {}
        )
        
        # Instant reflection in retrieval
        return True

# Usage
updater = IncrementalIndexUpdater(vector_db)

# Product price changed
updater.upsert_document(
    doc_id='product_123',
    doc_text='iPhone 15 Pro - Now $799 (was $999)',
    metadata={'price': 799, 'updated_at': datetime.now()}
)

# Retrieval immediately sees updated document
results = retriever.search('iPhone price')
# Returns: iPhone 15 Pro - Now $799

Advantages:

• Real-time reflection (seconds)
• Minimal latency impact
• Works well for frequent small updates

Disadvantages:

• Network overhead if many updates
• Embedding computation for every change

Strategy 2: Batch Updates with Change Data Capture (CDC)

Collect changes, apply batch updates periodically:

class CDCBasedIndexUpdater:
    def __init__(self, source_db, index, batch_interval_seconds=300):
        self.source_db = source_db
        self.index = index
        self.batch_interval = batch_interval_seconds
        self.pending_changes = []
    
    def start_change_capture(self):
        """Listen to source database change stream."""
        
        for change in self.source_db.listen_changes():
            # Capture change
            self.pending_changes.append(change)
            
            # Every 5 minutes, apply batch update
            if time.time() % self.batch_interval < 1:
                self.apply_batch_update()
    
    def apply_batch_update(self):
        """Batch process all pending changes."""
        
        if not self.pending_changes:
            return
        
        print(f"Applying {len(self.pending_changes)} changes...")
        
        # Deduplicate (keep latest version of each doc)
        latest_changes = {}
        for change in self.pending_changes:
            latest_changes[change['doc_id']] = change
        
        # Embed all changed documents
        doc_ids = list(latest_changes.keys())
        documents = [self.source_db.get(did) for did in doc_ids]
        embeddings = self.embedding_model.encode([d['text'] for d in documents])
        
        # Batch upsert to index
        self.index.upsert_batch([
            {
                'id': did,
                'embedding': emb,
                'metadata': latest_changes[did]['metadata']
            }
            for did, emb in zip(doc_ids, embeddings)
        ])
        
        # Clear pending
        self.pending_changes = []
        print(f"Index updated. New total: {self.index.count()}")

# Usage
updater = CDCBasedIndexUpdater(postgres_db, pinecone_index, batch_interval_seconds=300)
updater.start_change_capture()  # Runs in background, batches every 5 min

Advantages:

• Efficient batch processing (embed multiple docs at once)
• Deduplication (only embed latest version)
• Lower embedding cost

Disadvantages:

• Updates are delayed (every 5 min)
• More complex to implement

Strategy 3: TTL-Based Invalidation

Mark documents as stale, re-embed on cache miss:

class TTLBasedIndexUpdater:
    def __init__(self, index, ttl_seconds=86400):
        self.index = index
        self.ttl = ttl_seconds
    
    def get_with_ttl(self, doc_id):
        """Retrieve doc, invalidate if past TTL."""
        
        doc = self.index.get(doc_id)
        
        if doc is None:
            return None
        
        # Check if expired
        updated_at = doc['metadata'].get('updated_at')
        age = (datetime.now() - updated_at).total_seconds()
        
        if age > self.ttl:
            # Stale, mark for refresh
            self.index.mark_stale(doc_id)
            return None  # Force re-fetch from source
        
        return doc
    
    def refresh_stale_documents(self):
        """Background task: refresh documents past TTL."""
        
        stale_docs = self.index.get_stale_docs()
        
        for doc_id in stale_docs:
            # Fetch fresh version from source
            fresh_doc = self.source_db.get(doc_id)
            
            if fresh_doc:
                # Re-embed
                new_embedding = self.embedding_model.encode(fresh_doc['text'])
                self.index.update(
                    doc_id,
                    embedding=new_embedding,
                    metadata={'updated_at': datetime.now()}
                )
            else:
                # Deleted in source, remove from index
                self.index.delete(doc_id)

# Usage
updater = TTLBasedIndexUpdater(index, ttl_seconds=86400)  # 24 hour TTL

# Background refresh task
scheduler.add_job(
    updater.refresh_stale_documents,
    'interval',
    hours=6  # Check every 6 hours
)

Advantages:

• Simple to implement
• Automatic staleness detection
• Works for deletions

Disadvantages:

• Stale data served until TTL expires
• Extra latency if refresh on query

Strategy 4: Hybrid: Real-Time + Batch

Real-time updates for critical fields, batch for others:

class HybridIndexUpdater:
    def __init__(self, index, embedding_model):
        self.index = index
        self.embedding_model = embedding_model
    
    def update_document(self, doc_id, changes):
        """
        Update document with smart staleness handling.
        
        Critical fields (price, availability): Real-time upsert
        Other fields (description): Batch update
        """
        
        # Separate changes by criticality
        critical_fields = {'price', 'in_stock', 'availability'}
        critical_changes = {k: v for k, v in changes.items() if k in critical_fields}
        other_changes = {k: v for k, v in changes.items() if k not in critical_fields}
        
        # Real-time update for critical fields
        if critical_changes:
            doc = self.index.get(doc_id)
            doc['metadata'].update(critical_changes)
            self.index.upsert(doc_id, doc['embedding'], doc['metadata'])
        
        # Queue other fields for batch embedding
        if other_changes:
            self.batch_queue.append({
                'doc_id': doc_id,
                'changes': other_changes
            })
        
        return True

Comparison of Strategies

Answer:

Deletions are a special case of staleness: documents removed from source must also be removed from index.

Challenge: Soft Deletes vs. Hard Deletes

Hard delete: Document permanently removed from database
  Problem: Can't easily detect what was deleted (no historical record)
  
Soft delete: Document marked as deleted but not removed
  {
    'id': 'doc_123',
    'text': '...',
    'deleted': True,          # Soft delete marker
    'deleted_at': '2024-01-15'
  }
  
  Advantage: Easy to detect and handle in index

Strategy 1: Soft Delete Filtering

class SoftDeleteAwareRetriever:
    def __init__(self, index):
        self.index = index
    
    def search(self, query, k=5):
        """Search index, filter out soft-deleted documents."""
        
        # Retrieve more than k to account for deletions
        candidates = self.index.search(query, k=k*2)
        
        # Filter out soft-deleted
        active_results = [
            doc for doc in candidates
            if not doc['metadata'].get('deleted', False)
        ]
        
        return active_results[:k]

Limitation: Still returns soft-deleted docs initially, then filters.

Strategy 2: Proactive Deletion from Index

class ProactiveDeletionHandler:
    def __init__(self, index, source_db):
        self.index = index
        self.source_db = source_db
    
    def process_deletion(self, doc_id):
        """When document deleted from source, remove from index."""
        
        # Verify deletion
        if self.source_db.exists(doc_id):
            # Not actually deleted in source
            return False
        
        # Remove from index completely
        self.index.delete(doc_id)
        
        log_metric('documents_deleted', 1)
        return True
    
    def sync_deletions(self):
        """Background task: sync deletions from source."""
        
        # Get IDs of all documents in index
        indexed_ids = set(self.index.all_ids())
        
        # Get IDs of all documents still in source
        source_ids = set(self.source_db.all_ids())
        
        # Documents in index but not in source = deleted
        deleted_ids = indexed_ids - source_ids
        
        if deleted_ids:
            print(f"Syncing {len(deleted_ids)} deletions...")
            for did in deleted_ids:
                self.index.delete(did)
        
        return len(deleted_ids)

Strategy 3: Versioned Indices

Keep multiple index versions, switch on deployment:

class VersionedIndexManager:
    def __init__(self):
        self.current_index = None
        self.index_versions = {}  # {'v1': index, 'v2': index}
    
    def create_new_version(self, version_name):
        """Build a fresh index version."""
        
        # Create empty index
        new_index = VectorDB(name=f'index_{version_name}')
        
        # Populate with current source data
        for doc in self.source_db.all_documents():
            if not doc.get('deleted', False):  # Skip soft-deleted
                embedding = self.embedding_model.encode(doc['text'])
                new_index.add(doc['id'], embedding, doc['metadata'])
        
        self.index_versions[version_name] = new_index
        return new_index
    
    def activate_version(self, version_name):
        """Switch to a new index version."""
        
        if version_name not in self.index_versions:
            raise ValueError(f"Version {version_name} not found")
        
        old_version = self.current_index
        self.current_index = self.index_versions[version_name]
        
        print(f"Switched from {old_version} to {version_name}")
        
        # Can keep old version for rollback
        return True
    
    def rollback(self, previous_version):
        """Rollback to previous version if new one is bad."""
        
        self.current_index = self.index_versions[previous_version]
        print(f"Rolled back to {previous_version}")

# Usage
manager = VersionedIndexManager()

# Build v2 with latest data (including handling deletions)
manager.create_new_version('v2')

# Test v2
test_results = run_tests_on_index(manager.index_versions['v2'])

if test_results['quality'] > manager.current_quality:
    # Promote v2 to production
    manager.activate_version('v2')
else:
    # Keep v1, discard v2
    pass

Advantages:

• Clean cut-over (no gradual inconsistency)
• Easy rollback
• Batch handles deletions naturally

Disadvantages:

• Requires 2x storage temporarily
• Can't update individual docs

Strategy 4: Deletion Webhooks

Source database sends webhooks when documents are deleted:

from flask import Flask, request

app = Flask(__name__)
index_manager = None

@app.route('/webhooks/document-deleted', methods=['POST'])
def on_document_deleted():
    """Webhook endpoint: called when document deleted in source."""
    
    payload = request.json
    doc_id = payload['doc_id']
    deleted_at = payload['deleted_at']
    
    # Remove immediately from index
    index_manager.delete(doc_id)
    
    log_metric('deletion_webhook_processed', 1)
    
    return {'status': 'ok'}, 200

# Source DB configuration
# When document deleted:
#   DELETE FROM documents WHERE id = $1
#   POST /webhooks/document-deleted {doc_id: $1, deleted_at: now()}

Advantages:

• Real-time deletion reflection
• Minimal latency

Disadvantages:

• Requires webhook infrastructure
• Single point of failure if webhook endpoint down

Best Practices

deletion_policy = {
    'strategy': 'soft_delete_with_sync',  # Use soft deletes + periodic sync
    
    'soft_delete_field': 'is_deleted',    # Marker field name
    
    'sync_frequency': '6h',                # Sync deletions every 6 hours
    
    'filter_in_retrieval': True,           # Always filter deleted in queries
    
    'hard_delete_after': '90d',            # Permanently remove old deletes after 90d
    
    'audit_trail': True,                   # Keep deletion log for compliance
}

Answer:

End-to-End Update Pipeline

Source of Truth (Database/API)
        ↓ (CDC / Webhook)
    Event Stream (Kafka, Redis)
        ↓ (Consume & Batch)
    Update Worker Pool
        ↓ (Embed Documents)
    Embedding Cache
        ↓ (Batch Upload)
    Vector Index
        ↓ (Served via API)
    RAG Application

Implementation Example

from kafka import KafkaConsumer
import json
from datetime import datetime, timedelta

class ProductionUpdatePipeline:
    def __init__(self, config):
        self.config = config
        self.kafka_consumer = KafkaConsumer(
            'document-changes',
            bootstrap_servers=['kafka:9092'],
            group_id='index-updater'
        )
        self.embedding_model = SentenceTransformer(config['embedding_model'])
        self.vector_db = PineconeIndex(config['pinecone_index'])
        self.source_db = PostgresConnection(config['postgres_url'])
        
        # Batching config
        self.batch_size = config.get('batch_size', 100)
        self.batch_timeout_seconds = config.get('batch_timeout', 30)
        self.pending_updates = []
        self.last_batch_time = datetime.now()
    
    def run(self):
        """Main update pipeline loop."""
        
        print("Starting production update pipeline...")
        
        for message in self.kafka_consumer:
            try:
                # 1. Parse change event
                change = json.loads(message.value)
                change_type = change['type']  # 'insert', 'update', 'delete'
                doc_id = change['doc_id']
                
                # 2. Route based on change type
                if change_type == 'delete':
                    # Deletions: immediate (hard delete from index)
                    self.vector_db.delete(doc_id)
                    log_metric('deletions_processed', 1)
                
                else:  # 'insert' or 'update'
                    # Queue for batch processing
                    self.pending_updates.append(change)
                
                # 3. Check if time to batch
                should_batch = (
                    len(self.pending_updates) >= self.batch_size or
                    (datetime.now() - self.last_batch_time).total_seconds() > self.batch_timeout_seconds
                )
                
                if should_batch:
                    self.process_batch()
            
            except Exception as e:
                log_error(f"Error processing change: {e}")
                # Continue processing despite errors
    
    def process_batch(self):
        """Batch embed and upsert documents."""
        
        if not self.pending_updates:
            return
        
        start_time = datetime.now()
        batch = self.pending_updates[:self.batch_size]
        
        print(f"Processing batch of {len(batch)} updates...")
        
        try:
            # 1. Fetch fresh documents from source
            doc_ids = [change['doc_id'] for change in batch]
            documents = []
            
            for did in doc_ids:
                doc = self.source_db.get(did)
                if doc and not doc.get('deleted', False):
                    documents.append(doc)
            
            # 2. Compute embeddings
            texts = [doc['text'] for doc in documents]
            embeddings = self.embedding_model.encode(texts, batch_size=32)
            
            # 3. Prepare upsert payload
            upsert_items = []
            for doc, embedding in zip(documents, embeddings):
                upsert_items.append({
                    'id': doc['id'],
                    'values': embedding,
                    'metadata': {
                        'title': doc.get('title'),
                        'updated_at': datetime.now().isoformat(),
                        'version': doc.get('version')
                    }
                })
            
            # 4. Batch upsert to index
            self.vector_db.upsert(upsert_items)
            
            # 5. Metrics
            elapsed = (datetime.now() - start_time).total_seconds()
            log_metric('batch_processed_count', len(batch))
            log_metric('batch_embedding_latency_s', elapsed)
            
            print(f"✓ Processed {len(batch)} docs in {elapsed:.1f}s")
            
            # 6. Clear processed updates
            self.pending_updates = self.pending_updates[self.batch_size:]
            self.last_batch_time = datetime.now()
        
        except Exception as e:
            log_error(f"Batch processing failed: {e}")
            # Retry logic would go here

Configuration for Different Update Frequencies

UPDATE_PIPELINES = {
    'critical_metadata': {
        'description': 'Real-time: prices, availability, stock',
        'strategy': 'incremental_upsert',
        'latency_target': '10s',
        'embedding_computation': False,  # Metadata-only updates
        'cost': 'Low'
    },
    'moderate_freshness': {
        'description': 'Batch: product descriptions, reviews',
        'strategy': 'cdc_batch',
        'batch_size': 100,
        'batch_interval': '5m',
        'latency_target': '5m',
        'embedding_computation': True,
        'cost': 'Medium'
    },
    'low_priority': {
        'description': 'Nightly: historical data, archives',
        'strategy': 'full_reindex',
        'frequency': 'daily',
        'latency_target': 'Hours',
        'embedding_computation': True,
        'cost': 'High-upfront'
    }
}

Monitoring and Alerting

class UpdatePipelineMonitor:
    def __init__(self):
        self.slos = {
            'max_batch_lag_seconds': 300,      # Batches processed within 5 min
            'max_document_lag_seconds': 600,   # Documents updated within 10 min
            'max_deletion_lag_seconds': 60,    # Deletions processed within 1 min
            'successful_batch_rate': 0.99,     # 99% of batches succeed
            'embedding_accuracy': 0.98,        # 98% of embeddings computed correctly
        }
    
    def check_slo(self):
        """Monitor pipeline health."""
        
        metrics = {
            'avg_batch_processing_time_s': get_metric('batch_latency').mean(),
            'pending_updates_count': get_metric('queue_depth'),
            'successful_batches': get_metric('successful_batches'),
            'failed_batches': get_metric('failed_batches'),
        }
        
        # Alert if exceeds SLO
        if metrics['avg_batch_processing_time_s'] > self.slos['max_batch_lag_seconds']:
            alert(f"Batch processing lag exceeded: {metrics['avg_batch_processing_time_s']:.0f}s")
        
        if metrics['pending_updates_count'] > 10000:
            alert(f"Update queue backlog: {metrics['pending_updates_count']} pending")
        
        success_rate = (metrics['successful_batches'] / 
                       (metrics['successful_batches'] + metrics['failed_batches']))
        
        if success_rate < self.slos['successful_batch_rate']:
            alert(f"Batch success rate: {success_rate:.1%} (SLO: {self.slos['successful_batch_rate']:.1%})")

Cost Optimization

def optimize_pipeline_cost(doc_size_chars, monthly_updates, embedding_cost_per_1m=0.02):
    """Estimate and optimize pipeline cost."""
    
    # Option 1: Real-time incremental updates
    cost_realtime = monthly_updates * embedding_cost_per_1m / 1_000_000
    
    # Option 2: Batch every 5 minutes
    batches_per_month = (monthly_updates / 100) * (30 * 24 * 12)  # Assuming 100/batch
    cost_batch_5m = batches_per_month * embedding_cost_per_1m / 1_000_000
    
    # Option 3: Batch daily reindex
    cost_daily_reindex = cost_per_daily_reindex(monthly_updates)
    
    print(f"Real-time: ${cost_realtime:.2f}/month")
    print(f"Batch (5m): ${cost_batch_5m:.2f}/month")
    print(f"Daily reindex: ${cost_daily_reindex:.2f}/month")
    
    # Recommendation: Use option with lowest cost that meets freshness SLO

Answer:

Evaluation Framework

Metric 1: Freshness Score

def compute_freshness_score(retrieved_doc, source_doc):
    """
    Score how fresh retrieved doc is compared to source.
    0 = completely stale, 1 = perfectly fresh.
    """
    
    if source_doc is None:
        # Document deleted in source, but returned from index
        return 0.0
    
    # Time-based freshness
    index_age = (datetime.now() - retrieved_doc['updated_at']).total_seconds()
    source_age = (datetime.now() - source_doc['updated_at']).total_seconds()
    
    # If source is newer, index is stale
    if index_age > source_age:
        stale_seconds = index_age - source_age
        decay_factor = math.exp(-stale_seconds / 3600)  # Half-life: 1 hour
        return decay_factor
    
    return 1.0  # Fully fresh

def compute_corpus_freshness(retrieved_docs, source_db):
    """Average freshness across corpus."""
    
    scores = []
    
    for doc in retrieved_docs:
        source_doc = source_db.get(doc['id'])
        score = compute_freshness_score(doc, source_doc)
        scores.append(score)
    
    return np.mean(scores)

# Example
freshness = compute_corpus_freshness(retriever.retrieve_all(), postgres_db)
print(f"Corpus freshness: {freshness:.2%}")

if freshness < 0.95:
    alert(f"Corpus freshness below SLO (95%): {freshness:.2%}")

Metric 2: Staleness Distribution

def analyze_staleness_distribution(index, source_db):
    """What fraction of index is how old?"""
    
    ages_hours = []
    
    for doc_id in index.all_ids():
        indexed_doc = index.get(doc_id)
        source_doc = source_db.get(doc_id)
        
        if source_doc is None:
            # Deleted
            ages_hours.append(999)  # Very stale
        else:
            age = (datetime.now() - indexed_doc['updated_at']).total_seconds() / 3600
            ages_hours.append(age)
    
    # Percentiles
    p50 = np.percentile(ages_hours, 50)
    p95 = np.percentile(ages_hours, 95)
    p99 = np.percentile(ages_hours, 99)
    
    print(f"Staleness distribution:")
    print(f"  P50: {p50:.1f} hours")
    print(f"  P95: {p95:.1f} hours")
    print(f"  P99: {p99:.1f} hours")
    
    # Alert if P95 > SLO
    if p95 > 24:  # SLO: 24 hours
        alert(f"P95 staleness ({p95:.1f}h) exceeds SLO (24h)")
    
    return {'p50': p50, 'p95': p95, 'p99': p99}

Metric 3: Deletion Lag

def measure_deletion_lag(index, source_db):
    """How long does it take for deleted items to leave index?"""
    
    # Find items deleted in source but still in index
    stale_deletes = []
    
    for doc_id in index.all_ids():
        if not source_db.exists(doc_id):
            # In index but not in source = stale delete
            indexed_doc = index.get(doc_id)
            
            # How long since it was deleted?
            # (approximated by deletion marker timestamp if available)
            deleted_at = source_db.get_deletion_timestamp(doc_id) or datetime.now()
            lag = (datetime.now() - deleted_at).total_seconds()
            
            stale_deletes.append(lag)
    
    if stale_deletes:
        avg_lag = np.mean(stale_deletes)
        max_lag = np.max(stale_deletes)
        
        print(f"Deletion lag: avg={avg_lag:.0f}s, max={max_lag:.0f}s")
        
        if max_lag > 600:  # 10 minutes
            alert(f"Some deletions not reflected for {max_lag/60:.1f} minutes")
    
    return stale_deletes

Continuous Monitoring Dashboard

class StalenessDashboard:
    def __init__(self):
        self.metrics = {}
    
    def update_metrics(self, index, source_db):
        """Update all staleness metrics."""
        
        self.metrics['freshness_score'] = compute_corpus_freshness(
            index.sample_documents(100),
            source_db
        )
        
        self.metrics['staleness_dist'] = analyze_staleness_distribution(
            index, source_db
        )
        
        self.metrics['deletion_lag'] = measure_deletion_lag(index, source_db)
        
        self.metrics['index_age_hours'] = (
            datetime.now() - index.get_metadata()['built_at']
        ).total_seconds() / 3600
        
        self.metrics['reindex_overdue'] = (
            self.metrics['index_age_hours'] > 24  # SLO: daily reindex
        )
    
    def render_summary(self):
        """Human-readable summary."""
        
        print("=" * 50)
        print("STALENESS DASHBOARD")
        print("=" * 50)
        print(f"Freshness Score: {self.metrics['freshness_score']:.1%}")
        print(f"P95 Staleness: {self.metrics['staleness_dist']['p95']:.1f}h")
        print(f"Index Age: {self.metrics['index_age_hours']:.1f}h")
        print(f"Reindex Overdue: {self.metrics['reindex_overdue']}")
        print("=" * 50)

# Usage
dashboard = StalenessDashboard()

scheduler.add_job(
    lambda: dashboard.update_metrics(index, source_db),
    'interval',
    minutes=30  # Update every 30 min
)

Answer:

Staleness doesn't exist in isolation; it interacts with and amplifies other failure modes:

Interaction 1: Staleness + Hallucination = High Confidence in Old Data

Scenario: Stock price
  
Real-world: Apple stock = $210
Index (stale): Apple stock = $180 (from 2 weeks ago)

User query: "What is Apple stock price?"

Retrieval returns: "Apple stock is $180"
LLM sees: [Document says "$180"]
LLM responds: "Apple stock is $180"  (with high confidence, since grounded by retrieval)

User believes: $180 (confidently wrong)

Impact: Higher trust in false information due to grounding.

Mitigation:

def detect_stale_data_hallucination(query, retrieved_doc, llm_answer):
    """Detect when hallucination is enabled by stale data."""
    
    # Is doc stale?
    doc_age_hours = (datetime.now() - retrieved_doc['updated_at']).total_seconds() / 3600
    is_stale = doc_age_hours > 24
    
    # Does query imply recency sensitivity?
    recency_keywords = ['current', 'now', 'latest', 'today', 'recent']
    is_recency_sensitive = any(kw in query.lower() for kw in recency_keywords)
    
    if is_stale and is_recency_sensitive:
        # High risk: stale data + recency-sensitive query
        # LLM may confidently answer with outdated info
        flag_staleness_hallucination_risk(query, retrieved_doc)
        
        # Return fallback
        return "I cannot confidently answer this question due to potential data staleness. Please verify with current source."
    
    return llm_answer

Interaction 2: Staleness + Retrieval Failure = Cascading Failures

Scenario: E-commerce search

Real-world state:
  - Product A: In stock, $50
  - Product B: Out of stock (deleted 1 week ago)
  - Product C: In stock, $75

Stale index (not yet updated):
  - Product A: (missing - update not yet synced)
  - Product B: (still present - deletion not synced)
  - Product C: In stock, $75 (outdated price)

User query: "In-stock products under $70"

Retrieval fails to find Product A (retrieval failure)
  Because it hasn't been indexed yet.

Retrieval returns Product B (stale data failure)
  Which is actually out of stock.

Combined effect: User sees incomplete + incorrect results
  Missing: Product A (should be top result)
  Wrong: Product B (should be excluded)

Interaction 3: Staleness + Embedding Mismatch = Cascading Misses

Scenario: Medical terminology

Real situation (current):
  - New drug interaction discovered
  - Medical docs updated with new terminology

Stale index:
  - Old terminology embedded
  - New documents not yet added

Embedding mismatch + staleness:
  Query (new terminology): "What are interactions with new_drug_X?"
  Embedding model doesn't recognize new terminology well (mismatch)
  Index doesn't contain relevant new docs (staleness)
  
  Result: Both retrieval dimensions fail
  Compounding effect: User gets 0 results or wrong results

Amplification Effect Diagram

                    Staleness
                        ↓
        ┌───────────────┼───────────────┐
        ↓               ↓               ↓
    Hallucination  Retrieval       Embedding
     (confident   Failure         Mismatch
      false)     (missing)        (semantic gap)
        ↓               ↓               ↓
        └───────────────┼───────────────┘
                        ↓
              COMPOUND FAILURE
              (worse together than sum of parts)

Detection of Compounded Failures

def detect_compounded_failure(query, retrieved_docs, llm_answer, source_db):
    """Identify when multiple failure modes reinforce each other."""
    
    failure_modes = {
        'staleness': False,
        'hallucination': False,
        'retrieval_failure': False,
        'embedding_mismatch': False,
    }
    
    # Check for staleness
    avg_doc_age = np.mean([(datetime.now() - d['updated_at']).total_seconds() / 3600 
                           for d in retrieved_docs])
    if avg_doc_age > 24:
        failure_modes['staleness'] = True
    
    # Check for hallucination (answer not in docs)
    context = " ".join([d['text'] for d in retrieved_docs])
    faithfulness = nli_model.predict([[context, llm_answer]])[0][2]
    if faithfulness < 0.6:
        failure_modes['hallucination'] = True
    
    # Check for retrieval failure (no relevant docs)
    if len(retrieved_docs) == 0:
        failure_modes['retrieval_failure'] = True
    
    # Check for embedding mismatch
    if retrieved_docs and all(d['similarity'] < 0.6 for d in retrieved_docs):
        failure_modes['embedding_mismatch'] = True
    
    # Count failures
    failure_count = sum(failure_modes.values())
    
    if failure_count >= 2:
        # Multiple failure modes detected
        mode_names = [k for k, v in failure_modes.items() if v]
        alert(f"COMPOUND FAILURE: {' + '.join(mode_names)}")
        
        return {
            'compound_failure': True,
            'modes': mode_names,
            'severity': 'critical' if failure_count >= 3 else 'high'
        }
    
    return {'compound_failure': False}

Answer:

Index update strategy involves balancing three competing goals: cost (infrastructure + compute), latency (how long updates take), and freshness (how recent data is).

Cost-Freshness-Latency Matrix

def analyze_update_strategy_tradeoffs(corpus_size_docs=100000, monthly_updates=500000):
    """Compare strategies across cost, freshness, latency."""
    
    strategies = {
        'real_time_incremental': {
            'description': 'Upsert immediately on change',
            'freshness_lag_seconds': 5,
            'cost_per_query': 0.0020,  # High: embed every change
            'latency_p99': 50,  # ms
            'monthly_embedding_calls': monthly_updates,  # Worst case
            'requires_infrastructure': ['Message queue', 'Worker pool', 'Vector DB'],
        },
        'batch_5_minutes': {
            'description': 'Batch and process every 5 min',
            'freshness_lag_seconds': 300,
            'cost_per_query': 0.0005,  # Low: batch embed
            'latency_p99': 200,  # ms (batch processing)
            'monthly_embedding_calls': (monthly_updates / 100) * (30 * 24 * 12),
            'requires_infrastructure': ['Kafka', 'Worker pool', 'Vector DB'],
        },
        'batch_1_hour': {
            'description': 'Batch and process hourly',
            'freshness_lag_seconds': 3600,
            'cost_per_query': 0.0002,  # Very low
            'latency_p99': 500,  # ms
            'monthly_embedding_calls': (monthly_updates / 1000) * (30 * 24),
            'requires_infrastructure': ['Kafka', 'Single worker', 'Vector DB'],
        },
        'daily_reindex': {
            'description': 'Full reindex once per day',
            'freshness_lag_seconds': 86400,
            'cost_per_query': 0.0001,  # Minimal per-query cost
            'latency_p99': 0,  # Batch job, not per-query
            'monthly_embedding_calls': corpus_size_docs,
            'requires_infrastructure': ['Scheduled job', 'Temp storage'],
        },
        'on_demand_no_updates': {
            'description': 'Never update, only compute on query',
            'freshness_lag_seconds': 999999,  # Forever stale (initiallyup-to-date only)
            'cost_per_query': 0.0010,  # Very high: embed on every query
            'latency_p99': 200,  # ms (embedding latency)
            'monthly_embedding_calls': monthly_updates * 100,  # Assume each doc queried 100x
            'requires_infrastructure': ['GPU for embedding'],
        },
    }
    
    for name, metrics in strategies.items():
        embedding_cost = metrics['monthly_embedding_calls'] * 0.00001  # Cost per embedding
        total_monthly = embedding_cost + 1000  # Add infrastructure
        
        print(f"\n{name.upper()}")
        print(f"  Freshness lag: {metrics['freshness_lag_seconds']:.0f}s ({metrics['freshness_lag_seconds']/3600:.1f}h)")
        print(f"  Query latency P99: {metrics['latency_p99']:.0f}ms")
        print(f"  Monthly cost: ${total_monthly:.2f}")
        print(f"  Infrastructure: {', '.join(metrics['requires_infrastructure'])}")

analyze_update_strategy_tradeoffs()

# Output:
# REAL_TIME_INCREMENTAL
#   Freshness lag: 5s
#   Query latency P99: 50ms
#   Monthly cost: $5000+
#   Infrastructure: Message queue, Worker pool, Vector DB
#
# BATCH_5_MINUTES
#   Freshness lag: 300s (5 min)
#   Query latency P99: 200ms
#   Monthly cost: $1200
#   Infrastructure: Kafka, Worker pool, Vector DB
#
# DAILY_REINDEX
#   Freshness lag: 86400s (24 h)
#   Query latency P99: 0ms
#   Monthly cost: $100
#   Infrastructure: Scheduled job, Temp storage

Decision Matrix by Use Case

Cost Breakdown: Real-Time Incremental vs. Daily Reindex

def compare_cost_detailed(corpus_size=100000, monthly_changes=500000):
    """Detailed cost comparison."""
    
    print("REAL-TIME INCREMENTAL")
    print("-" * 40)
    
    # Every change triggers an embedding
    embeddings_per_month = monthly_changes
    embedding_cost = embeddings_per_month * 0.00001
    print(f"Embeddings: {embeddings_per_month:,} × $0.00001 = ${embedding_cost:.2f}")
    
    # Infrastructure: workers, queues, monitoring
    infrastructure_cost = 3000
    print(f"Infrastructure: ${infrastructure_cost:.2f}")
    
    realtime_total = embedding_cost + infrastructure_cost
    print(f"TOTAL: ${realtime_total:.2f}/month")
    
    print("\n" + "=" * 40)
    print("DAILY REINDEX")
    print("-" * 40)
    
    # Once per day: embed all docs
    embeddings_per_month = corpus_size
    embedding_cost = embeddings_per_month * 0.00001
    print(f"Embeddings: {embeddings_per_month:,} × $0.00001 = ${embedding_cost:.2f}")
    
    # Infrastructure: scheduled job only
    infrastructure_cost = 100
    print(f"Infrastructure: ${infrastructure_cost:.2f}")
    
    daily_total = embedding_cost + infrastructure_cost
    print(f"TOTAL: ${daily_total:.2f}/month")
    
    print("\n" + "=" * 40)
    print(f"SAVINGS: ${realtime_total - daily_total:.2f}/month ({(1 - daily_total/realtime_total)*100:.0f}%)")
    print(f"COST OF FRESHNESS: ${realtime_total - daily_total:.2f}/month for ~1 hour freshness improvement")

compare_cost_detailed(corpus_size=100000, monthly_changes=500000)

# Output:
# REAL-TIME INCREMENTAL
# Embeddings: 500,000 × $0.00001 = $5.00
# Infrastructure: $3000.00
# TOTAL: $3005.00/month
#
# DAILY REINDEX
# Embeddings: 100,000 × $0.00001 = $1.00
# Infrastructure: $100.00
# TOTAL: $101.00/month
#
# SAVINGS: $2904.00/month (97% cost reduction)

ROI of Freshness

def estimate_freshness_roi(use_case, freshness_improvement_hours, business_impact):
    """Estimate business value of improved freshness."""
    
    # Rough estimates of business impact per hour of staleness
    impact_per_hour = {
        'ecommerce_price': 100,    # $100 value per hour of incorrect pricing
        'financial_trading': 10000,  # $10k per hour of stale data
        'medical': 5000,            # Hard to quantify, but critical
        'news': 50,                 # Low impact of staleness
    }
    
    impact = impact_per_hour.get(use_case, 100)
    daily_value = impact * freshness_improvement_hours
    monthly_value = daily_value * 30
    
    # Strategy cost
    upgrade_cost = 2000  # Cost to upgrade from daily to hourly batch
    
    roi = monthly_value / upgrade_cost
    
    print(f"Business impact: ${monthly_value:.2f}/month")
    print(f"Upgrade cost: ${upgrade_cost:.2f}")
    print(f"ROI: {roi:.1f}x ({roi >= 1 and 'POSITIVE' or 'NEGATIVE'})")
    
    return roi > 1

# Examples
estimate_freshness_roi('ecommerce_price', freshness_improvement_hours=23, business_impact=100)
# Business impact: $69000.00/month
# Upgrade cost: $2000.00
# ROI: 34.5x (POSITIVE)  ← Worth upgrading

estimate_freshness_roi('news', freshness_improvement_hours=23, business_impact=50)
# Business impact: $34500.00/month
# Upgrade cost: $2000.00
# ROI: 17.3x (POSITIVE)

estimate_freshness_roi('blog_archive', freshness_improvement_hours=23, business_impact=1)
# Business impact: $1500.00/month
# Upgrade cost: $2000.00
# ROI: 0.8x (NEGATIVE)  ← Not worth upgrading

Optimization Checklist

Choose update strategy:
  ☐ Identify freshness SLO (what is "stale" for your domain?)
  ☐ Estimate monthly change volume
  ☐ Calculate cost of each strategy
  ☐ Calculate business value of improved freshness
  ☐ Choose strategy with best ROI
  
Implementation:
  ☐ Set up CDC or webhooks from source
  ☐ Build worker pool for batching/embedding
  ☐ Implement deletion handling
  ☐ Add monitoring for staleness
  ☐ Set up alerting for SLO breaches
  
Monitoring:
  ☐ Track freshness score continuously
  ☐ Monitor update lag (time from source change to index)
  ☐ Alert if freshness < SLO for 30+ minutes
  ☐ Report monthly on freshness metrics