# Root Cause Hypotheses

The intermittent duplicate benchmark rows in a Python/FastAPI service backed by SQLite typically stem from a combination of async concurrency patterns, missing idempotency guarantees, and SQLite's single-writer semantics interacting poorly with retry logic. Below are the most probable root causes, ranked by likelihood:

1. **Async Race Condition on `create_run`**: FastAPI's async event loop allows multiple coroutines to execute concurrently. If two requests (e.g., OpenWebUI retry + benchmark runner) invoke `create_run` simultaneously, both may pass a `SELECT` check for existence, proceed to `INSERT`, and commit before either transaction completes. SQLite's default journal mode does not prevent this logical race; it only serializes physical writes.

2. **Missing Database-Level Uniqueness**: If the `runs` table lacks a `UNIQUE` constraint on the run identifier (e.g., `run_id`, `benchmark_name`, `timestamp`, or `client_id`), SQLite will happily accept duplicate rows. Application-level checks (`if not exists: insert`) are inherently racy in async contexts without explicit locking.

3. **Idempotency Violation in Retry Logic**: Clients (OpenWebUI, agent client, benchmark harness) often retry on 5xx, timeouts, or network drops. If the retry logic does not propagate or validate an idempotency key, the server treats each retry as a new request, triggering duplicate `create_run` calls.

4. **Streaming Endpoint Lifecycle Overlap**: `StreamingResponse` in FastAPI does not block the event loop. The endpoint may call `create_run`, begin yielding chunks, and then be interrupted by a client disconnect or timeout. If the client retries, the endpoint may re-execute `create_run` before the previous run's metadata is fully committed or cleaned up, resulting in duplicate rows.

5. **SQLite Busy/Timeout Misconfiguration**: Without `PRAGMA busy_timeout` and WAL mode, concurrent writes can fail with `database is locked`. If the application silently swallows this error or retries without checking the DB state, it may issue a second `INSERT`, creating duplicates.

6. **Helper Function Transaction Boundaries**: If `create_run`, `create_llm_request`, `finish_llm_request`, and `add_event` are called across separate database connections or without explicit transaction wrapping, partial commits can leave orphaned or duplicate metadata. SQLite's default autocommit mode exacerbates this.

# Evidence To Collect

Before applying fixes, gather the following evidence to confirm hypotheses and validate the patch:

1. **Database Schema Inspection**: Run `PRAGMA table_info(runs);` and `PRAGMA index_list(runs);` to verify existing constraints and indexes. Check for missing `UNIQUE` constraints on run identifiers.
2. **Log Analysis**: Search for patterns like `run_id=...`, `INSERT OR IGNORE`, `database is locked`, `retry`, and `timeout`. Look for duplicate `run_id` generation timestamps within <100ms windows.
3. **SQLite Statistics**: Enable `PRAGMA cache_size=-2000;` and monitor `PRAGMA page_count;`, `PRAGMA freelist_count;`, and `PRAGMA wal_checkpoint(TRUNCATE);` to assess write contention.
4. **Request Tracing**: Instrument FastAPI with OpenTelemetry or middleware to track request lifecycle: `create_run` → `create_llm_request` → streaming → `finish_llm_request`. Identify overlapping request IDs.
5. **Client Behavior Audit**: Check OpenWebUI, benchmark runner, and agent client retry configurations. Look for missing idempotency headers (`Idempotency-Key`, `X-Request-ID`) or aggressive retry policies (e.g., exponential backoff without jitter).
6. **Connection Pool Metrics**: If using `aiosqlite` or `sqlite3` with connection pooling, check for pool exhaustion, connection churn, or stale connections causing re-initialization of DB state.
7. **Error Rate Correlation**: Map duplicate row spikes to specific endpoints, client types, or time windows. Correlate with network latency spikes or benchmark scheduler cycles.
8. **Transaction Log Review**: If using `sqlite3` with `isolation_level=None`, verify that explicit `BEGIN`/`COMMIT` blocks are used. Autocommit mode can cause partial commits in async contexts.

# Patch Plan

The fix prioritizes database-level guarantees, application-level idempotency, and async-safe concurrency control. It is designed for a private home lab but follows production engineering standards.

1. **Enforce DB-Level Idempotency**:
   - Add a `UNIQUE` constraint on the run identifier column(s).
   - Replace `INSERT` with `INSERT OR IGNORE` or `ON CONFLICT DO NOTHING`.
   - Wrap metadata writes in explicit transactions.

2. **Application-Level Concurrency Guardrails**:
   - Introduce an `asyncio.Lock` scoped to the benchmark/agent client namespace to serialize `create_run` calls.
   - Implement idempotency key validation in middleware.
   - Ensure all helper functions use the same DB connection/transaction context.

3. **SQLite Configuration Hardening**:
   - Enable WAL mode: `PRAGMA journal_mode=WAL;`
   - Set busy timeout: `PRAGMA busy_timeout=5000;`
   - Set synchronous mode: `PRAGMA synchronous=NORMAL;`
   - Optimize cache: `PRAGMA cache_size=-2000;`

4. **Streaming Endpoint Refactor**:
   - Commit run metadata before initiating `StreamingResponse`.
   - Use a two-phase approach: phase 1 creates run + LLM request, phase 2 streams and updates on completion/failure.
   - Handle `ClientDisconnected` gracefully without retrying metadata creation.

5. **Observability & Alerting**:
   - Add metrics for `sqlite_insert_conflicts`, `async_lock_acquisitions`, `streaming_retries`.
   - Implement a periodic deduplication job (e.g., `DELETE FROM runs WHERE rowid NOT IN (SELECT MIN(rowid) FROM runs GROUP BY run_id);`).

6. **Code Review & Refactoring**:
   - Ensure `create_run`, `create_llm_request`, `finish_llm_request`, and `add_event` share a single DB connection per request.
   - Remove implicit autocommit reliance; use `async with db.transaction():` patterns.
   - Add type hints and explicit error handling for `sqlite3.IntegrityError`.

# SQLite Constraints And Indexes

Proper schema design is the first line of defense against duplicates. Below is a production-grade SQLite schema tailored for this workload.

```sql
-- Enable WAL mode and concurrency settings
PRAGMA journal_mode=WAL;
PRAGMA busy_timeout=5000;
PRAGMA synchronous=NORMAL;
PRAGMA cache_size=-2000;

-- Runs table with strict uniqueness
CREATE TABLE IF NOT EXISTS runs (
    run_id TEXT PRIMARY KEY,
    benchmark_name TEXT NOT NULL,
    client_type TEXT NOT NULL CHECK(client_type IN ('openwebui', 'benchmark', 'agent')),
    status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'running', 'completed', 'failed')),
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    updated_at TEXT NOT NULL DEFAULT (datetime('now')),
    UNIQUE(benchmark_name, client_type, created_at) -- Composite uniqueness to prevent logical duplicates
);

-- LLM requests table
CREATE TABLE IF NOT EXISTS llm_requests (
    request_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    model TEXT NOT NULL,
    prompt_hash TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'streaming', 'completed', 'failed')),
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE
);

-- Events table for granular tracking
CREATE TABLE IF NOT EXISTS events (
    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id TEXT NOT NULL,
    request_id TEXT,
    event_type TEXT NOT NULL,
    payload TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE,
    FOREIGN KEY (request_id) REFERENCES llm_requests(request_id) ON DELETE SET NULL
);

-- Indexes for query performance
CREATE INDEX IF NOT EXISTS idx_runs_client_status ON runs(client_type, status);
CREATE INDEX IF NOT EXISTS idx_llm_requests_run_status ON llm_requests(run_id, status);
CREATE INDEX IF NOT EXISTS idx_events_run_type ON events(run_id, event_type);
```

**Python-Level Guardrails:**

```python
import asyncio
import sqlite3
from contextlib import asynccontextmanager
from typing import Optional

# Asyncio lock to serialize metadata creation per benchmark/client
_run_creation_locks: dict[str, asyncio.Lock] = {}

def get_run_lock(key: str) -> asyncio.Lock:
    if key not in _run_creation_locks:
        _run_creation_locks[key] = asyncio.Lock()
    return _run_creation_locks[key]

@asynccontextmanager
async def db_transaction(db_path: str):
    conn = sqlite3.connect(db_path, timeout=5.0)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=5000")
    conn.execute("PRAGMA synchronous=NORMAL")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

async def create_run(run_id: str, benchmark_name: str, client_type: str) -> dict:
    lock_key = f"{benchmark_name}:{client_type}"
    lock = get_run_lock(lock_key)
    
    async with lock:
        async with db_transaction("benchmark.db") as conn:
            # Idempotent insert with conflict resolution
            stmt = """
                INSERT OR IGNORE INTO runs (run_id, benchmark_name, client_type, status)
                VALUES (?, ?, ?, 'pending')
            """
            cursor = conn.execute(stmt, (run_id, benchmark_name, client_type))
            if cursor.rowcount == 0:
                # Already exists, fetch existing
                row = conn.execute("SELECT * FROM runs WHERE run_id = ?", (run_id,)).fetchone()
                return dict(zip([desc[0] for desc in cursor.description], row)) if row else {}
            
            return {"run_id": run_id, "status": "pending"}
```

**Concurrency Risks Addressed:**
- `asyncio.Lock` prevents logical race conditions in the event loop.
- `INSERT OR IGNORE` prevents physical duplicates even if lock is bypassed.
- Explicit transaction boundaries ensure atomicity across helper functions.
- `busy_timeout` prevents `database is locked` errors from triggering silent retries.

# Streaming Edge Cases

Streaming endpoints introduce unique concurrency and lifecycle challenges:

1. **Partial Stream Completion**: If a client disconnects mid-stream, `finish_llm_request` may never be called. The run remains in `running` state. Fix: Use `asyncio.TaskGroup` or `contextlib.AsyncExitStack` to guarantee cleanup on disconnect. Register a callback that updates status to `failed` if not completed within a timeout.

2. **Retry During Streaming**: Clients may retry while the stream is active. The server must recognize the existing `run_id` and either resume the stream or return a `409 Conflict` with the current status. Implement idempotency middleware that checks `runs.status` before allowing new `create_run` calls.

3. **Event Loop Blocking**: Streaming yields chunks asynchronously, but if `add_event` or `finish_llm_request` are called synchronously within the stream loop, they can block the event loop. Fix: Use `asyncio.to_thread()` for DB writes or queue them via `asyncio.Queue` with a background worker.

4. **Connection Pool Exhaustion**: Long-lived streaming connections hold DB connections if not properly released. Fix: Use connection pooling with `max_connections=10`, `min_connections=2`, and ensure connections are returned to the pool immediately after metadata writes.

5. **Idempotency Key Propagation**: Streaming endpoints must accept and validate `Idempotency-Key` headers. If present, check if a run with that key exists. If so, return the existing run ID and resume streaming from the last checkpoint (if supported) or return a `200 OK` with cached metadata.

6. **Graceful Shutdown**: On service restart, streaming connections drop. Implement a shutdown hook that scans for `runs.status = 'running'` older than 5 minutes and marks them `failed`. This prevents orphaned runs from accumulating.

# Test Plan

The following 12+ acceptance tests validate the fix across normal, concurrent, streaming, and failure scenarios. Each test includes validation criteria and implementation hints.

1. **T01: Normal Run Creation**
   - *Description*: Send a single non-streaming request.
   - *Validation*: Exactly one row in `runs` with `status='pending'`, `created_at` matches request time.
   - *Hint*: Use `pytest` with `aiosqlite` fixture.

2. **T02: Concurrent Identical Requests**
   - *Description*: Fire 10 identical requests simultaneously with same `run_id`.
   - *Validation*: Exactly one row in `runs`. `INSERT OR IGNORE` prevents duplicates.
   - *Hint*: Use `asyncio.gather()` to simulate concurrency.

3. **T03: Idempotency Key Enforcement**
   - *Description*: Send two requests with same `Idempotency-Key` but different `run_id`.
   - *Validation*: Second request returns `409 Conflict` with existing `run_id`. No new row created.
   - *Hint*: Middleware extracts key, checks `runs` table, returns early if exists.

4. **T04: Streaming Run Creation**
   - *Description*: Send streaming request, verify metadata written before first chunk.
   - *Validation*: `runs` row exists with `status='running'` before `StreamingResponse` yields.
   - *Hint*: Use `asyncio.Event` to signal metadata commit before stream starts.

5. **T05: Streaming Client Disconnect**
   - *Description*: Disconnect client after 50% stream.
   - *Validation*: Run status updates to `failed` within 2 seconds. No duplicate rows.
   - *Hint*: Use `contextlib.AsyncExitStack` with cleanup callback on `ClientDisconnected`.

6. **T06: Retry Storm Simulation**
   - *Description*: Simulate 50 retries with exponential backoff.
   - *Validation*: Exactly one row in `runs`. `sqlite_insert_conflicts` metric increments.
   - *Hint*: Use `tenacity` library with `retry_if_exception_type`.

7. **T07: SQLite Busy Timeout Handling**
   - *Description*: Force `database is locked` error via concurrent writes.
   - *Validation*: Request retries successfully without creating duplicate. Status remains consistent.
   - *Hint*: Monkey-patch `sqlite3` to raise `OperationalError` on 3rd write.

8. **T08: Transaction Rollback on Error**
   - *Description*: Fail `create_llm_request` after `create_run`.
   - *Validation*: `runs` row rolled back. No orphaned metadata.
   - *Hint*: Use `try/except` with `conn.rollback()` in transaction context.

9. **T09: Index Usage Verification**
   - *Description*: Query `runs` by `client_type` and `status`.
   - *Validation*: `EXPLAIN QUERY PLAN` shows index scan, not full table scan.
   - *Hint*: Use `sqlite3` CLI or `aiosqlite` with `EXPLAIN`.

10. **T10: WAL Mode Checkpoint Performance**
    - *Description*: Run 1000 concurrent writes, trigger checkpoint.
    - *Validation*: No `database is locked` errors. Checkpoint completes in <1s.
    - *Hint*: Use `PRAGMA wal_checkpoint(TRUNCATE);` after batch.

11. **T11: Helper Function Atomicity**
    - *Description*: Call `create_run`, `create_llm_request`, `add_event` in sequence.
    - *Validation*: All three succeed or all three roll back. No partial state.
    - *Hint*: Use single transaction context across all calls.

12. **T12: Graceful Shutdown Cleanup**
    - *Description*: Stop service while 5 runs are `running`.
    - *Validation*: On restart, all 5 marked `failed`. No duplicates.
    - *Hint*: Register `atexit` or FastAPI `shutdown` event with cleanup query.

13. **T13: Idempotency Key Collision Prevention**
    - *Description*: Send requests with same `Idempotency-Key` but different `benchmark_name`.
    - *Validation*: Both succeed. Keys are scoped to `benchmark_name:client_type`.
    - *Hint*: Composite key in middleware: `f"{benchmark_name}:{client_type}:{idempotency_key}"`.

14. **T14: Memory Leak Verification**
    - *Description*: Run 10,000 requests over 10 minutes.
    - *Validation*: Memory usage stable. No unclosed connections or locks.
    - *Hint*: Use `tracemalloc` and `psutil` to monitor.

# Rollback Plan

A safe rollback ensures minimal downtime and data integrity if the patch introduces unexpected issues.

1. **Version Control & Deployment**:
   - Tag the current stable version (`v1.2.0`).
   - Deploy patch as `v1.3.0-rc1` with feature flag `enable_sqlite_idempotency=false`.
   - Use blue-green or canary deployment to isolate impact.

2. **Database Migration Rollback**:
   - Create reverse migration script:
     ```sql
     DROP INDEX IF EXISTS idx_runs_client_status;
     DROP INDEX IF EXISTS idx_llm_requests_run_status;
     DROP INDEX IF EXISTS idx_events_run_type;
     -- Keep UNIQUE constraint for safety; remove only if absolutely necessary
     -- ALTER TABLE runs DROP CONSTRAINT unique_benchmark_client_created;
     ```
   - Test rollback in staging with production-like data volume.

3. **Feature Flag Toggle**:
   - If duplicates reappear, set `enable_sqlite_idempotency=false`.
   - Revert to legacy `INSERT` logic without locks or `INSERT OR IGNORE`.
   - Monitor for 15 minutes to confirm stability.

4. **Data Cleanup**:
   - If duplicates exist post-rollback, run:
     ```sql
     DELETE FROM runs WHERE rowid NOT IN (
         SELECT MIN(rowid) FROM runs GROUP BY run_id
     );
     ```
   - Verify referential integrity with `PRAGMA foreign_key_check;`.

5. **Communication & Monitoring**:
   - Notify team of rollback via incident channel.
   - Keep dashboards active for `sqlite_insert_conflicts`, `async_lock_acquisitions`, `streaming_retries`.
   - Document rollback steps in runbook for future incidents.

# Decision Summary

The duplicate benchmark rows stem from async race conditions, missing idempotency guarantees, and SQLite's single-writer semantics interacting poorly with retry logic and streaming endpoints. The fix enforces database-level uniqueness via `UNIQUE` constraints and `INSERT OR IGNORE`, applies application-level concurrency control with `asyncio.Lock`, and hardens SQLite configuration with WAL mode and busy timeouts. Streaming endpoints are refactored to commit metadata before yielding chunks, with cleanup callbacks for disconnects and retries.

This approach is safe for a private home lab (no external dependencies, minimal operational overhead) while meeting engineering-grade standards (idempotency, transactional integrity, observability, rollback safety). The patch plan includes concrete SQL, Python guardrails, concurrency risk mitigation, streaming edge case handling, 14 acceptance tests, and a structured rollback strategy. Monitoring and deduplication jobs ensure long-term data integrity. The solution balances simplicity with robustness, prioritizing correctness over performance where contention occurs, and scales gracefully within home-lab resource constraints.