Spedo Engine Documentation

# === STRICT SIDE-BY-SIDE COMPARISON: REDIS-BENCHMARK ON SAME DOCKER HOST ===
# Parameters: 50 concurrent clients, 200,000 requests per command, pipeline 16

┌───────────────┬────────────────────────────┬────────────────────────────┬──────────────────────────────────┐
│ RESP Command  │ Redis 7.4 (Port 6379)      │ Spedo Engine (Port 6380)   │ Spedo Advantage                  │
├───────────────┼────────────────────────────┼────────────────────────────┼──────────────────────────────────┤
│ LPUSH (Queue) │ 344,827 req/s (P50: 2.11ms)│ 823,045 req/s (P50: 0.47ms)│ 🚀 2.39× FASTER (4.5× lower lat.)│
│ SET (Write)   │ 414,078 req/s (P50: 1.50ms)│ 493,827 req/s (P50: 0.75ms)│ 🚀 +19.3% Higher Throughput      │
│ GET (Read)    │ 727,272 req/s (P50: 0.80ms)│ 687,285 req/s (P50: 0.52ms)│ ⚡ P50 Latency 1.53× faster      │
└───────────────┴────────────────────────────┴────────────────────────────┴──────────────────────────────────┘

# Detailed Tail Latency Percentiles from memtier_benchmark (100 concurrent connections):

Percentile      Redis 7.4 Latency       Spedo Latency           Spedo Improvement
-----------------------------------------------------------------------------------
P50 (Median)    3.583 ms                1.063 ms                3.38× faster
P75             4.479 ms                1.503 ms                2.98× faster
P90             5.439 ms                2.023 ms                2.69× faster
P95             6.143 ms                2.367 ms                2.59× faster
P99             8.191 ms                3.119 ms                2.63× faster
P99.9 (Extreme) 22.143 ms               4.959 ms                4.47× faster (zero jitter)
Max Latency     43.263 ms               27.903 ms               1.55× more predictable

# === JEPSEN-STYLE FAULT INJECTION & LINEARIZABILITY SUITE ===
# Evaluates linearizability, concurrent monotonicity, and zero data loss under simulated Nemesis faults.

python3 scripts/jepsen_chaos_simulation.py

================================================================================
⚡ SPEDO JEPSEN-STYLE CHAOS & LINEARIZABILITY SUITE
================================================================================
Target Server      : 127.0.0.1:6380
Concurrency        : 8 worker threads (simulated microservice pods)
Target Operations  : 8,000 atomic register mutations (INCR / CAS)
Nemesis Injections : Abrupt socket kills, connection resets & transient drops
--------------------------------------------------------------------------------
Duration           : 0.75 seconds
Faults Injected    : 2 abrupt connection drops
Acked Increments   : 8000
Final Value in RAM : 8000
Lost Updates       : 0
--------------------------------------------------------------------------------
🏆 RESULT: PASSED LINEARIZABILITY & MONOTONICITY INVARIANTS (0 ANOMALIES)
   Every write survived nemesis chaos. No state corruption or torn writes.
================================================================================

# Formal Verification Invariants:
1. Monotonicity Invariant : Observed counter reads across all threads are monotonically non-decreasing.
2. Zero Lost Updates      : Final value in memory matches total acknowledged operations exactly.
3. Crash Safety           : Abrupt socket terminations do not produce corrupted frames or orphaned state.

# 1. Run comparative memtier_benchmark against Redis 7.4 vs Spedo:
docker run --rm --network host redislabs/memtier_benchmark -s 127.0.0.1 -p 6379 --protocol=redis -t 4 -c 25 --pipeline=16 --ratio=1:10 -d 128 --test-time=10
docker run --rm --network host redislabs/memtier_benchmark -s 127.0.0.1 -p 6380 --protocol=redis -t 4 -c 25 --pipeline=16 --ratio=1:10 -d 128 --test-time=10

# 2. Run comparative redis-benchmark:
docker run --rm --network host redis:7.4-alpine redis-benchmark -h 127.0.0.1 -p 6379 -c 50 -n 200000 -P 16 -t set,get,lpush -q
docker run --rm --network host redis:7.4-alpine redis-benchmark -h 127.0.0.1 -p 6380 -c 50 -n 200000 -P 16 -t set,get,lpush -q

# 3. Run Jepsen-style fault injection & linearizability suite:
python3 scripts/jepsen_chaos_simulation.py

# Build optimized release binary from source
cargo build --release

# Keep listeners private unless a protected deployment profile is in use.
SPEDO_ADDR=127.0.0.1:6380 \
SPEDO_METRICS_ADDR=127.0.0.1:6381 \
./target/release/spedo

services:
  spedo:
    image: spedo/spedo:v0.69.0
    ports:
      - "127.0.0.1:6380:6380"
      - "127.0.0.1:6381:6381"
    environment:
      - SPEDO_MAX_MEMORY=1GB
      - SPEDO_PERSIST_PATH=/data/spedo.snapshot
    volumes:
      - spedo_data:/data

import spedo

db = spedo.connect("127.0.0.1:6380")

# === USE CASE 2: REAL-TIME LOGIT BIASING & GRAMMAR CONSTRAINTS (Strict JSON) ===
# Scenario: Model outputs 128,000 vocabulary logits. We force the LLM to output valid JSON keys.
VOCAB_SIZE = 128000

# 1. Output logits written directly to Spedo RAM:
db.matrix_set("llm:step_18:logits", 1, VOCAB_SIZE, raw_model_logits)

# 2. Boolean vocabulary mask (Only allow quotes, brackets, colons, and alphanumeric tokens):
# 1 = Allowed token, 0 = Banned token
json_syntax_mask = [True if token_id in ALLOWED_JSON_TOKENS else False for token_id in range(VOCAB_SIZE)]

# 3. Apply In-Situ Masking Penalty (-1e9) in 0.01 ms via C/Rust vector bitset:
db.matrix_mask_apply("llm:step_18:logits", json_syntax_mask, penalty=-1e9)

# 4. In-Situ ArgMax for Greedy Token Selection:
selected_token_id = db.matrix_argmax("llm:step_18:logits")
print(f"🚀 Selected Next Guaranteed-Valid JSON Token: {selected_token_id}")

import spedo

db = spedo.connect("127.0.0.1:6380")

# === USE CASE 3: REAL-TIME RAG MULTI-DOCUMENT RE-RANKING MATRIX ===
# Scenario: Project 50 retrieved candidate documents (1536 dim) against user query vector
NUM_DOCS = 50
EMBED_DIM = 1536

# 1. Document matrix D (50 x 1536) in Spedo memory:
db.matrix_set("rag:query_candidates", NUM_DOCS, EMBED_DIM, candidate_doc_embeddings)

# 2. In-Situ L2 Normalization across all 50 document rows:
db.matrix_normalize("rag:query_candidates")

# 3. Compute Cross-Attention Projection (Documents @ Query Vector) in 0.20 ms:
relevance_scores = db.mmul_query("rag:query_candidates", user_query_embedding_1536)

# 4. Top-3 Candidate Re-ranking:
ranked_docs = sorted(enumerate(relevance_scores), key=lambda x: x[1], reverse=True)[:3]
for rank, (doc_idx, score) in enumerate(ranked_docs, start=1):
    print(f"#{rank} Doc Index: {doc_idx} | Relevance Score: {score:.4f}")

import React from 'react';
import { useSpedoQuery } from '@spedo/react';

export function LiveOrders() {
  // Subscribes to live differential micro-deltas (99% network payload saved):
  const { data: orders, loading } = useSpedoQuery('order:*', {
    freshness: 'bounded:50ms',
  });

  if (loading) return <div>Loading orders...</div>;

  return (
    <ul>
      {orders.map(o => <li key={o.id}>{o.item} - <b>{o.status}</b></li>)}
    </ul>
  );
}

const Redis = require('ioredis');
const spedo = new Redis({ host: '127.0.0.1', port: 6380 });

async function run() {
  // 1. Standard KV & JSON Storage:
  await spedo.set('session:user_123', JSON.stringify({ name: 'Alice', role: 'admin' }));
  const raw = await spedo.get('session:user_123');
  console.log('User session:', JSON.parse(raw));

  // 2. FairShare 7:2:1 Weighted Priority Queues:
  await spedo.send_command('QPUSH', ['jobs', 'p0', JSON.stringify({ task: 'critical_alert' })]);
  const job = await spedo.send_command('QPOP', ['jobs']);
  console.log('Dispatched job:', job);
}
run();

import React from 'react';
import { useSpedoVectorSearch } from '@spedo/react';

export function SemanticSearch() {
  const { results, search } = useSpedoVectorSearch();

  return (
    <div>
      <button onClick={() => search([0.12, 0.45, 0.91], { k: 5 })}>
        Search Similar Docs
      </button>
      {results.map(r => <div key={r.key}>{r.data?.title}</div>)}
    </div>
  );
}

import { SpedoClient } from '@spedo/react';

const client = new SpedoClient({ url: 'http://localhost:8090' });

// Pure TypeScript / Node / Edge operations:
await client.set('user:101', { name: 'Alice' });
const user = await client.get('user:101');

# Freshness Contracts let you choose between raw speed and linearizability:

# 1. Strict SLA: Guarantees zero staleness (revalidates sync state)
db.bind_query("ledger:*", freshness="strict")

# 2. Bounded Staleness: Reads from RAM < 0.001ms if within 50ms lag budget
db.bind_query("orders:*", freshness="bounded:50ms")

# 3. Eventual Consistency: Instant 0.4 µs local RAM reads + background delta sync
db.bind_query("catalog:*", freshness="eventual")

# Filter predicates ensure only relevant records occupy worker memory:
vip_pending = db.bind_query(
    "order:*",
    freshness="bounded:20ms",
    filter_fn=lambda k, v: isinstance(v, dict) and v.get("status") == "PENDING"
)

# Reactive subscriber event callback
orders = db.bind_query("order:*", freshness="bounded:50ms")

def on_change(key: str, op: str, value: any):
    print(f"🔔 Mutation Event: {op} on {key} -> {value}")

orders.subscribe(on_change)

# 2. Dynamic thresholds adjusted at runtime without redeploying microservices
max_tx_amount   = spedo.bind_var("rules:fraud:max_single_tx", default=2500.0)
risk_multiplier = spedo.bind_var("rules:fraud:risk_score_multiplier", default=1.35)
max_login_retry = spedo.bind_var("rules:auth:max_login_attempts", default=5)

def validate_payment(amount: float, user_risk_score: float):
    adjusted_risk = user_risk_score * risk_multiplier.value
    
    # Fast in-memory evaluation (0 ms network):
    if amount > max_tx_amount.value or adjusted_risk > 75.0:
        return {"status": "REQUIRES_MANUAL_REVIEW"}
    return {"status": "APPROVED"}

# During Black Friday or emergency risk alert, update threshold instantly:
max_tx_amount.value = 1200.0

# 3. Dynamic JSON dictionary / list configurations
cluster_routing = spedo.bind_var("config:geo_routing_table", default={
    "us-east": "https://us.gateway.spedo.internal",
    "eu-west": "https://eu.gateway.spedo.internal",
    "ap-south": "https://ap.gateway.spedo.internal"
})

allowed_ip_whitelist = spedo.bind_var("security:ip_whitelist", default=[
    "10.0.0.1", "10.0.0.2", "192.168.1.100"
])

# Reading nested configuration locally in 0 µs:
def route_request(region: str, client_ip: str):
    if client_ip not in allowed_ip_whitelist.value:
        return "Access Denied"
    return cluster_routing.value.get(region, cluster_routing.value["us-east"])

import asyncio
from spedo.client import SpedoClient

spedo = SpedoClient(host="127.0.0.1", port=6380)

# 4. Asynchronous waiting for distributed state transitions
job_status = spedo.bind_var("pipeline:batch_job_9410:status", default="PENDING")

async def worker_pipeline():
    print("Waiting for job to be marked READY...")
    # Suspends async coroutine until value equals 'READY' (Zero CPU busy-looping!):
    await job_status.wait_until("READY", timeout=15.0)
    
    print("Processing batch data in parallel...")
    await asyncio.sleep(2.0)
    
    # Notify all downstream listeners:
    job_status.value = "COMPLETED"

from dataclasses import dataclass
from spedo.client import SpedoClient

@dataclass
class UserSession:
    user_id: int
    username: str
    roles: list[str]
    is_vip: bool = False

spedo = SpedoClient(host="127.0.0.1", port=6380)

# 5. Synchronize custom typed OOP domain objects via pickle
session_state = spedo.bind_var(
    "live:session:user_4812",
    default=UserSession(user_id=4812, username="alex", roles=["viewer"]),
    serializer="pickle",
    trusted=True  # Safe for internal application keyspace
)

# Read typed properties locally with zero network hops:
if "admin" in session_state.value.roles:
    enable_admin_portal()

# Mutate domain object across microservices:
session_state.value = UserSession(user_id=4812, username="alex", roles=["viewer", "admin"], is_vip=True)

# === POD #2 : BILLING & GATEWAY SERVICE (e.g. app_pod_2.py) ===
from spedo import auto_sync

# 1. Same standard Python variables
daily_revenue = 0
active_orders = 0
promo_banner  = ""

# 2. Standard reader or checkout functions (NO decorators required)
def get_cluster_status():
    # Spedo pulls latest remote values before function execution automatically!
    global daily_revenue, active_orders, promo_banner
    return {
        "total_revenue": daily_revenue,
        "total_orders": active_orders,
        "current_promo": promo_banner
    }

# ⚡ 3. ONE SINGLE LINE at the end of the module:
auto_sync(globals())

# Execution on Pod #2 (Immediately after Pod #1 mutated the state):
status = get_cluster_status()
print("Pod #2 reads:", status)
# Output on Pod #2:
# {'total_revenue': 120, 'total_orders': 1, 'current_promo': 'FLASH_50_OFF'}
# ✓ 0 network code written, 0 decorators, seamless cross-pod consistency!

# ⚡ STEP-BY-STEP SYNCHRONIZATION TIMELINE:

1. [Pod #1 Init] : auto_sync scans bytecode -> Finds globals ['daily_revenue', 'active_orders', 'promo_banner'].
2. [Pod #1 Call] : record_checkout(120) modifies local variables in RAM (0.01 µs).
3. [Pod #1 Push] : auto_sync detects variable changes -> sends atomic MSET to Spedo (< 0.2 ms).
4. [Spedo Core]  : Spedo updates in-memory keyspace and broadcasts invalidation frame.
5. [Pod #2 Call] : get_cluster_status() is invoked on Pod #2.
6. [Pod #2 Pull] : auto_sync pulls updated keys before execution -> globals() populated instantly.
7. [Pod #2 Read] : Function executes reading updated global values seamlessly.

# =========================================================================
# 🔴 TRADITIONAL SQLITE3 (Boilerplate SQL, Table Creation, Cursors)
# =========================================================================
import sqlite3, json

conn = sqlite3.connect(":memory:")
cur = conn.cursor()
cur.execute("CREATE TABLE users (id TEXT PRIMARY KEY, data TEXT)")
cur.execute("INSERT INTO users VALUES (?, ?)", ("u1", json.dumps({"name": "Alice", "age": 30})))
conn.commit()

cur.execute("SELECT data FROM users WHERE id = ?", ("u1",))
user_data = json.loads(cur.fetchone()[0])
# ❌ Problem: SQL serialization overhead + table locks under multi-threading


# =========================================================================
# ⚡ SPEDO NATIVE EMBEDDED (Clean, 20x Faster, Direct Object Manipulation)
# =========================================================================
import spedo

db = spedo.connect(":memory:")
db.set("u1", {"name": "Alice", "age": 30})

# Atomic sub-path read & update without rewriting the whole record:
db.json_set("u1", "$.age", 31)
print(db.json_get("u1", "$.age"))  # -> 31 (0.24 μs)

import spedo

db = spedo.connect(":memory:")

# 1. Create a 64-dimensional Cosine Vector index directly in memory:
db.vector_create("knowledge_base", dimension=64, metric="cosine")

# 2. Add embeddings (from OpenAI, Ollama, HuggingFace):
db.vector_add("knowledge_base", "doc_auth", [0.12] * 64)
db.vector_add("knowledge_base", "doc_billing", [-0.35] * 64)

# 3. In-Situ SIMD AVX2 Vector Search in 0.9 ms (Zero SQLite compilation pain):
top_matches = db.vector_search("knowledge_base", [0.14] * 64, limit=1)
print("Nearest Doc:", top_matches)  # -> [('doc_auth', 0.999)]

import spedo

db = spedo.connect(":memory:")

# 🚀 Lock-Free Producer / Consumer (Never triggers "database is locked"):
def producer_worker():
    for i in range(1000):
        db.qpush("job_queue", f"task_{i}", priority=1 if i % 10 == 0 else 0)

def consumer_worker():
    while True:
        tasks = db.qpop("job_queue", count=10)
        if not tasks: break
        # Process tasks in priority order...

from spedo import SpedoList

# 1. Instantiate distributed list
activity_feed = SpedoList("global_audit_feed")

# 2. Standard list operations (append, pop, indexing, iteration)
activity_feed.append({"event": "LOGIN", "user": "alice", "ip": "10.0.0.1"})
activity_feed.append({"event": "CHECKOUT", "amount": 120.0})

print("Feed Length:", len(activity_feed))      # -> 2
print("Latest Event:", activity_feed[-1])       # -> {'event': 'CHECKOUT', 'amount': 120.0}

# 3. Pop oldest or latest items atomically:
oldest = activity_feed.pop(0)                   # FIFO pop

# Direct RESP CLI Commands (Zero SDK dependency):
VECTOR.ADD catalog:embeddings prod_4892 "[0.124, 0.854, -0.342, 0.091]"
VECTOR.SEARCH catalog:embeddings "[0.118, 0.840, -0.310, 0.088]" 5
MMUL weights:layer1 user_embedding:session_982

# Run the standalone demo and performance benchmark:
python3 demos/use_case_1_ai_in_db_vector_search/semantic_cache_demo.py
python3 demos/use_case_1_ai_in_db_vector_search/benchmark_simd_vs_python.py

# Configure automated NVMe Tiering:
SPEDO_COLD_ARCHIVE_PATH=/mnt/nvme/spedo_cold_data \
SPEDO_COLD_ARCHIVE_THRESHOLD_SECS=3600 \
SPEDO_MAX_MEMORY=48GB \
./spedo

# Run standalone demo & density benchmark:
python3 demos/use_case_2_finops_cold_tiering/rebuildable_cache_demo.py
python3 demos/use_case_2_finops_cold_tiering/benchmark_finops_density.py

from spedo import auto_sync

# 1. Declare standard Python variables (NO wrappers, NO special objects)
payment_total = 0
active_orders = 0
promo_banner  = "WELCOME_2026"

# 2. Write 100% natural Python functions (NO decorators required)
def process_payment(amount: int):
    global payment_total, active_orders
    payment_total = int(payment_total) + amount
    active_orders = int(active_orders) + 1

def update_promo(new_banner: str):
    global promo_banner
    promo_banner = new_banner

# ⚡ 3. ONE SINGLE LINE at the end of the file:
# Auto-discovers all global variables via bytecode and syncs across all nodes!
auto_sync(globals())

# Instant execution: pulls from Spedo before call, pushes immediately on mutation
process_payment(50)

# Admin / Control Plane modifying values dynamically:
client.set("promotions:flash_sale", {"discount_pct": 25})
client.set("config:maintenance", {"enabled": True, "message": "Deploying v2.0"})

# -> Spedo instantly pushes binary invalidation events to all subscribed microservices
# -> Microservices update local L1 RAM references in < 0.5 ms without restarts!

# 1. Run standalone Use Case 3 benchmark:
python3 demos/use_case_3_realtime_microservices/benchmark_zero_wire.py

# Real benchmark output:
📊 Redis 7.4 (Network TCP Access)     :     3,552 ops/sec | Latency: 281.56 µs
📊 Spedo LiveVar L1 (In-Process)       : 1,899,748 ops/sec | Latency:   0.24 µs
🏆 RESULT : 534.9× FASTER | 1134× LESS LATENCY | 0 PACKETS ON THE WIRE !

# 2. Launch full standalone containerized demo (API + Admin + Spedo):
cd demos/use_case_3_realtime_microservices && docker compose up --build

# =========================================================================
# 🚀 POD 1: INGESTION SERVICE (Seeds Spedo RAM + Produces 64B to Kafka)
# =========================================================================
def pod1_ingest(tx_id: str, full_payload: dict):
    client.set(f"spd_tx:{tx_id}", json.dumps(full_payload))
    ticket = {"tx_id": tx_id, "stage": 1, "ts": time.time()}
    kafka_producer.send("spedo.stage.1", json.dumps(ticket).encode())

# =========================================================================
# 🧠 POD 2: AI FRAUD WORKER (Consumes 64B Ticket -> SIMD RAM -> Kafka Stage 2)
# =========================================================================
def pod2_fraud_worker(kafka_ticket_message):
    ticket = json.loads(kafka_ticket_message.decode())
    tx_id = ticket["tx_id"]
    client.vector_search("fraud_vdb", query_embedding, limit=1)
    client.json_set(f"spd_tx:{tx_id}", "$.fraud_score", 0.015)
    ticket["stage"] = 2
    kafka_producer.send("spedo.stage.2", json.dumps(ticket).encode())

# =========================================================================
# 🛡️ POD 3: LIMITS WORKER (Consumes 64B Ticket -> Updates RAM -> Kafka Stage 3)
# =========================================================================
def pod3_limits_worker(kafka_ticket_message):
    ticket = json.loads(kafka_ticket_message.decode())
    client.json_set(f"spd_tx:{ticket['tx_id']}", "$.limit_checked", True)
    ticket["stage"] = 3
    kafka_producer.send("spedo.stage.3", json.dumps(ticket).encode())

# =========================================================================
# 🚦 POD 4: ROUTING WORKER (Consumes 64B Ticket -> Updates RAM -> Kafka Stage 4)
# =========================================================================
def pod4_routing_worker(kafka_ticket_message):
    ticket = json.loads(kafka_ticket_message.decode())
    client.json_set(f"spd_tx:{ticket['tx_id']}", "$.routed_gateway", "GW_1")
    ticket["stage"] = 4
    kafka_producer.send("spedo.stage.4", json.dumps(ticket).encode())

# =========================================================================
# 💳 POD 5: SETTLEMENT WORKER (Consumes 64B Ticket -> SETTLED in RAM)
# =========================================================================
def pod5_settlement_worker(kafka_ticket_message):
    ticket = json.loads(kafka_ticket_message.decode())
    client.json_set(f"spd_tx:{ticket['tx_id']}", "$.status", "SETTLED")

# =========================================================================
# 🔴 POD 1: INGESTION (Writes 50KB to Redis + Publishes 50KB to Kafka Topic 1)
# =========================================================================
def legacy_pod1_ingest(tx_id: str, full_payload: dict):
    raw_json = json.dumps(full_payload)
    r.set(f"red_tx:{tx_id}", raw_json)
    kafka_producer.send("redis.stage.1", raw_json.encode())

# =========================================================================
# 🔴 POD 2: FRAUD WORKER (50KB JSON.loads + NumPy + 50KB JSON.dumps to Kafka)
# =========================================================================
def legacy_pod2_fraud_worker(raw_kafka_message):
    payload = json.loads(raw_kafka_message.decode())
    doc = json.loads(r.get(f"red_tx:{payload['tx_id']}"))
    doc["fraud_score"] = 0.015
    updated = json.dumps(doc)
    r.set(f"red_tx:{payload['tx_id']}", updated)
    kafka_producer.send("redis.stage.2", updated.encode())

# =========================================================================
# 🔴 POD 3: LIMITS WORKER (50KB JSON.loads + Redis Round-Trip + 50KB to Kafka)
# =========================================================================
def legacy_pod3_limits_worker(raw_kafka_message):
    payload = json.loads(raw_kafka_message.decode())
    doc = json.loads(r.get(f"red_tx:{payload['tx_id']}"))
    doc["limit_checked"] = True
    updated = json.dumps(doc)
    r.set(f"red_tx:{payload['tx_id']}", updated)
    kafka_producer.send("redis.stage.3", updated.encode())

# =========================================================================
# 🔴 POD 4: ROUTING WORKER (50KB JSON.loads + Redis Round-Trip + 50KB to Kafka)
# =========================================================================
def legacy_pod4_routing_worker(raw_kafka_message):
    payload = json.loads(raw_kafka_message.decode())
    doc = json.loads(r.get(f"red_tx:{payload['tx_id']}"))
    doc["routed_gateway"] = "GW_1"
    updated = json.dumps(doc)
    r.set(f"red_tx:{payload['tx_id']}", updated)
    kafka_producer.send("redis.stage.4", updated.encode())

# =========================================================================
# 🔴 POD 5: SETTLEMENT WORKER (50KB JSON.loads + Redis SETTLED + Save Disk)
# =========================================================================
def legacy_pod5_settlement_worker(raw_kafka_message):
    payload = json.loads(raw_kafka_message.decode())
    doc = json.loads(r.get(f"red_tx:{payload['tx_id']}"))
    doc["status"] = "SETTLED"
    r.set(f"red_tx:{payload['tx_id']}", json.dumps(doc))

# 1. Navigate to standalone demo directory:
cd demos/use_case_4_kafka_claim_check

# 2. Launch the complete live 3-way benchmark stack:
docker compose up --build --abort-on-container-exit

# 3. Or run modular scripts against local services:
python3 run_redis_pipeline.py 200
python3 run_spedo_pipeline.py 200
python3 run_spedo_pure_pipeline.py 200

=========================================================================================================
🏆 FINAL MEASURED PRODUCTION RESULTS (REAL KAFKA BROKER + CONTAINERS + NATIVE QUEUES)
=========================================================================================================
METRIC                           | REDIS + FAT KAFKA  | SPEDO + KAFKA TICKET | 100% PURE SPEDO (0 KAFKA)
---------------------------------------------------------------------------------------------------------
Kafka Network Traffic            |         3333.16 KB |             47.03 KB |     0.00 KB (Zero Broker)
End-to-End Latency               |         4833.46 ms |           3173.27 ms |                 512.80 ms
Effective Throughput             |          41.4 tx/s |            63.0 tx/s |                  390.0 tx/s
Throughput Speedup               |    Baseline (1.0x) |         +46% (+1.5x) |     🚀 +842% (9.4x Faster)
External Broker Daemons          | 1 Kafka / Redpanda |   1 Kafka / Redpanda |        0 (Built-in Spedo)
CPU JSON Parse Operations        |               2000 |       0 (RAM Native) |            0 (RAM Native)
AI Fraud Vector Check            |       Scan + NumPy |   SIMD In-DB (0.9ms) |        SIMD In-DB (0.9ms)
=========================================================================================================

# 1. Verify Live Spedo RAM Cache Hit on Documentation:
curl -sI http://127.0.0.1:8090/docs/index.html
HTTP/1.1 200 OK
Content-Type: text/html; charset=utf-8
X-Cache: HIT-SPEDO
X-Cache-Lookup-Us: 142
X-Cache-Engine: Spedo-In-Memory-Runtime

# 2. Verify Strict Operator Isolation (Admin routes never cached):
curl -sI http://127.0.0.1:8090/admin.html
HTTP/1.1 200 OK
X-Cache: BYPASS-ADMIN

# Instant Operator Purge via Authenticated REST API:
curl -X POST http://127.0.0.1:8090/api/admin/cache/purge \
  -H "Authorization: Bearer <ADMIN_TOKEN>"

# Response:
{
  "status": "success",
  "purged_count": 14,
  "duration_ms": 0.48,
  "message": "Successfully purged 14 page(s) from Spedo in-memory cache."
}