# Root Cause Hypotheses

The intermittent duplicate benchmark rows in a Python/FastAPI service backed by SQLite typically stem from a combination of application-level race conditions, missing database-level safeguards, and proxy/client retry behavior. Below are the primary hypotheses, ranked by likelihood and impact:

1. **TOCTOU Race in `create_run`**: The most common cause. If `create_run` performs a `SELECT` to check for an existing run, then issues an `INSERT` if none exists, concurrent FastAPI coroutines can both pass the check before either commits. SQLite serializes writes at the database level, but Python's async event loop interleaves coroutines between I/O waits, allowing the race window to materialize.
2. **Missing Unique Constraints & Atomic Inserts**: Without a `UNIQUE` constraint on logical identifiers (e.g., `benchmark_id`, `client_id`, `session_token`, or a composite key), SQLite will happily accept duplicate rows. Relying on application logic alone for deduplication is fragile under concurrency.
3. **Proxy/Client Retry Amplification**: OpenWebUI, benchmark runners, and agent clients often implement exponential backoff retries on 5xx/4xx or timeouts. If the proxy forwards a request, the server begins processing, but the client times out and retries, the proxy may route the retry to the same or another worker, triggering a second `create_run` before the first completes.
4. **Streaming Lifecycle Mismatch**: FastAPI's `StreamingResponse` does not block the event loop. If a client disconnects mid-stream or the proxy load-balances, the server may still execute `finish_llm_request` or `add_event` in background tasks. Without idempotency guards, these functions can execute multiple times, creating phantom or duplicate metadata rows.
5. **SQLite Connection/Transaction Misconfiguration**: Using `sqlite3` in async contexts without explicit transaction boundaries, or running without WAL mode, can cause partial commits, journal contention, or lost updates. Python's `asyncio.Lock` may be missing around DB writes, or connections are being shared across tasks without proper isolation.
6. **Idempotency Key Absence**: The API lacks a client-provided idempotency header (e.g., `Idempotency-Key` or `X-Request-ID`). Without it, the server cannot distinguish between a legitimate new run and a retry of the same logical operation.

# Evidence To Collect

Before patching, gather the following telemetry to confirm hypotheses and validate the fix:

1. **SQLite Configuration**: Run `PRAGMA journal_mode;` and `PRAGMA wal_checkpoint(PASSIVE);`. Confirm if WAL is enabled. Check `PRAGMA synchronous;` (should be `FULL` or `NORMAL` for home-lab safety).
2. **Schema & Constraints**: Dump the `runs` table schema. Verify absence of `UNIQUE` indexes on logical keys. Check for foreign keys and their enforcement (`PRAGMA foreign_keys;`).
3. **Request Correlation Logs**: Enable structured logging with `X-Request-ID`, `X-Idempotency-Key`, and `correlation_id`. Capture timestamps for `create_run`, `create_llm_request`, `finish_llm_request`, and `add_event`.
4. **Proxy & Client Behavior**: Inspect Nginx/Traefik/Envoy logs for retry counts, timeout triggers, and concurrent connection spikes. Check OpenWebUI/agent client retry policies and streaming chunk intervals.
5. **Async Driver & Connection Pool**: Identify the SQLite driver (`aiosqlite`, `sqlite-async`, `tortoise-orm`, etc.). Check connection pool size, max retries, and whether transactions are explicitly scoped.
6. **Error Logs**: Search for `sqlite3.IntegrityError`, `sqlite3.OperationalError: database is locked`, `asyncio.TimeoutError`, and `ClientDisconnect` traces.
7. **Duplicate Pattern Analysis**: Query for duplicate rows: `SELECT benchmark_id, client_id, COUNT(*) FROM runs GROUP BY benchmark_id, client_id HAVING COUNT(*) > 1;`. Check if duplicates share identical timestamps or differ by milliseconds.
8. **Streaming Metrics**: Monitor `StreamingResponse` duration, chunk count, and client disconnect rates. Correlate with duplicate spikes.

# Patch Plan

The fix must be atomic, idempotent, and safe for async concurrency. Below is a phased, engineering-grade approach:

### Phase 1: Database-Level Safeguards
- Enable WAL mode for concurrent read/write safety.
- Add a composite `UNIQUE` constraint on logical identifiers.
- Replace `SELECT`-then-`INSERT` with atomic `INSERT OR IGNORE` or `INSERT ... ON CONFLICT DO NOTHING`.
- Wrap all write operations in explicit transactions.

### Phase 2: Application-Level Idempotency & Concurrency Guards
- Require an `Idempotency-Key` header for all run-creation endpoints.
- Implement an async lock scoped to `run_id` or `idempotency_key` to serialize concurrent attempts.
- Refactor `create_run` to return an existing run if a duplicate is detected, rather than failing or creating a new row.
- Ensure `finish_llm_request` and `add_event` check run status before executing, preventing double-finalization.

### Phase 3: Streaming Lifecycle Hardening
- Attach `finish_llm_request` to a `BackgroundTasks` queue with idempotency checks.
- Handle `ClientDisconnect` gracefully: cancel pending background tasks, but allow in-flight LLM requests to complete if already committed.
- Use `asyncio.Lock` per streaming session to prevent concurrent `add_event` calls from the same client.

### Phase 4: Code Refactoring (Python)
```python
# app/services/run_service.py
import asyncio
import logging
from typing import Optional
from contextlib import asynccontextmanager

logger = logging.getLogger(__name__)

# Global lock registry to prevent TOCTOU on run creation
_run_creation_locks: dict[str, asyncio.Lock] = {}
_locks_lock = asyncio.Lock()

async def _get_run_lock(key: str) -> asyncio.Lock:
    async with _locks_lock:
        if key not in _run_creation_locks:
            _run_creation_locks[key] = asyncio.Lock()
        return _run_creation_locks[key]

@asynccontextmanager
async def atomic_db_transaction(db_conn):
    """Ensure explicit transaction boundaries for aiosqlite."""
    await db_conn.execute("BEGIN IMMEDIATE;")
    try:
        yield db_conn
        await db_conn.commit()
    except Exception:
        await db_conn.rollback()
        raise

async def create_run(
    db_conn,
    benchmark_id: str,
    client_id: str,
    idempotency_key: str,
    metadata: dict
) -> dict:
    lock = await _get_run_lock(idempotency_key)
    async with lock:
        async with atomic_db_transaction(db_conn) as tx:
            # Atomic insert with conflict resolution
            await tx.execute(
                """
                INSERT OR IGNORE INTO runs 
                (idempotency_key, benchmark_id, client_id, status, metadata, created_at)
                VALUES (?, ?, ?, 'pending', ?, datetime('now'));
                """,
                (idempotency_key, benchmark_id, client_id, str(metadata))
            )
            # Fetch the row (either newly created or pre-existing)
            row = await tx.execute(
                "SELECT * FROM runs WHERE idempotency_key = ?",
                (idempotency_key,)
            ).fetchone()
            if not row:
                raise RuntimeError("Failed to create or locate run after atomic insert")
            return dict(row)

async def finish_llm_request(db_conn, run_id: str, final_status: str, metrics: dict):
    lock = await _get_run_lock(run_id)
    async with lock:
        async with atomic_db_transaction(db_conn) as tx:
            # Idempotent update: only transition if still in-progress
            await tx.execute(
                """
                UPDATE runs 
                SET status = ?, metrics = ?, finished_at = datetime('now')
                WHERE id = ? AND status IN ('running', 'pending');
                """,
                (final_status, str(metrics), run_id)
            )
            if (await tx.execute("SELECT changes();").fetchone())[0] == 0:
                logger.warning(f"finish_llm_request skipped for run {run_id}: already finalized or missing")
```

### Phase 5: Proxy & Client Alignment
- Configure proxy to forward `Idempotency-Key` unchanged.
- Set client retry timeout > expected LLM latency to reduce unnecessary retries.
- Enable `Connection: keep-alive` to reduce TCP handshake retries.

# SQLite Constraints And Indexes

SQLite's performance and correctness depend heavily on schema design. Below are the required constraints and indexes for a production-grade home-lab setup.

### DDL with Constraints
```sql
-- Enable WAL and synchronous mode for crash safety + concurrency
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA foreign_keys=ON;

CREATE TABLE IF NOT EXISTS runs (
    id TEXT PRIMARY KEY,
    idempotency_key TEXT NOT NULL UNIQUE,
    benchmark_id TEXT NOT NULL,
    client_id TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending', 'running', 'completed', 'failed', 'cancelled')),
    metadata TEXT,
    metrics TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    finished_at TEXT
);

CREATE TABLE IF NOT EXISTS llm_requests (
    id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    model TEXT NOT NULL,
    prompt_hash TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'queued' CHECK(status IN ('queued', 'streaming', 'completed', 'failed')),
    input_tokens INTEGER DEFAULT 0,
    output_tokens INTEGER DEFAULT 0,
    latency_ms REAL,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS events (
    id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    payload TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE
);
```

### Indexes
```sql
-- Fast lookup for idempotency and deduplication
CREATE INDEX idx_runs_idempotency ON runs(idempotency_key);

-- Query optimization for benchmark filtering and status tracking
CREATE INDEX idx_runs_benchmark_status ON runs(benchmark_id, status);
CREATE INDEX idx_runs_client_created ON runs(client_id, created_at DESC);

-- LLM request routing and aggregation
CREATE INDEX idx_llm_requests_run_status ON llm_requests(run_id, status);
CREATE INDEX idx_llm_requests_model_latency ON llm_requests(model, latency_ms);

-- Event timeline ordering
CREATE INDEX idx_events_run_created ON events(run_id, created_at);
```

### Why This Works
- `UNIQUE(idempotency_key)` + `INSERT OR IGNORE` eliminates TOCTOU races at the storage layer.
- `CHECK` constraints enforce state machine validity, preventing invalid transitions.
- `ON DELETE CASCADE` ensures cleanup when runs are removed.
- Composite indexes reduce full-table scans during concurrent benchmark queries.
- WAL mode allows readers to proceed while writers commit, critical for FastAPI's async concurrency.

# Streaming Edge Cases

Streaming responses introduce unique concurrency and lifecycle challenges. Below are the critical edge cases and mitigations:

### 1. Client Disconnect Mid-Stream
- **Risk**: `StreamingResponse` catches `ClientDisconnect`, but background tasks may still call `finish_llm_request` or `add_event`, creating duplicate or orphaned metadata.
- **Mitigation**: Use `asyncio.Lock` per `run_id`. In the streaming endpoint, attach cleanup to `BackgroundTasks` but wrap it in a status check. If the run is already finalized, skip.
- **Code Pattern**:
  ```python
  async def stream_response(run_id: str):
      try:
          async for chunk in generate_chunks(run_id):
              yield chunk
      except asyncio.CancelledError:
          logger.info(f"Stream cancelled for run {run_id}")
      finally:
          # Idempotent finalization
          await finish_llm_request(db, run_id, "completed", {})
  ```

### 2. Proxy Retry on Streaming Timeout
- **Risk**: Proxy times out waiting for first chunk, retries request. Server creates duplicate run.
- **Mitigation**: Idempotency key validation at the router level. Reject retries with `409 Conflict` or return the existing run's streaming state.
- **Header Enforcement**: `X-Idempotency-Key` required. If missing, generate server-side but log warning.

### 3. Concurrent `add_event` Calls
- **Risk**: Multiple client connections or proxy workers send events simultaneously.
- **Mitigation**: Use `INSERT OR IGNORE` on `events` with a composite unique key `(run_id, event_type, created_at)`. Or use `add_event` with an async lock per `run_id`.

### 4. Memory/Backpressure Leaks
- **Risk**: Unbounded streaming without client-side backpressure causes memory growth.
- **Mitigation**: Implement chunk size limits, `asyncio.sleep` for rate limiting, and `asyncio.wait_for` with timeout. Monitor `sys.getsizeof()` for response buffers.

### 5. LLM Request State Drift
- **Risk**: `create_llm_request` inserts a row, but streaming fails before `finish_llm_request` updates it. Run stays in `running` indefinitely.
- **Mitigation**: Implement a cron/periodic task that transitions stuck `running` runs to `failed` after a threshold. Use `status` checks in `finish_llm_request` to prevent double-updates.

# Test Plan

The following 12 acceptance tests validate the fix across concurrency, idempotency, streaming, and database integrity. Tests use `pytest-asyncio`, `httpx`, and in-memory SQLite with WAL enabled.

### 1. Concurrent Duplicate Prevention
- **Setup**: Spawn 10 async tasks calling `/runs` with identical `idempotency_key`.
- **Assert**: Exactly 1 row in `runs` table. All tasks return same `run_id`.
- **Pass/Fail**: Fail if `COUNT(*) > 1` or `run_id` differs.

### 2. Idempotency Key Enforcement
- **Setup**: Call `/runs` without `Idempotency-Key` header.
- **Assert**: Returns `400 Bad Request` or generates server-side key with warning log.
- **Pass/Fail**: Fail if duplicate rows created or silent acceptance.

### 3. Streaming Disconnect Handling
- **Setup**: Start stream, disconnect client after 3 chunks.
- **Assert**: `finish_llm_request` called exactly once. Run status transitions to `completed` or `cancelled`. No duplicate events.
- **Pass/Fail**: Fail if status stuck in `running` or multiple `finished_at` updates.

### 4. Proxy Retry Simulation
- **Setup**: Simulate proxy retry by sending 3 identical requests with same `Idempotency-Key` and 100ms delay.
- **Assert**: Only first request creates run. Subsequent return `200` with existing `run_id` or `409`.
- **Pass/Fail**: Fail if duplicate rows or inconsistent responses.

### 5. SQLite WAL Concurrency Stress
- **Setup**: 50 concurrent writers, 20 concurrent readers. 10s duration.
- **Assert**: Zero `database is locked` errors. All writes commit. No data corruption.
- **Pass/Fail**: Fail on any SQLite operational errors or missing rows.

### 6. Transaction Isolation Boundary
- **Setup**: Call `create_run` and `create_llm_request` in same transaction. Simulate failure mid-way.
- **Assert**: Both operations roll back. Table state unchanged.
- **Pass/Fail**: Fail if partial commit occurs.

### 7. Missing Idempotency Key Rejection
- **Setup**: Send 5 requests without key, different `benchmark_id`.
- **Assert**: All succeed but generate unique server-side keys. No duplicates.
- **Pass/Fail**: Fail if duplicates created or 500 errors.

### 8. Multiple Events Per Run
- **Setup**: Call `add_event` 20 times concurrently for same `run_id`.
- **Assert**: All events inserted. No duplicate payloads. Order preserved by `created_at`.
- **Pass/Fail**: Fail if `COUNT(events) != 20` or integrity errors.

### 9. Finish Request Idempotency
- **Setup**: Call `finish_llm_request` 5 times concurrently for same `run_id`.
- **Assert**: Only first succeeds. Subsequent return `200` with warning log. Status unchanged after first.
- **Pass/Fail**: Fail if `finished_at` updates multiple times or status invalid.

### 10. Index Performance Under Load
- **Setup**: Insert 10,000 runs, query by `benchmark_id` and `status`.
- **Assert**: Query time < 50ms. `EXPLAIN QUERY PLAN` shows index usage.
- **Pass/Fail**: Fail on full table scan or timeout.

### 11. Connection Pool Exhaustion Recovery
- **Setup**: Hold 100 connections open, then release. Spawn 50 new requests.
- **Assert**: Requests succeed after pool recovers. No connection leaks.
- **Pass/Fail**: Fail on `too many connections` or hanging tasks.

### 12. Schema Migration Safety
- **Setup**: Run migration adding `UNIQUE` constraint to existing 5,000 rows with duplicates.
- **Assert**: Migration fails gracefully or deduplicates first. Service remains available.
- **Pass/Fail**: Fail if migration crashes or leaves inconsistent state.

# Rollback Plan

Engineering-grade fixes require safe rollback paths, especially in a home-lab environment where downtime impacts benchmark continuity.

### 1. Versioned Deployment
- Deploy fix via container image tag (e.g., `v1.2.1-fix-duplicates`).
- Use Kubernetes/Docker Compose with `restart: unless-stopped` and health checks.
- Maintain previous image (`v1.2.0`) in registry for instant revert.

### 2. Database Backward Compatibility
- Migrations are additive only. New constraints/indexes do not break old schema.
- Duplicate rows remain readable. Application layer handles them gracefully until cleanup.
- Rollback script: `DROP INDEX IF EXISTS idx_runs_idempotency;` (if needed), revert code.

### 3. Feature Flag Toggle
- Wrap idempotency enforcement behind `ENABLE_IDEMPOTENCY=true` env var.
- Disable to revert to legacy behavior instantly if false positives occur.
- Monitor error rates post-disable.

### 4. Duplicate Data Cleanup
- Post-fix, run safe deduplication:
  ```sql
  DELETE FROM runs 
  WHERE id NOT IN (
    SELECT MIN(id) FROM runs GROUP BY idempotency_key
  );
  ```
- Verify with `SELECT COUNT(*) FROM runs;` before/after.
- Archive duplicates to `runs_archive` table for audit.

### 5. Rollback Triggers
- Error rate > 5% for 2 minutes.
- SQLite lock contention spikes.
- Client retry storms causing 409 floods.
- Action: `docker-compose down && docker-compose up -d v1.2.0 && monitor`.

# Decision Summary

The duplicate benchmark rows stem from a classic async concurrency race compounded by missing database constraints and streaming lifecycle gaps. The proposed fix adopts a defense-in-depth strategy:

1. **Database Layer**: `UNIQUE(idempotency_key)` + `INSERT OR IGNORE` + WAL mode eliminates TOCTOU races at the storage level. Composite indexes and `CHECK` constraints enforce data integrity and query performance.
2. **Application Layer**: Async locks per `idempotency_key` serialize concurrent creation attempts. Explicit transaction boundaries prevent partial commits. `finish_llm_request` and `add_event` include idempotency guards to prevent double-finalization.
3. **Streaming Layer**: `BackgroundTasks` with status checks, `ClientDisconnect` handling, and proxy retry alignment ensure streaming lifecycles don't leak duplicate metadata.
4. **Testing & Rollback**: 12 acceptance tests cover concurrency, idempotency, streaming, and database integrity. Rollback is versioned, backward-compatible, and feature-flagged for safe home-lab operation.

This approach is engineering-grade: it respects SQLite's single-writer model while leveraging async Python concurrency safely, enforces idempotency at every layer, and maintains observability for future incidents. It is production-ready for a private home-lab benchmark harness, balancing correctness, performance, and operational safety.