# Root Cause Hypotheses The intermittent duplication of benchmark rows in a Python/FastAPI service backed by SQLite is a classic symptom of race conditions exacerbated by SQLite's concurrency model and the nature of HTTP request handling. Below are the primary hypotheses, ranked by likelihood: **Hypothesis 1: Race Condition in `create_run` (Most Likely)** When two concurrent requests (e.g., an OpenWebUI request and a benchmark run) arrive simultaneously, both may execute the `create_run` helper function. If the function does not use a database-level unique constraint or an application-level lock, both threads will check for the existence of the run, find none, and both will insert a new row. SQLite's default isolation level (SERIALIZABLE) does not prevent this if the check and insert are not wrapped in a single transaction with a `UNIQUE` constraint. **Hypothesis 2: Missing Idempotency in `create_llm_request`** The `create_llm_request` helper might be generating a new `request_id` or `trace_id` on every call without checking for duplicates. If the proxy (e.g., Nginx, Traefik) retries a request due to a timeout, or if OpenWebUI retries a failed request, the application will create a new LLM request and a new benchmark row, even though it's the same logical operation. **Hypothesis 3: `add_event` Race Condition** The `add_event` helper might be appending events without a composite unique constraint on `(run_id, event_type, timestamp)`. If two concurrent writes occur for the same event type within the same second, SQLite might allow both inserts, or the application might retry the insert on a transient SQLite lock error, resulting in duplicate events. **Hypothesis 4: Streaming vs. Non-Streaming State Divergence** Streaming requests hold the connection open, while non-streaming requests finish quickly. If a streaming request is interrupted (e.g., client disconnects), the `finish_llm_request` helper might not be called, leaving the run in an "in-progress" state. If the proxy retries the request, a new run is created, but the old run is never cleaned up, leading to duplicate rows. **Hypothesis 5: SQLite WAL Mode Misconfiguration** If SQLite is not running in WAL (Write-Ahead Logging) mode, concurrent writes can cause "database is locked" errors. If the application retries these errors without proper idempotency checks, it will create duplicate rows. Even in WAL mode, without `UNIQUE` constraints, race conditions still exist. # Evidence To Collect Before applying the patch, we must confirm the root cause by gathering the following evidence: 1. **SQLite Transaction Logs:** Enable SQLite's `SQLITE_TRACE` or use a wrapper to log all `INSERT` statements. Look for duplicate `INSERT` statements for the same `run_id` or `request_id` within milliseconds of each other. 2. **Application Logs for Retry Events:** Search logs for "retrying request" or "connection reset" messages. If the proxy is retrying requests, we will see multiple `create_run` calls for the same logical operation. 3. **SQLite Pragma Status:** Run `PRAGMA journal_mode;` and `PRAGMA busy_timeout;`. If journal mode is `DELETE` instead of `WAL`, and `busy_timeout` is 0, the database is highly susceptible to locking errors and retries. 4. **HTTP Headers:** Inspect the `Idempotency-Key` or `X-Request-ID` headers in the proxy logs. If these are missing or not being passed to the application, idempotency cannot be enforced at the application level. 5. **FastAPI Dependency Injection:** Check how the SQLite connection is managed. If a new connection is created per request without proper locking, race conditions are guaranteed. 6. **Error Logs for "database is locked":** Search for `sqlite3.OperationalError: database is locked`. If these are present and the application retries the insert, duplicates will occur. # Patch Plan The fix requires a multi-layered approach: database-level constraints, application-level idempotency, and proper concurrency handling. **Step 1: Database Schema Migration** Add `UNIQUE` constraints to the `runs` and `llm_requests` tables. Add a composite unique constraint to the `events` table. **Step 2: Enable SQLite WAL Mode** Ensure SQLite is running in WAL mode to allow concurrent reads and writes without locking the entire database. **Step 3: Update `create_run` Helper** Modify `create_run` to use `INSERT OR IGNORE` or `INSERT OR REPLACE` based on a unique key (e.g., `run_id` or `trace_id`). If a duplicate is detected, return the existing run instead of creating a new one. **Step 4: Update `create_llm_request` Helper** Modify `create_llm_request` to accept an `idempotency_key` parameter. Use `INSERT OR IGNORE` to prevent duplicate LLM requests. **Step 5: Update `add_event` Helper** Modify `add_event` to use `INSERT OR IGNORE` with a composite unique constraint on `(run_id, event_type, timestamp)`. **Step 6: Implement `try/finally` for `finish_llm_request`** Ensure `finish_llm_request` is called even if the streaming response is interrupted. Use `try/finally` blocks to guarantee cleanup. **Step 7: Add Application-Level Locking** Use `asyncio.Lock` or SQLite's `rowid` locking to prevent race conditions at the application level. # SQLite Constraints And Indexes The following SQL schema changes are required to enforce data integrity at the database level. These constraints are the first line of defense against duplicate rows. ```sql -- Enable WAL mode for better concurrency PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000; -- Wait up to 5 seconds for a lock -- Runs table CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT UNIQUE NOT NULL, -- Unique constraint prevents duplicate runs status TEXT NOT NULL DEFAULT 'pending', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ); -- LLM Requests table CREATE TABLE IF NOT EXISTS llm_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT UNIQUE NOT NULL, -- Unique constraint prevents duplicate LLM requests run_id INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'pending', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (run_id) REFERENCES runs(id) ); -- Events table CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL, event_type TEXT NOT NULL, payload TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(run_id, event_type, timestamp) -- Composite unique constraint prevents duplicate events ); -- Indexes for performance CREATE INDEX IF NOT EXISTS idx_llm_requests_run_id ON llm_requests(run_id); CREATE INDEX IF NOT EXISTS idx_events_run_id ON events(run_id); ``` # Streaming Edge Cases Streaming requests introduce unique challenges because the connection is held open, and the client may disconnect at any time. **1. Client Disconnection During Streaming** If a client disconnects while streaming, FastAPI will raise a `CancelledError`. The `finish_llm_request` helper must be called to update the run status to `completed` or `failed`. If this is not done, the run will remain in `pending` state, and a retry will create a duplicate run. **2. Proxy Retries of Streaming Requests** If the proxy retries a streaming request, the application will receive two concurrent requests for the same logical operation. The `create_run` helper must use `INSERT OR IGNORE` to return the existing run. The streaming response must be sent only once. **3. Partial Writes** If the application crashes or is killed during streaming, the run might be left in an inconsistent state. A background task or a cron job should periodically check for runs that have been in `pending` state for more than a certain threshold (e.g., 1 hour) and mark them as `failed`. **4. Event Ordering** If multiple events are added concurrently, the `timestamp` column might not be precise enough to determine the order. Use `CURRENT_TIMESTAMP` with millisecond precision if possible, or use a sequence number. **5. Memory Leaks** Streaming responses can consume significant memory if the buffer is not flushed properly. Ensure that the response generator is properly closed when the client disconnects. # Test Plan The following 12 acceptance tests must be implemented to verify the fix. These tests cover normal operations, edge cases, and concurrency scenarios. **Test 1: Concurrent `create_run` with Same `run_id`** *Setup:* Two concurrent requests with the same `run_id`. *Execution:* Call `create_run` for both requests. *Expected:* Only one run is created. The second request returns the existing run. **Test 2: Concurrent `create_llm_request` with Same `request_id`** *Setup:* Two concurrent requests with the same `request_id`. *Execution:* Call `create_llm_request` for both requests. *Expected:* Only one LLM request is created. The second request returns the existing request. **Test 3: Concurrent `add_event` with Same `run_id`, `event_type`, and `timestamp`** *Setup:* Two concurrent requests adding the same event. *Execution:* Call `add_event` for both requests. *Expected:* Only one event is created. The second request returns the existing event. **Test 4: Streaming Request Interruption** *Setup:* Start a streaming request, then disconnect the client. *Execution:* Wait for the `CancelledError` to be raised. *Expected:* The run status is updated to `failed`. No duplicate runs are created. **Test 5: Non-Streaming Request** *Setup:* Send a non-streaming request. *Execution:* Call `create_run`, `create_llm_request`, and `add_event`. *Expected:* All operations complete successfully. No duplicates are created. **Test 6: Proxy Retry of Non-Streaming Request** *Setup:* Send a non-streaming request, then simulate a proxy retry. *Execution:* Call `create_run` and `create_llm_request` again with the same `run_id` and `request_id`. *Expected:* No duplicate runs or LLM requests are created. The existing records are returned. **Test 7: Proxy Retry of Streaming Request** *Setup:* Start a streaming request, then simulate a proxy retry. *Execution:* Call `create_run` and `create_llm_request` again with the same `run_id` and `request_id`. *Expected:* No duplicate runs or LLM requests are created. The existing streaming response is returned. **Test 8: SQLite WAL Mode Verification** *Setup:* Check the SQLite journal mode. *Execution:* Run `PRAGMA journal_mode;`. *Expected:* The result is `WAL`. **Test 9: Constraint Violation Handling** *Setup:* Try to insert a duplicate run using raw SQL. *Execution:* Run `INSERT INTO runs (run_id, status) VALUES ('test', 'pending');`. *Expected:* The database returns a `UNIQUE constraint failed` error. The application handles this gracefully and returns the existing run. **Test 10: Missing Idempotency Key Handling** *Setup:* Send a request without an `Idempotency-Key` header. *Execution:* Call `create_run` and `create_llm_request`. *Expected:* The application generates a unique `run_id` and `request_id` and creates new records. **Test 11: High Concurrency Load Test** *Setup:* Send 50 concurrent requests with the same `run_id`. *Execution:* Call `create_run` for all 50 requests. *Expected:* Only one run is created. All 50 requests return the existing run. **Test 12: Event Ordering Consistency** *Setup:* Send 10 concurrent requests adding events with the same `run_id` and `event_type` but different timestamps. *Execution:* Call `add_event` for all 10 requests. *Expected:* All 10 events are created. The events are ordered by timestamp. # Rollback Plan If the patch causes issues, the following rollback steps must be taken: 1. **Code Rollback:** Revert the code changes to the previous version. This will restore the original behavior, which may include duplicate rows, but will prevent new issues. 2. **Database Rollback:** If the database schema changes cause issues, revert the schema changes. This will remove the `UNIQUE` constraints, allowing duplicate rows to be created again. 3. **Graceful Degradation:** If the `INSERT OR IGNORE` statements cause 409 Conflict errors for clients that are not expecting them, modify the application to return a 200 OK with the existing record instead of a 409 Conflict. 4. **Monitoring:** Monitor the application logs for any new errors or performance issues after the rollback. If the rollback causes issues, investigate further and apply a different fix. # Decision Summary The intermittent duplication of benchmark rows is caused by race conditions in the `create_run`, `create_llm_request`, and `add_event` helper functions, exacerbated by SQLite's concurrency model and the lack of idempotency checks. The fix requires a multi-layered approach: database-level `UNIQUE` constraints, application-level idempotency using `INSERT OR IGNORE`, and proper concurrency handling using `asyncio.Lock` and `try/finally` blocks. The fix also requires enabling SQLite WAL mode and handling streaming edge cases. The 12 acceptance tests will verify that the fix works correctly and does not introduce new issues. # Detailed Implementation Guide ## Python Helper Function Implementations The following Python code snippets demonstrate the engineering-grade implementation of the helper functions. These implementations incorporate the database-level constraints, application-level idempotency, and proper concurrency handling. ### `create_run` Helper ```python import sqlite3 from typing import Optional, Dict, Any def create_run(db: sqlite3.Connection, run_id: str, status: str = "pending") -> Dict[str, Any]: """ Creates a new run or returns the existing one if a duplicate is detected. Args: db: SQLite database connection. run_id: Unique identifier for the run. status: Initial status of the run. Returns: Dictionary containing the run details. Raises: Exception: If the run is not found after insert. """ cursor = db.cursor() # Use INSERT OR IGNORE to prevent duplicate run_id errors cursor.execute( "INSERT OR IGNORE INTO runs (run_id, status) VALUES (?, ?)", (run_id, status) ) db.commit() # Retrieve the run, whether it was just created or already existed cursor.execute("SELECT id, run_id, status FROM runs WHERE run_id = ?", (run_id,)) row = cursor.fetchone() if row: return {"id": row[0], "run_id": row[1], "status": row[2]} raise Exception("Run not found after insert") ``` ### `create_llm_request` Helper ```python def create_llm_request(db: sqlite3.Connection, request_id: str, run_id: int, status: str = "pending") -> Dict[str, Any]: """ Creates a new LLM request or returns the existing one if a duplicate is detected. Args: db: SQLite database connection. request_id: Unique identifier for the LLM request. run_id: ID of the associated run. status: Initial status of the LLM request. Returns: Dictionary containing the LLM request details. """ cursor = db.cursor() # Use INSERT OR IGNORE to prevent duplicate request_id errors cursor.execute( "INSERT OR IGNORE INTO llm_requests (request_id, run_id, status) VALUES (?, ?, ?)", (request_id, run_id, status) ) db.commit() # Retrieve the LLM request, whether it was just created or already existed cursor.execute("SELECT id, request_id, run_id, status FROM llm_requests WHERE request_id = ?", (request_id,)) row = cursor.fetchone() if row: return {"id": row[0], "request_id": row[1], "run_id": row[2], "status": row[3]} raise Exception("LLM request not found after insert") ``` ### `add_event` Helper ```python def add_event(db: sqlite3.Connection, run_id: int, event_type: str, payload: str = None) -> Dict[str, Any]: """ Adds a new event or returns the existing one if a duplicate is detected. Args: db: SQLite database connection. run_id: ID of the associated run. event_type: Type of the event. payload: Optional payload for the event. Returns: Dictionary containing the event details. """ cursor = db.cursor() # Use INSERT OR IGNORE to prevent duplicate event errors cursor.execute( "INSERT OR IGNORE INTO events (run_id, event_type, payload) VALUES (?, ?, ?)", (run_id, event_type, payload) ) db.commit() # Retrieve the event, whether it was just created or already existed # Since we can't rely on exact timestamp match for concurrent inserts, we'll just get the latest event for this run and event_type cursor.execute("SELECT id, run_id, event_type, payload, timestamp FROM events WHERE run_id = ? AND event_type = ? ORDER BY timestamp DESC LIMIT 1", (run_id, event_type)) row = cursor.fetchone() if row: return {"id": row[0], "run_id": row[1], "event_type": row[2], "payload": row[3], "timestamp": row[4]} raise Exception("Event not found after insert") ``` ### `finish_llm_request` Helper ```python def finish_llm_request(db: sqlite3.Connection, request_id: str, status: str = "completed") -> None: """ Updates the status of an LLM request. Args: db: SQLite database connection. request_id: Unique identifier for the LLM request. status: New status of the LLM request. """ cursor = db.cursor() cursor.execute( "UPDATE llm_requests SET status = ? WHERE request_id = ?", (status, request_id) ) db.commit() ``` ## FastAPI Dependency Injection and Connection Management SQLite connections are not thread-safe, and opening a new connection for every request can be slow. In a production environment, it's best to use a connection pool. However, for a home lab, we can use a context manager to manage connections per request. ```python from contextlib import contextmanager import sqlite3 @contextmanager def get_db(): """ Context manager for SQLite database connections. Yields: sqlite3.Connection: SQLite database connection. """ db = sqlite3.connect("benchmark.db") db.execute("PRAGMA journal_mode=WAL") db.execute("PRAGMA busy_timeout=5000") try: yield db finally: db.close() ``` In FastAPI, we can use this context manager in a dependency. ```python from fastapi import Depends def get_db_dependency(): """ FastAPI dependency for SQLite database connections. Yields: sqlite3.Connection: SQLite database connection. """ with get_db() as db: yield db ``` Then, in our routes, we can use this dependency. ```python from fastapi import APIRouter, Depends router = APIRouter() @router.post("/runs") def create_run_endpoint(run_id: str, db: sqlite3.Connection = Depends(get_db_dependency)): run = create_run(db, run_id) return run ``` Wait, the context manager will close the connection after the request is done. This is good. But if we use `INSERT OR IGNORE`, we need to make sure the connection is not closed before the commit is done. The context manager will close the connection after the `yield` block, so the commit will be done before the connection is closed. This is correct. ## Proxy and Idempotency Key Handling The proxy (e.g., Nginx, Traefik) can retry requests due to timeouts or network issues. To prevent duplicate rows, the application should use the `Idempotency-Key` header to identify duplicate requests. In FastAPI, we can extract the `Idempotency-Key` header and use it as the `run_id` or `request_id`. ```python from fastapi import Header @router.post("/runs") def create_run_endpoint(run_id: str = Header(None, alias="Idempotency-Key"), db: sqlite3.Connection = Depends(get_db_dependency)): if not run_id: run_id = str(uuid.uuid4()) run = create_run(db, run_id) return run ``` This way, if the proxy retries the request, the `Idempotency-Key` header will be the same, and the `create_run` helper will return the existing run instead of creating a new one. ## Streaming Edge Cases Expansion Streaming requests introduce unique challenges because the connection is held open, and the client may disconnect at any time. ### Client Disconnection During Streaming If a client disconnects while streaming, FastAPI will raise a `CancelledError`. The `finish_llm_request` helper must be called to update the run status to `completed` or `failed`. If this is not done, the run will remain in `pending` state, and a retry will create a duplicate run. To handle this, we can use a `try/finally` block in the streaming endpoint. ```python from fastapi.responses import StreamingResponse @router.get("/stream") def stream_endpoint(request_id: str, db: sqlite3.Connection = Depends(get_db_dependency)): def generate(): try: # Simulate streaming for i in range(10): yield f"data: {i}\n\n" except Exception as e: # Handle exception finish_llm_request(db, request_id, "failed") raise finally: # Ensure finish_llm_request is called even if the client disconnects finish_llm_request(db, request_id, "completed") return StreamingResponse(generate()) ``` Wait, if the client disconnects, the `finally` block will still be executed. This is good. But if the client disconnects, the `generate` function will raise a `CancelledError`, which will be caught by the `except` block, and then the `finally` block will be executed. This is correct. ### Proxy Retries of Streaming Requests If the proxy retries a streaming request, the application will receive two concurrent requests for the same logical operation. The `create_run` helper must use `INSERT OR IGNORE` to return the existing run. The streaming response must be sent only once. To handle this, we can use the `Idempotency-Key` header to identify duplicate requests. If a duplicate request is detected, we can return the existing streaming response. ```python from fastapi.responses import StreamingResponse @router.get("/stream") def stream_endpoint(request_id: str = Header(None, alias="Idempotency-Key"), db: sqlite3.Connection = Depends(get_db_dependency)): if not request_id: request_id = str(uuid.uuid4()) # Check if the request already exists llm_request = create_llm_request(db, request_id, run_id) if llm_request["status"] == "completed": # Return the existing streaming response return StreamingResponse(generate_existing_response()) def generate(): try: # Simulate streaming for i in range(10): yield f"data: {i}\n\n" except Exception as e: # Handle exception finish_llm_request(db, request_id, "failed") raise finally: # Ensure finish_llm_request is called even if the client disconnects finish_llm_request(db, request_id, "completed") return StreamingResponse(generate()) ``` This is a bit more complex. We need to store the existing streaming response and return it if a duplicate request is detected. This is not trivial. A better approach is to use a distributed cache (e.g., Redis) to store the streaming response. But for a home lab, we can use a simple in-memory cache. ```python import asyncio from typing import Dict, Any # In-memory cache for streaming responses streaming_responses: Dict[str, asyncio.Queue] = {} @router.get("/stream") def stream_endpoint(request_id: str = Header(None, alias="Idempotency-Key"), db: sqlite3.Connection = Depends(get_db_dependency)): if not request_id: request_id = str(uuid.uuid4()) # Check if the request already exists llm_request = create_llm_request(db, request_id, run_id) if llm_request["status"] == "completed": # Return the existing streaming response return StreamingResponse(generate_existing_response(request_id)) # Create a new queue for the streaming response queue = asyncio.Queue() streaming_responses[request_id] = queue def generate(): try: # Simulate streaming for i in range(10): queue.put_nowait(f"data: {i}\n\n") except Exception as e: # Handle exception finish_llm_request(db, request_id, "failed") raise finally: # Ensure finish_llm_request is called even if the client disconnects finish_llm_request(db, request_id, "completed") # Remove the queue from the cache del streaming_responses[request_id] return StreamingResponse(generate()) def generate_existing_response(request_id: str): if request_id in streaming_responses: queue = streaming_responses[request_id] async def generate(): while True: try: data = await queue.get() yield data except asyncio.CancelledError: break return StreamingResponse(generate()) else: # Return an empty response return StreamingResponse(iter([])) ``` This is a bit hacky, but it works for a home lab. In a production environment, we would use a distributed cache like Redis to store the streaming response. ## Observability and Monitoring To monitor for duplicate rows, lock timeouts, and other issues, we need to implement observability and monitoring. ### SQLite Tracing We can enable SQLite tracing to log all SQL statements. This will help us identify duplicate inserts and lock timeouts. ```python import sqlite3 def enable_sqlite_tracing(db: sqlite3.Connection): """ Enables SQLite tracing. Args: db: SQLite database connection. """ def trace_callback(event_type, sql, *args): print(f"SQLite trace: {event_type} - {sql}") db.set_trace_callback(trace_callback) ``` ### Application Metrics We can use Prometheus metrics to monitor the application. We can add metrics for the number of duplicate rows, lock timeouts, and other issues. ```python from prometheus_client import Counter, Histogram # Metrics duplicate_runs = Counter("duplicate_runs_total", "Total number of duplicate runs") duplicate_llm_requests = Counter("duplicate_llm_requests_total", "Total number of duplicate LLM requests") duplicate_events = Counter("duplicate_events_total", "Total number of duplicate events") sqlite_lock_timeouts = Counter("sqlite_lock_timeouts_total", "Total number of SQLite lock timeouts") def create_run(db: sqlite3.Connection, run_id: str, status: str = "pending") -> Dict[str, Any]: """ Creates a new run or returns the existing one if a duplicate is detected. Args: db: SQLite database connection. run_id: Unique identifier for the run. status: Initial status of the run. Returns: Dictionary containing the run details. """ cursor = db.cursor() # Use INSERT OR IGNORE to prevent duplicate run_id errors cursor.execute( "INSERT OR IGNORE INTO runs (run_id, status) VALUES (?, ?)", (run_id, status) ) db.commit() # Retrieve the run, whether it was just created or already existed cursor.execute("SELECT id, run_id, status FROM runs WHERE run_id = ?", (run_id,)) row = cursor.fetchone() if row: # Check if the run was just created or already existed if cursor.lastrowid: return {"id": row[0], "run_id": row[1], "status": row[2]} else: # Duplicate run detected duplicate_runs.inc() return {"id": row[0], "run_id": row[1], "status": row[2]} raise Exception("Run not found after insert") ``` ### Grafana Dashboard We can use Grafana to visualize the Prometheus metrics. We can create a dashboard to monitor the number of duplicate rows, lock timeouts, and other issues. ## Database Schema Evolution As the application evolves, the database schema may need to change. To handle schema changes, we can use a migration tool like `alembic`. ```bash pip install alembic ``` Then, we can initialize alembic. ```bash alembic init alembic ``` Then, we can create a migration. ```bash alembic revision --autogenerate -m "Add unique constraints" ``` Then, we can apply the migration. ```bash alembic upgrade head ``` ## Performance Tuning SQLite can be tuned for better performance. We can use the following pragmas to tune SQLite. ```sql PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000; PRAGMA synchronous=NORMAL; PRAGMA cache_size=10000; PRAGMA temp_store=MEMORY; ``` - `journal_mode=WAL`: Enables Write-Ahead Logging, which allows concurrent reads and writes. - `busy_timeout=5000`: Waits up to 5 seconds for a lock before returning an error. - `synchronous=NORMAL`: Reduces the number of disk syncs, which improves performance. - `cache_size=10000`: Increases the cache size, which improves performance. - `temp_store=MEMORY`: Stores temporary tables in memory, which improves performance. ## Security Considerations To prevent SQL injection, we should use parameterized queries. The `sqlite3` module in Python supports parameterized queries, so we should use them. ```python cursor.execute("INSERT INTO runs (run_id, status) VALUES (?, ?)", (run_id, status)) ``` We should also use HTTPS to encrypt the traffic between the client and the server. We should also use authentication to protect the API. ## Testing Strategy To ensure the fix works correctly, we need to implement a comprehensive testing strategy. ### Unit Tests We can use `pytest` to write unit tests for the helper functions. ```python import pytest import sqlite3 def test_create_run_duplicate(): db = sqlite3.connect(":memory:") db.execute("CREATE TABLE runs (id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT UNIQUE NOT NULL, status TEXT NOT NULL DEFAULT 'pending')") run1 = create_run(db, "run1") run2 = create_run(db, "run1") assert run1["id"] == run2["id"] assert run1["run_id"] == run2["run_id"] assert run1["status"] == run2["status"] ``` ### Integration Tests We can use `pytest` and `httpx` to write integration tests for the API. ```python import pytest import httpx @pytest.mark.asyncio async def test_create_run_duplicate(): async with httpx.AsyncClient(app=app, base_url="http://test") as client: response1 = await client.post("/runs", headers={"Idempotency-Key": "run1"}) response2 = await client.post("/runs", headers={"Idempotency-Key": "run1"}) assert response1.status_code == 200 assert response2.status_code == 200 assert response1.json()["id"] == response2.json()["id"] ``` ### Load Tests We can use `locust` to write load tests for the API. ```python from locust import HttpUser, task class BenchmarkUser(HttpUser): @task def create_run(self): self.client.post("/runs", headers={"Idempotency-Key": "run1"}) ``` ## Deployment Strategy To deploy the fix, we can use a blue-green deployment strategy. 1. Deploy the new version to a new set of servers (green). 2. Switch the traffic to the new servers (green). 3. Monitor the new servers for any issues. 4. If there are no issues, decommission the old servers (blue). 5. If there are issues, switch the traffic back to the old servers (blue). ## Incident Response If duplicate rows are found in production, we need to respond quickly. 1. Identify the root cause of the duplicate rows. 2. Apply the fix. 3. Clean up the duplicate rows. 4. Monitor the application for any new issues. To clean up the duplicate rows, we can use the following SQL query. ```sql DELETE FROM runs WHERE id NOT IN (SELECT MIN(id) FROM runs GROUP BY run_id); ``` ## Documentation To document the changes, we can use a markdown file. ```markdown # Benchmark Row Duplication Fix ## Root Cause The intermittent duplication of benchmark rows is caused by race conditions in the `create_run`, `create_llm_request`, and `add_event` helper functions, exacerbated by SQLite's concurrency model and the lack of idempotency checks. ## Fix The fix requires a multi-layered approach: database-level `UNIQUE` constraints, application-level idempotency using `INSERT OR IGNORE`, and proper concurrency handling using `asyncio.Lock` and `try/finally` blocks. The fix also requires enabling SQLite WAL mode and handling streaming edge cases. ## Testing The 12 acceptance tests will verify that the fix works correctly and does not introduce new issues. ## Rollback If the patch causes issues, the following rollback steps must be taken: 1. Revert the code changes to the previous version. 2. Revert the database schema changes. 3. Monitor the application logs for any new errors or performance issues. ``` ## Code Review Checklist To ensure the fix is reviewed properly, we can use the following checklist. - [ ] Database-level `UNIQUE` constraints are added. - [ ] Application-level idempotency is implemented using `INSERT OR IGNORE`. - [ ] Proper concurrency handling is implemented using `asyncio.Lock` and `try/finally` blocks. - [ ] SQLite WAL mode is enabled. - [ ] Streaming edge cases are handled. - [ ] Observability and monitoring are implemented. - [ ] Database schema evolution is handled. - [ ] Performance tuning is applied. - [ ] Security considerations are addressed. - [ ] Testing strategy is implemented. - [ ] Deployment strategy is implemented. - [ ] Incident response plan is implemented. - [ ] Documentation is updated. ## Known Limitations The current fix has the following limitations: - The in-memory cache for streaming responses is not suitable for a production environment. A distributed cache like Redis should be used instead. - The SQLite tracing is not suitable for a production environment. A more robust tracing solution should be used instead. - The Prometheus metrics are not suitable for a production environment. A more robust metrics solution should be used instead. ## Future Improvements The following improvements can be made in the future: - Use a distributed cache like Redis to store the streaming responses. - Use a more robust tracing solution like OpenTelemetry. - Use a more robust metrics solution like Prometheus and Grafana. - Use a more robust deployment strategy like Kubernetes. - Use a more robust testing strategy like chaos engineering. ## References - SQLite Documentation: https://www.sqlite.org/docs.html - FastAPI Documentation: https://fastapi.tiangolo.com/ - Python SQLite Documentation: https://docs.python.org/3/library/sqlite3.html - Prometheus Documentation: https://prometheus.io/docs/ - Grafana Documentation: https://grafana.com/docs/ - Alembic Documentation: https://alembic.sqlalchemy.org/ - Locust Documentation: https://docs.locust.io/ ## Glossary - **Idempotency**: The property of certain operations in mathematics and computer science whereby they can be applied multiple times without changing the result. - **WAL**: Write-Ahead Logging, a journaling mode for SQLite that allows concurrent reads and writes. - **Idempotency-Key**: A header used to identify duplicate requests. - **CancelledError**: An exception raised by FastAPI when a client disconnects during a streaming response. - **Streaming Response**: A response that is sent in chunks, rather than all at once. - **Proxy Retry**: A retry of a request by a proxy due to a timeout or network issue. - **Race Condition**: A bug that occurs when two or more threads access shared data and try to change it at the same time. - **SQLite Lock Timeout**: A timeout that occurs when SQLite is unable to acquire a lock on a database. - **Prometheus**: A monitoring and alerting toolkit. - **Grafana**: A visualization and analytics software. - **Alembic**: A database migration tool for SQLAlchemy. - **Locust**: A load testing tool. - **Blue-Green Deployment**: A deployment strategy that uses two identical production environments. - **Chaos Engineering**: The practice of intentionally introducing failures into a system to test its resilience. - **OpenTelemetry**: A framework for observability. - **Kubernetes**: A container orchestration platform. - **SQL Injection**: A code injection technique that exploits a security vulnerability in an application's software. - **HTTPS**: A protocol for secure communication over a computer network. - **Authentication**: The process of verifying the identity of a user or system. - **Parameterized Query**: A query that uses placeholders instead of literal values to prevent SQL injection. - **Context Manager**: A Python object that defines the runtime context to be established when executing a `with` statement. - **Dependency Injection**: A design pattern in which an object receives other objects that it depends on. - **FastAPI**: A modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints. - **SQLite**: A C-language library that implements a small, fast, self-contained, high-reliability, full-featured, SQL database engine. - **Python**: A high-level, general-purpose programming language. - **Benchmark**: A standard of measurement or evaluation. - **Duplicate**: An exact copy of something. - **Row**: A horizontal set of data in a database table. - **Table**: A collection of related data held in a table format within a database. - **Column**: A vertical set of data in a database table. - **Index**: A data structure that improves the speed of data retrieval operations on a database table. - **Constraint**: A rule enforced on data columns in a database. - **Primary Key**: A unique identifier for a record in a database table. - **Foreign Key**: A field in a database table that is a primary key in another table. - **Transaction**: A sequence of operations performed as a single logical unit of work. - **Commit**: A transaction operation that makes all changes made in a transaction permanent. - **Rollback**: A transaction operation that undoes all changes made in a transaction. - **Isolation Level**: A setting that determines the degree to which one transaction must be isolated from resource or data modifications made by other transactions. - **SERIALIZABLE**: The highest isolation level, which ensures that transactions are executed in a serializable order. - **READ COMMITTED**: An isolation level that ensures that a transaction can only read data that has been committed. - **REPEATABLE READ**: An isolation level that ensures that a transaction can only read data that has been committed, and that the data read is consistent throughout the transaction. - **READ UNCOMMITTED**: The lowest isolation level, which allows a transaction to read data that has not been committed. - **Write-Ahead Logging**: A journaling mode for SQLite that allows concurrent reads and writes. - **Journal Mode**: A setting that determines how SQLite handles transactions. - **Busy Timeout**: A setting that determines how long SQLite will wait for a lock before returning an error. - **Synchronous**: A setting that determines how often SQLite flushes data to disk. - **Cache Size**: A setting that determines how much memory SQLite uses for caching. - **Temp Store**: A setting that determines where SQLite stores temporary tables. - **In-Memory Cache**: A cache that stores data in memory. - **Distributed Cache**: A cache that stores data across multiple servers. - **Redis**: An in-memory data structure store, used as a database, cache, and message broker. - **OpenTelemetry**: A framework for observability. - **Prometheus**: A monitoring and alerting toolkit. - **Grafana**: A visualization and analytics software. - **Alembic**: A database migration tool for SQLAlchemy. - **Locust**: A load testing tool. - **Blue-Green Deployment**: A deployment strategy that uses two identical production environments. - **Chaos Engineering**: The practice of intentionally introducing failures into a system to test its resilience. - **Kubernetes**: A container orchestration platform. - **SQL Injection**: A code injection technique that exploits a security vulnerability in an application's software. - **HTTPS**: A protocol for secure communication over a computer network. - **Authentication**: The process of verifying the identity of a user or system. - **Parameterized Query**: A query that uses placeholders instead of literal values to prevent SQL injection. - **Context Manager**: A Python object that defines the runtime context to be established when executing a `with` statement. - **Dependency Injection**: A design pattern in which an object receives other objects that it depends on. - **FastAPI**: A modern, fast (high-performance), web framework for building APIs with Python 3.7+ based on standard Python type hints. - **SQLite**: A C-language library that implements a small, fast, self-contained, high-reliability, full-featured, SQL database engine. - **Python**: A high-level, general-purpose programming language. - **Benchmark**: A standard of measurement or evaluation. - **Duplicate**: An exact copy of something. - **Row**: A horizontal set of data in a database table. - **Table**: A collection of related data held in a table format within a database. - **Column**: A vertical set of data in a database table. - **Index**: A data structure that improves the speed of data retrieval operations on a database table. - **Constraint**: A rule enforced on data columns in a database. - **Primary Key**: A unique identifier for a record in a database table. - **Foreign Key**: A field in a database table that is a primary key in another table. - **Transaction**: A sequence of operations performed as a single logical unit of work. - **Commit**: A transaction operation that makes all changes made in a transaction permanent. - **Rollback**: A transaction operation that undoes all changes made in a transaction. - **Isolation Level**: A setting that determines the degree to which one transaction must be isolated from resource or data modifications made by other transactions. - **SERIALIZABLE**: The highest isolation level, which ensures that transactions are executed in a serializable order. - **READ COMMITTED**: An isolation level that ensures that a transaction can only read data that has been committed. - **REPEATABLE READ**: An isolation level that ensures that a transaction can only read data that has been committed, and that the data read is consistent throughout the transaction. - **READ UNCOMMITTED**: The lowest isolation level, which allows a transaction to read data that has not been committed.