Build a Ride Dispatch Engine

Introduction

30 Second Summary

When a concert ends, a wave of riders can compete for a limited pool of nearby drivers. Every match must stay fast without assigning the same driver twice.

In this project, you will build a deterministic backend experiment for that concert-ending rush. This educational simulation uses synthetic locations to reproduce a duplicate-assignment race before enforcing one active ride per driver with an atomic claim.

What You'll Build

You'll watch one concert-spike workload expose duplicate rides on the unsafe path before the atomic path preserves one active ride per driver.

By the end of this project, you'll have:

• A queryable FastAPI simulator that returns deterministic ride demand alongside dispatch outcomes.
• A terminal preview where H3 candidate search reveals when the closest driver loses to the ETA-aware winner. The same preview shows surge pricing bounded from 1.0x to 3.0x.
• A repeatable concurrency race that creates an actual duplicate active assignment. An atomic Redis Lua claim then protects the winning ride. A measured Markdown report captures the before-and-after evidence.
• Secret Mission: Process 500 deterministic ride requests with two competing workers to prove zero duplicate active assignments on the atomic path.

Are there any prerequisites?

You'll need the provided starter repository plus Visual Studio Code and Docker Desktop.

Docker Desktop runs the Python and Redis services locally at no cost.

Before We Start

Before the hands-on work begins, commit to the purpose and boundary of this educational local dispatch experiment. Your reliability claim rests on one correctness invariant: each driver can have at most one active ride assignment.

Launch the Dispatch Lab

Concurrency evidence is trustworthy only when every run starts from the same local environment. Docker Compose builds the pinned Python service. Its health dependency delays FastAPI until Redis is ready.

Your starter repository already contains the lab. You will inspect it in Visual Studio Code before proving the stack through its health endpoint plus smoke test.

In this step, get ready to:

• Open the starter repository in Visual Studio Code.
• Start the local API plus Redis services with Docker Compose.
• Prove FastAPI reaches Redis through the health check.

Inspect the starter repository

The starter files lock the runtime versions plus service relationships used by every experiment. Inspecting them gives you a concrete baseline before the containers start.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type Visual Studio Code.
• Press Enter.
• Click File in the top menu.
• Select Open Folder.
• Choose the provided starter repository folder.
• Click Open.

You will see the repository files in the Explorer sidebar.

• Select compose.yaml in the Explorer sidebar.

The file defines the api plus redis services. The Redis image is pinned to redis:8.8.4-alpine.

Why does the API wait for Redis?

The Redis service runs a health check every 2s. The service_healthy condition prevents the API from starting before that check succeeds.

This startup order removes timing differences from later dispatch experiments.

• Select requirements.txt in the Explorer sidebar.

What is pinned?

Exact dependency pins keep each container build reproducible.

• FastAPI is pinned as fastapi==0.162.0.
• Uvicorn is pinned as uvicorn==0.54.0.
• redis-py is pinned as redis==8.1.0.
• h3-py is pinned as h3==4.5.0.
• pytest is pinned as pytest==9.2.0.
• HTTPX is pinned as httpx==0.28.1.

• Select Dockerfile in the Explorer sidebar.

The API image starts from python:3.13.18. It installs the pinned packages before launching Uvicorn on port 8000.

• Select app/main.py in the Explorer sidebar.

The starter API exposes /health. That endpoint calls store.ping() to check the Redis connection.

• Select tests/test_smoke.py in the Explorer sidebar.

The smoke test checks the HTTP status plus the Redis result returned by the health endpoint.

• Compare the starter files with the target artifacts using the tabs below:

✔️ The starter files match

Your repository contains the required runtime image, service configuration, API route, dependency pins, plus smoke test.

ⓧ I'd like to double check the full code

FROM python:3.13.18
WORKDIR /workspace
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

How the API image is built

This image installs the pinned dependencies inside /workspace. Its default command starts the FastAPI application through Uvicorn.

Why this file exists

The empty app/__init__.py file identifies app as the application package.

from fastapi import FastAPI

from app.store import RedisAssignmentStore

app = FastAPI(title="Ride Dispatch Reliability Lab")
store = RedisAssignmentStore()

@app.get("/health")
def health() -> dict:
    return {"status": "ok", "redis": store.ping()}

How the health check works

The route returns the API status plus the result of a live Redis ping. A true Redis value proves the application can reach its assignment store.

services:
  redis:
    image: redis:8.8.4-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 2s
      timeout: 2s
      retries: 10
    volumes:
      - redis-data:/data
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000
    depends_on:
      redis:
        condition: service_healthy
    ports:
      - "8000:8000"
    volumes:
      - .:/workspace
volumes:
  redis-data:

How startup order works

Compose starts Redis with persistent local data. It starts the API after the Redis health check passes.

Port 8000 exposes the API to your browser.

fastapi==0.162.0
uvicorn==0.54.0
redis==8.1.0
h3==4.5.0
pytest==9.2.0
httpx==0.28.1

Why exact pins matter

These pins give every API build the same framework, server, Redis client, geographic library, test runner, plus HTTP client versions.

from fastapi.testclient import TestClient
from app.main import app

def test_health_endpoint_reaches_redis() -> None:
    with TestClient(app) as client:
        response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok", "redis": True}

What the smoke test proves

The test calls /health through FastAPI's test client. Its assertions require a successful response plus a working Redis connection.

Start the local services

Docker Compose sends the service plan to the engine managed by Docker Desktop. The engine must be ready before it can build the API image.

Docker Desktop can take about a minute to finish starting. Give it that time before moving back to Visual Studio Code.

• Press Cmd+Space on macOS or the Windows key on Windows to open system search.
• Type Docker Desktop.
• Press Enter.
• Wait until Docker Desktop reports that its engine is running.
• Return to Visual Studio Code.
• Click Terminal in the top menu.
• Select New Terminal.
• Build plus start the local services by running this command:

docker compose up --build -d

What does this command do?

The --build flag rebuilds the API image from the pinned Dockerfile. The -d flag leaves the services running in the background.

Compose waits for the Redis health check before it starts the API service. That ordering gives later tests a stable assignment store.

You will see Redis become healthy before the API finishes starting. The local stack is now running in the background.

Services did not start?

Confirm Docker Desktop has finished starting. Check that the terminal is attached to the starter repository folder.

If the build stops while downloading an image or package, confirm your internet connection before running the same command again.

Help me diagnose the Docker Compose startup failure.

Prove FastAPI can reach Redis

A running container only proves that a process started. The health endpoint verifies that FastAPI can make a live request to Redis.

• Open the local health endpoint by selecting http://localhost:8000/health.

You will see status set to ok. You will also see redis: true.

Health page unavailable?

Confirm the address uses port 8000. Return to the Compose terminal output to check whether both services finished starting.

Help me diagnose why the local health endpoint is unavailable.

Before you run it, predict whether the smoke test will produce one passing result.

• Run the smoke test inside the API container with this command:

docker compose exec api pytest -q tests/test_smoke.py

What does this prove?

The test client requests /health inside the container. It requires HTTP status 200 plus a true Redis result.

One passing test proves the FastAPI process can reach the healthy Redis service.

You will see one passing test. That result establishes the reproducible baseline for every dispatch experiment that follows.

Smoke test did not pass?

Confirm tests/test_smoke.py matches the reference code. Check that the browser health response still reports a true Redis value.

Help me debug the failing smoke test.

Solid work: your verified local stack is ready. Next, you will see how geography changes dispatch decisions.

Partition the City and Rank Drivers

Your Docker Compose lab now has FastAPI reaching Redis. That healthy foundation gives every dispatch experiment the same local services.

Straight-line proximity can select a nearby driver whose traffic factor produces a slower ETA. You will use H3 cells to narrow the candidate pool before comparing distance-based ranking with traffic-adjusted arrival time.

In this step, get ready to:

• Create deterministic marketplace models with Tampa-area driver and rider fixtures.
• Partition coordinates into nearby H3 cells before ranking candidate drivers.
• Compare nearest-distance selection with traffic-adjusted ETA scoring and bounded surge pricing.

Define the deterministic marketplace data

Deterministic data gives each run the same drivers and paired rider requests. This makes every ranking result reproducible without live traffic or changing locations.

• Create app/models.py inside the existing app folder by using the file sidebar in VS Code.
• Add the driver, rider, and claim models by pasting this first chunk into app/models.py:

from dataclasses import dataclass

@dataclass(frozen=True)
class Driver:
    driver_id: str
    lat: float
    lng: float

@dataclass(frozen=True)
class RiderRequest:
    ride_id: str
    lat: float
    lng: float

@dataclass(frozen=True)
class ClaimResult:
    success: bool
    driver_id: str
    ride_id: str
    existing_ride_id: str | None = None

What Do These Models Represent?

• The frozen dataclasses keep fixture values unchanged after creation.
• The Driver coordinates become inputs for geographic partitioning.
• The RiderRequest coordinates identify each dispatch origin.
• The ClaimResult fields record the outcome of a later assignment attempt.

• Save app/models.py.
• Confirm that VS Code shows the three frozen dataclasses without red syntax markers.

Seeing Syntax Markers in the Models?

Check that each decorator sits directly above its class. Keep every field indented beneath the class that owns it.

Use this if the structure still looks wrong: Help me compare my first three dataclasses with the required model definitions.

Dispatch results preserve the selected driver plus the candidates that were attempted. Batch metrics keep assignment outcomes separate from contention and correctness failures.

• Add the remaining result models below existing_ride_id: str | None = None by pasting this chunk:

@dataclass(frozen=True)
class DispatchResult:
    ride_id: str
    driver_id: str | None
    rejected_claims: int
    attempted_driver_ids: tuple[str, ...]
    latency_ms: float

@dataclass(frozen=True)
class BatchMetrics:
    mode: str
    requests: int
    workers: int
    successful_assignments: int
    rejected_claims: int
    unmatched_riders: int
    duplicate_assignment_attempts: int
    duplicate_active_assignments: int
    strategy_disagreements: int
    throughput_rps: float
    p50_latency_ms: float
    p95_latency_ms: float

What Do the Result Models Capture?

• The DispatchResult model records one rider's selected driver.
• The attempted driver tuple preserves the ranking path taken during dispatch.
• The BatchMetrics model separates successes, rejections, unmatched riders, contention, and invariant violations.
• The latency fields hold measured local performance evidence.

• Save app/models.py.
• Confirm that the file now contains five frozen dataclasses.

Are the Result Models Misaligned?

Place the second chunk after the complete ClaimResult class. Keep every metric field at one indentation level inside BatchMetrics.

Use this to inspect the combined file: Help me find an indentation or type annotation problem in my model file.

The fixtures place synthetic coordinates around central Tampa. Integer division gives each consecutive pair of rider requests the same base location for later contention experiments.

• Create app/fixtures.py inside the existing app folder by using the file sidebar in VS Code.
• Add the deterministic Tampa-area fixtures by pasting this code into app/fixtures.py:

from app.models import Driver, RiderRequest

CENTER_LAT = 27.9506
CENTER_LNG = -82.4572

def generate_drivers(count: int) -> list[Driver]:
    return [
        Driver(
            f"driver-{index:03d}",
            CENTER_LAT + (index // 25 - 5) * 0.00018,
            CENTER_LNG + (index % 25 - 12) * 0.00018,
        )
        for index in range(count)
    ]

def generate_riders(count: int) -> list[RiderRequest]:
    riders = []
    for index in range(count):
        pair = index // 2
        riders.append(
            RiderRequest(
                f"ride-{index:03d}",
                CENTER_LAT + ((pair // 25) % 10 - 5) * 0.00018 + 0.00005,
                CENTER_LNG + (pair % 25 - 12) * 0.00018 + 0.00005,
            )
        )
    return riders

How Do the Fixtures Stay Deterministic?

• The center coordinates anchor generated locations around the same Tampa reference point.
• The driver index produces a repeatable grid offset.
• The formatted index produces a stable driver or ride identifier.
• The pair value places two consecutive riders at the same fixture location.

• Save app/fixtures.py.
• Confirm that models.py and fixtures.py appear inside the existing app folder.

Are the Fixture Imports Marked Invalid?

Check that fixtures.py sits inside the existing app folder. A file beside that folder cannot use the required app.models import path.

Use this to verify the file locations: Help me check where my model and fixture files belong.

Build H3 candidate search and scoring

H3 maps each coordinate to a geographic cell. A two-ring grid disk limits ranking to drivers near the rider without pretending to model a road network.

• Create app/geo.py inside the existing app folder by using the file sidebar in VS Code.
• Add the geographic configuration and candidate-cell helpers by pasting this first chunk into app/geo.py:

from collections import Counter
from math import asin, cos, radians, sin, sqrt
from typing import Literal

import h3

from app.models import Driver, RiderRequest

H3_RESOLUTION = 9
SEARCH_RADIUS = 2
SURGE_MIN = 1.0
SURGE_MAX = 3.0
AVERAGE_SPEED_KPH = 30.0
Strategy = Literal["nearest", "eta"]

def cell_for_location(lat: float, lng: float) -> str:
    return h3.latlng_to_cell(lat, lng, H3_RESOLUTION)

def nearby_cells(origin_cell: str) -> set[str]:
    return set(h3.grid_disk(origin_cell, SEARCH_RADIUS))

How Does Candidate Partitioning Work?

• The resolution constant gives every location the same cell granularity.
• The search radius includes cells within two grid steps of the rider's origin.
• The Strategy type limits ranking to nearest-distance or ETA-aware scoring.
• The candidate search remains a geographic filter for the scoring stages that follow.

• Save app/geo.py.
• Confirm that VS Code recognizes cell_for_location() and nearby_cells() without red syntax markers.

Is the H3 Import Unresolved?

Confirm that the import is lowercase h3. The dependency runs inside the existing API container even if the host editor cannot resolve it.

Use this if the function calls still look wrong: Help me diagnose the H3 import and helper calls.

Both ranking strategies need the same distance baseline. Great-circle distance supplies a deterministic measurement from latitude and longitude.

The distance formula is the fiddliest code in this step. Keep its parentheses and variable names exactly as shown.

• Add the distance helper below return set(h3.grid_disk(origin_cell, SEARCH_RADIUS)) by pasting this chunk:

def great_circle_km(a_lat: float, a_lng: float, b_lat: float, b_lng: float) -> float:
    radius_km = 6371.0
    lat_delta = radians(b_lat - a_lat)
    lng_delta = radians(b_lng - a_lng)
    left = radians(a_lat)
    right = radians(b_lat)
    value = sin(lat_delta / 2) ** 2 + cos(left) * cos(right) * sin(lng_delta / 2) ** 2
    return 2 * radius_km * asin(sqrt(value))

What Does the Distance Helper Calculate?

The helper converts coordinate differences into radians before applying the great-circle calculation. Its output gives both ranking strategies the same distance in kilometers.

• Save app/geo.py.
• Confirm that great_circle_km() ends with one numeric return expression.

Is the Distance Expression Marked Invalid?

Check that every opening parenthesis has a matching closing parenthesis. Keep the complete value expression inside the function.

Use this to inspect the formula: Help me compare my great-circle helper with the required implementation.

Traffic remains reproducible because every factor comes from an occupied H3 cell identifier. The same fixtures therefore produce the same traffic map on every run.

• Add the deterministic traffic function below great_circle_km() by pasting this chunk:

def traffic_conditions(drivers: list[Driver], riders: list[RiderRequest]) -> dict[str, float]:
    cells = {cell_for_location(item.lat, item.lng) for item in [*drivers, *riders]}
    levels = (1.0, 1.5, 2.2)
    return {cell: levels[int(cell[-2:], 16) % len(levels)] for cell in cells}

How Does Fixture Traffic Work?

• The set comprehension collects each occupied driver or rider cell.
• The traffic levels provide three deterministic adjustment factors.
• The final mapping derives one repeatable factor from each cell identifier.

• Save app/geo.py.
• Confirm that traffic_conditions() appears below the distance helper.

Is the Traffic Mapping Incomplete?

Keep the three levels inside one tuple. Confirm that the return expression indexes that tuple with the final two cell characters.

Use this to trace the mapping: Help me inspect how my deterministic traffic factors are created.

ETA converts distance into travel minutes at the configured average speed. A driver's cell factor stretches that baseline when fixture traffic is higher.

• Add the ETA function below traffic_conditions() by pasting this chunk:

def estimate_eta_minutes(
    rider: RiderRequest,
    driver: Driver,
    traffic_by_cell: dict[str, float],
) -> float:
    driver_cell = cell_for_location(driver.lat, driver.lng)
    traffic_factor = traffic_by_cell.get(driver_cell, 1.0)
    distance = great_circle_km(rider.lat, rider.lng, driver.lat, driver.lng)
    return (distance / AVERAGE_SPEED_KPH) * 60.0 * traffic_factor

How Is ETA Estimated?

• The driver's coordinates identify the traffic cell used for scoring.
• The default factor keeps a missing cell at normal traffic.
• The distance calculation supplies the unadjusted travel length.
• The final expression converts distance to minutes before applying traffic.

• Save app/geo.py.
• Confirm that estimate_eta_minutes() receives one rider, one driver, and one traffic mapping.

Is the ETA Function Incomplete?

Check that the function reads traffic from the driver's cell. Confirm that the return expression multiplies by traffic_factor.

Use this to trace the score: Help me follow the deterministic ETA calculation.

Both strategies rank the same H3 candidate set. The score changes from geographic distance to traffic-adjusted ETA, which allows the winning driver to change.

• Add candidate filtering and strategy ranking below estimate_eta_minutes() by pasting this chunk:

def rank_candidate_drivers(
    rider: RiderRequest,
    drivers: list[Driver],
    traffic_by_cell: dict[str, float],
    strategy: Strategy,
) -> list[Driver]:
    allowed = nearby_cells(cell_for_location(rider.lat, rider.lng))
    candidates = [d for d in drivers if cell_for_location(d.lat, d.lng) in allowed]
    if strategy == "nearest":
        return sorted(
            candidates,
            key=lambda d: (great_circle_km(rider.lat, rider.lng, d.lat, d.lng), d.driver_id),
        )
    if strategy == "eta":
        return sorted(
            candidates,
            key=lambda d: (estimate_eta_minutes(rider, d, traffic_by_cell), d.driver_id),
        )
    raise ValueError(f"Unknown strategy: {strategy}")

How Are Candidate Drivers Ranked?

• The grid disk creates the allowed cell set around the rider.
• The candidate list excludes drivers outside that neighborhood.
• The nearest strategy sorts candidates by great-circle distance.
• The ETA strategy sorts candidates by estimated arrival time.

• Save app/geo.py.
• Confirm that both strategy branches return a sorted driver list.

Is a Ranking Branch Marked Invalid?

Check that each sorting function returns a tuple containing its score plus d.driver_id. The identifier resolves equal scores deterministically.

Use this to compare the branches: Help me inspect my nearest and ETA sort keys.

Regional surge divides demand by available supply in each rider cell. Fixed bounds keep the educational multiplier between 1.0x and 3.0x.

• Add bounded surge and regional pricing below the ranking function by pasting this chunk:

def surge_multiplier(demand: int, supply: int) -> float:
    return round(max(SURGE_MIN, min(SURGE_MAX, demand / max(supply, 1))), 2)

def regional_pricing(drivers: list[Driver], riders: list[RiderRequest]) -> list[dict]:
    supply = Counter(cell_for_location(d.lat, d.lng) for d in drivers)
    demand = Counter(cell_for_location(r.lat, r.lng) for r in riders)
    return [
        {
            "cell": cell,
            "demand": count,
            "supply": supply.get(cell, 0),
            "surge": surge_multiplier(count, supply.get(cell, 0)),
        }
        for cell, count in sorted(demand.items())
    ]

How Is Regional Surge Bounded?

• The inner supply guard prevents division by zero.
• The minimum keeps low-demand regions at 1.0x.
• The maximum caps high-demand regions at 3.0x.
• The regional output keeps demand, supply, and surge visible for each rider cell.

• Save app/geo.py.
• Confirm that the file ends with the completed regional_pricing() list comprehension.

Is Regional Pricing Marked Invalid?

Check the braces around the regional dictionary. Keep the final comprehension inside the returned list.

Use this to inspect the final helpers: Help me check the surge bounds and regional pricing structure.

Preview and test the dispatch strategies

The preview turns the scoring rules into visible evidence. One driver starts closer to the rider while a higher traffic factor makes the other driver faster by ETA.

• Create a scripts folder beside the existing app folder by using the file sidebar in VS Code.
• Confirm that the new scripts folder appears in the repository.

• Create an empty scripts/__init__.py file by using the file sidebar in VS Code.
• Confirm that __init__.py appears inside the scripts folder.

• Create scripts/preview.py inside the existing scripts folder by using the file sidebar in VS Code.
• Add the deterministic strategy preview by pasting this code into scripts/preview.py:

from app.fixtures import CENTER_LAT, CENTER_LNG, generate_drivers, generate_riders
from app.geo import cell_for_location, rank_candidate_drivers, regional_pricing
from app.models import Driver, RiderRequest

def main() -> None:
    rider = RiderRequest("preview-ride", CENTER_LAT, CENTER_LNG)
    near = Driver("driver-nearest", CENTER_LAT, CENTER_LNG + 0.001)
    eta = Driver("driver-eta", CENTER_LAT, CENTER_LNG - 0.002)
    drivers = [near, eta]
    traffic = {
        cell_for_location(rider.lat, rider.lng): 1.0,
        cell_for_location(near.lat, near.lng): 3.0,
        cell_for_location(eta.lat, eta.lng): 1.0,
    }
    nearest_winner = rank_candidate_drivers(rider, drivers, traffic, "nearest")[0]
    eta_winner = rank_candidate_drivers(rider, drivers, traffic, "eta")[0]
    pricing = regional_pricing(generate_drivers(8), generate_riders(20))
    surges = [region["surge"] for region in pricing]
    print(f"H3 rider cell: {cell_for_location(rider.lat, rider.lng)}")
    print(f"Nearest-driver winner: {nearest_winner.driver_id}")
    print(f"ETA-aware winner: {eta_winner.driver_id}")
    print(f"Observed surge range: {min(surges):.1f}x to {max(surges):.1f}x")
    print("Educational model only. No live traffic or proprietary data is used.")

if __name__ == "__main__":
    main()

What Does the Preview Prove?

• The preview gives driver-nearest the shorter geographic distance.
• The larger traffic factor slows that driver's ETA score.
• The preview therefore selects driver-eta for ETA-aware ranking.
• The generated regional data exposes the surge range from deterministic fixtures.

• Save scripts/__init__.py.
• Save scripts/preview.py.
• Confirm that both files appear inside the scripts folder.

Are the Preview Imports Marked Invalid?

Confirm that scripts/__init__.py exists. Check that preview.py sits in the same folder.

Use this to inspect the package structure: Help me diagnose the preview module imports.

Before you run the preview, do you think geographic distance and traffic-adjusted ETA will choose the same driver?

• Run the deterministic strategy preview in the existing project terminal with this command:

docker compose exec api python -m scripts.preview

What Does This Command Run?

Docker Compose executes the preview module inside the running API service. The repository mount exposes your saved files to the container.

You should see an H3 rider cell followed by driver-nearest as the nearest-distance winner. You should see driver-eta as the ETA-aware winner.

The observed surge range stays between 1.0x and 3.0x. The final line states the educational scope of the model.

That disagreement is your first dispatch result. The simulator can now show why the closest driver may have the worse arrival time.

Does the Preview Fail or Choose One Winner Twice?

Check that all files are saved before running the command. Confirm that the preview assigns 3.0 to the nearest driver's cell.

Use this to trace unexpected output: Help me diagnose why the preview does not show different winners.

Automated tests preserve three geographic guarantees. They cover H3 neighborhood membership, the strategy disagreement, and the surge bounds.

• Create tests/test_geo.py inside the existing tests folder by using the file sidebar in VS Code.
• Add the geographic tests by pasting this code into tests/test_geo.py:

from app.fixtures import CENTER_LAT, CENTER_LNG
from app.geo import cell_for_location, nearby_cells, rank_candidate_drivers, surge_multiplier
from app.models import Driver, RiderRequest

def test_h3_cell_is_in_its_own_search_disk() -> None:
    cell = cell_for_location(CENTER_LAT, CENTER_LNG)
    assert cell in nearby_cells(cell)

def test_eta_aware_ranking_can_differ_from_nearest() -> None:
    rider = RiderRequest("ride-test", CENTER_LAT, CENTER_LNG)
    near = Driver("driver-nearest", CENTER_LAT, CENTER_LNG + 0.001)
    eta_driver = Driver("driver-eta", CENTER_LAT, CENTER_LNG - 0.002)
    drivers = [near, eta_driver]
    traffic = {
        cell_for_location(rider.lat, rider.lng): 1.0,
        cell_for_location(near.lat, near.lng): 3.0,
        cell_for_location(eta_driver.lat, eta_driver.lng): 1.0,
    }
    assert rank_candidate_drivers(rider, drivers, traffic, "nearest")[0].driver_id == "driver-nearest"
    assert rank_candidate_drivers(rider, drivers, traffic, "eta")[0].driver_id == "driver-eta"

def test_surge_multiplier_is_bounded() -> None:
    assert surge_multiplier(2, 10) == 1.0
    assert surge_multiplier(12, 8) == 1.5
    assert surge_multiplier(100, 0) == 3.0

What Do the Geographic Tests Protect?

• The first test proves that a rider's origin cell remains inside its own search disk.
• The second test preserves the deliberate ranking disagreement.
• The third test verifies the lower bound.
• The third test also verifies an intermediate multiplier plus the upper bound.

• Save tests/test_geo.py.
• Confirm that VS Code shows three geographic test functions.

Are the Geographic Tests Marked Invalid?

Confirm that test_geo.py sits inside the existing tests folder. Check that every imported function matches its definition in app/geo.py.

Use this to inspect the test file: Help me compare my geographic tests with the implemented helper names.

Before you run the suite, do you think the candidate search, ranking comparison, and surge bounds will all pass?

• Run the geographic tests in the existing project terminal with this command:

docker compose exec api pytest -q tests/test_geo.py

What Does This Test Command Check?

The command runs only the geographic test file inside the running API service. A passing result confirms the entire deterministic geography slice.

You should see three passing tests. Your simulator now produces stable H3 candidates, distinct strategy winners, and bounded regional surge values.

Are the Geographic Tests Failing?

Use the failing test name to narrow the problem. Inspect the H3 constants for the search test, the traffic factors for the ranking test, or the nested bounds for the surge test.

Use this to investigate the exact failure: Help me diagnose my failing geographic test.

Use this final comparison to confirm that every new file matches the deterministic implementation.

✔️ Awesome, I've got everything!

Your preview shows different strategy winners. Your three geographic tests pass.

ⓧ I'd like to double check the full code

from dataclasses import dataclass

@dataclass(frozen=True)
class Driver:
    driver_id: str
    lat: float
    lng: float

@dataclass(frozen=True)
class RiderRequest:
    ride_id: str
    lat: float
    lng: float

@dataclass(frozen=True)
class ClaimResult:
    success: bool
    driver_id: str
    ride_id: str
    existing_ride_id: str | None = None

@dataclass(frozen=True)
class DispatchResult:
    ride_id: str
    driver_id: str | None
    rejected_claims: int
    attempted_driver_ids: tuple[str, ...]
    latency_ms: float

@dataclass(frozen=True)
class BatchMetrics:
    mode: str
    requests: int
    workers: int
    successful_assignments: int
    rejected_claims: int
    unmatched_riders: int
    duplicate_assignment_attempts: int
    duplicate_active_assignments: int
    strategy_disagreements: int
    throughput_rps: float
    p50_latency_ms: float
    p95_latency_ms: float

from app.models import Driver, RiderRequest

CENTER_LAT = 27.9506
CENTER_LNG = -82.4572

def generate_drivers(count: int) -> list[Driver]:
    return [
        Driver(
            f"driver-{index:03d}",
            CENTER_LAT + (index // 25 - 5) * 0.00018,
            CENTER_LNG + (index % 25 - 12) * 0.00018,
        )
        for index in range(count)
    ]

def generate_riders(count: int) -> list[RiderRequest]:
    riders = []
    for index in range(count):
        pair = index // 2
        riders.append(
            RiderRequest(
                f"ride-{index:03d}",
                CENTER_LAT + ((pair // 25) % 10 - 5) * 0.00018 + 0.00005,
                CENTER_LNG + (pair % 25 - 12) * 0.00018 + 0.00005,
            )
        )
    return riders

from collections import Counter
from math import asin, cos, radians, sin, sqrt
from typing import Literal

import h3

from app.models import Driver, RiderRequest

H3_RESOLUTION = 9
SEARCH_RADIUS = 2
SURGE_MIN = 1.0
SURGE_MAX = 3.0
AVERAGE_SPEED_KPH = 30.0
Strategy = Literal["nearest", "eta"]

def cell_for_location(lat: float, lng: float) -> str:
    return h3.latlng_to_cell(lat, lng, H3_RESOLUTION)

def nearby_cells(origin_cell: str) -> set[str]:
    return set(h3.grid_disk(origin_cell, SEARCH_RADIUS))

def great_circle_km(a_lat: float, a_lng: float, b_lat: float, b_lng: float) -> float:
    radius_km = 6371.0
    lat_delta = radians(b_lat - a_lat)
    lng_delta = radians(b_lng - a_lng)
    left = radians(a_lat)
    right = radians(b_lat)
    value = sin(lat_delta / 2) ** 2 + cos(left) * cos(right) * sin(lng_delta / 2) ** 2
    return 2 * radius_km * asin(sqrt(value))

def traffic_conditions(drivers: list[Driver], riders: list[RiderRequest]) -> dict[str, float]:
    cells = {cell_for_location(item.lat, item.lng) for item in [*drivers, *riders]}
    levels = (1.0, 1.5, 2.2)
    return {cell: levels[int(cell[-2:], 16) % len(levels)] for cell in cells}

def estimate_eta_minutes(
    rider: RiderRequest,
    driver: Driver,
    traffic_by_cell: dict[str, float],
) -> float:
    driver_cell = cell_for_location(driver.lat, driver.lng)
    traffic_factor = traffic_by_cell.get(driver_cell, 1.0)
    distance = great_circle_km(rider.lat, rider.lng, driver.lat, driver.lng)
    return (distance / AVERAGE_SPEED_KPH) * 60.0 * traffic_factor

def rank_candidate_drivers(
    rider: RiderRequest,
    drivers: list[Driver],
    traffic_by_cell: dict[str, float],
    strategy: Strategy,
) -> list[Driver]:
    allowed = nearby_cells(cell_for_location(rider.lat, rider.lng))
    candidates = [d for d in drivers if cell_for_location(d.lat, d.lng) in allowed]
    if strategy == "nearest":
        return sorted(
            candidates,
            key=lambda d: (great_circle_km(rider.lat, rider.lng, d.lat, d.lng), d.driver_id),
        )
    if strategy == "eta":
        return sorted(
            candidates,
            key=lambda d: (estimate_eta_minutes(rider, d, traffic_by_cell), d.driver_id),
        )
    raise ValueError(f"Unknown strategy: {strategy}")

def surge_multiplier(demand: int, supply: int) -> float:
    return round(max(SURGE_MIN, min(SURGE_MAX, demand / max(supply, 1))), 2)

def regional_pricing(drivers: list[Driver], riders: list[RiderRequest]) -> list[dict]:
    supply = Counter(cell_for_location(d.lat, d.lng) for d in drivers)
    demand = Counter(cell_for_location(r.lat, r.lng) for r in riders)
    return [
        {
            "cell": cell,
            "demand": count,
            "supply": supply.get(cell, 0),
            "surge": surge_multiplier(count, supply.get(cell, 0)),
        }
        for cell, count in sorted(demand.items())
    ]

The scripts/__init__.py file should be present and empty.

from app.fixtures import CENTER_LAT, CENTER_LNG, generate_drivers, generate_riders
from app.geo import cell_for_location, rank_candidate_drivers, regional_pricing
from app.models import Driver, RiderRequest

def main() -> None:
    rider = RiderRequest("preview-ride", CENTER_LAT, CENTER_LNG)
    near = Driver("driver-nearest", CENTER_LAT, CENTER_LNG + 0.001)
    eta = Driver("driver-eta", CENTER_LAT, CENTER_LNG - 0.002)
    drivers = [near, eta]
    traffic = {
        cell_for_location(rider.lat, rider.lng): 1.0,
        cell_for_location(near.lat, near.lng): 3.0,
        cell_for_location(eta.lat, eta.lng): 1.0,
    }
    nearest_winner = rank_candidate_drivers(rider, drivers, traffic, "nearest")[0]
    eta_winner = rank_candidate_drivers(rider, drivers, traffic, "eta")[0]
    pricing = regional_pricing(generate_drivers(8), generate_riders(20))
    surges = [region["surge"] for region in pricing]
    print(f"H3 rider cell: {cell_for_location(rider.lat, rider.lng)}")
    print(f"Nearest-driver winner: {nearest_winner.driver_id}")
    print(f"ETA-aware winner: {eta_winner.driver_id}")
    print(f"Observed surge range: {min(surges):.1f}x to {max(surges):.1f}x")
    print("Educational model only. No live traffic or proprietary data is used.")

if __name__ == "__main__":
    main()

from app.fixtures import CENTER_LAT, CENTER_LNG
from app.geo import cell_for_location, nearby_cells, rank_candidate_drivers, surge_multiplier
from app.models import Driver, RiderRequest

def test_h3_cell_is_in_its_own_search_disk() -> None:
    cell = cell_for_location(CENTER_LAT, CENTER_LNG)
    assert cell in nearby_cells(cell)

def test_eta_aware_ranking_can_differ_from_nearest() -> None:
    rider = RiderRequest("ride-test", CENTER_LAT, CENTER_LNG)
    near = Driver("driver-nearest", CENTER_LAT, CENTER_LNG + 0.001)
    eta_driver = Driver("driver-eta", CENTER_LAT, CENTER_LNG - 0.002)
    drivers = [near, eta_driver]
    traffic = {
        cell_for_location(rider.lat, rider.lng): 1.0,
        cell_for_location(near.lat, near.lng): 3.0,
        cell_for_location(eta_driver.lat, eta_driver.lng): 1.0,
    }
    assert rank_candidate_drivers(rider, drivers, traffic, "nearest")[0].driver_id == "driver-nearest"
    assert rank_candidate_drivers(rider, drivers, traffic, "eta")[0].driver_id == "driver-eta"

def test_surge_multiplier_is_bounded() -> None:
    assert surge_multiplier(2, 10) == 1.0
    assert surge_multiplier(12, 8) == 1.5
    assert surge_multiplier(100, 0) == 3.0

Your dispatch lab can now partition Tampa into candidate cells and explain why the closest driver may lose to a better ETA. Next, you will use competing workers to expose the assignment race that ranking alone cannot prevent.

Reproduce the Assignment Race

Your simulator now partitions Tampa into H3 cells. It can rank drivers by distance or deterministic ETA.

The assignment boundary still has a concurrency race. A separate Redis read can let both workers observe the same driver as available before either worker writes.

You will force that timing with two worker threads. The resulting evidence makes the invariant violation reproducible.

In this step, get ready to:

• Build the unsafe read-then-write assignment path.
• Audit active ride records for duplicate driver assignments.
• Force two workers to claim the same driver concurrently.

Build the unsafe Redis assignment store

Each assignment uses a driver key to record the active ride. A reverse ride key records which driver accepted that ride.

• Switch back to the starter repository in Visual Studio Code.
• Create app/store.py inside the app folder.
• Paste this first section into app/store.py:

from collections import Counter
from pathlib import Path
from threading import Barrier

import redis

from app.models import ClaimResult

class RedisAssignmentStore:
    def __init__(self, client: redis.Redis | None = None) -> None:
        self.client = client or redis.Redis(
            host="redis", port=6379, db=0, decode_responses=True
        )
        path = Path(__file__).parent / "lua" / "claim_driver.lua"
        self.claim_script = path.read_text(encoding="utf-8")

    @staticmethod
    def driver_key(driver_id: str) -> str:
        return f"dispatch:driver:{driver_id}:active"

    @staticmethod
    def ride_key(ride_id: str) -> str:
        return f"dispatch:ride:{ride_id}:active"

    def ping(self) -> bool:
        return bool(self.client.ping())

What does this code do?

• The constructor connects to the Compose Redis service with decoded string responses.
• The key helpers give every driver assignment and ride assignment a predictable address.
• The ping() method provides a direct Redis health check.
• The script file is loaded now so the next step can reuse this store without changing its interface.

• Save app/store.py.

You should now see store.py inside the app folder. The file contains the connection plus both assignment-key helpers.

Seeing import warnings in the store?

Check that store.py sits directly inside the existing app folder. Keep the imports aligned with the left edge of the file.

The Python packages run inside the API container. A local editor warning does not mean the container is missing the package.

Help me diagnose the imports in app/store.py.

The unsafe claim splits one decision across multiple Redis commands. The optional barrier pauses both workers after the availability read so each worker reaches the writes with the same stale observation.

• Add these methods below ping() in app/store.py:

    def reset(self) -> None:
        keys = self.client.keys("dispatch:*")
        if keys:
            self.client.delete(*keys)

    def claim_unsafe(
        self, driver_id: str, ride_id: str, barrier: Barrier | None = None
    ) -> ClaimResult:
        existing = self.client.get(self.driver_key(driver_id))
        if existing is not None:
            return ClaimResult(False, driver_id, ride_id, existing)
        if barrier is not None:
            barrier.wait()
        self.client.set(self.driver_key(driver_id), ride_id)
        self.client.set(self.ride_key(ride_id), driver_id)
        return ClaimResult(True, driver_id, ride_id)

    def driver_assignment(self, driver_id: str) -> str | None:
        return self.client.get(self.driver_key(driver_id))

    def ride_assignment(self, ride_id: str) -> str | None:
        return self.client.get(self.ride_key(ride_id))

Where is the race?

• The availability read happens before either assignment write.
• The optional Barrier holds both workers after they observe an available driver.
• Each released worker writes the driver key plus its own ride key.
• Both workers return a successful ClaimResult because neither repeats the availability check.

• Save app/store.py.

The editor should now show claim_unsafe() between reset() and driver_assignment().

Does the new method sit outside the class?

Keep every new method indented by four spaces beneath RedisAssignmentStore. Keep each method body indented by another four spaces.

Help me correct the indentation in RedisAssignmentStore.

Audit duplicate active assignments

The driver key can hold only the most recent ride ID. Counting reverse ride records by driver exposes every active ride that survived the race.

• Add these audit methods below ride_assignment() in app/store.py:

    def active_ride_assignments(self) -> dict[str, str]:
        assignments = {}
        for key in sorted(self.client.keys("dispatch:ride:*:active")):
            driver_id = self.client.get(key)
            if driver_id is not None:
                assignments[key.split(":")[2]] = driver_id
        return assignments

    def duplicate_active_assignments(self) -> int:
        counts = Counter(self.active_ride_assignments().values())
        return sum(max(0, count - 1) for count in counts.values())

Why count the ride records?

Duplicate assignment attempts measure contention around a driver. Rejected claims show that current state blocked an attempt.

Duplicate active assignments are different. They count extra active ride records grouped by driver, which directly measures violations of the one-active-ride invariant.

• Save app/store.py.

Your store now exposes the raw ride assignments plus the number of extra active rides per driver.

Does the duplicate count look unclear?

Check that Counter receives the driver IDs from active_ride_assignments().values(). Counting ride IDs would measure a different property.

Help me understand the duplicate active assignment calculation.

A Pytest fixture gives every race test a clean assignment store. The cleanup after yield prevents one test from leaking Redis keys into another.

• Create tests/conftest.py inside the existing tests folder.
• Paste this fixture into tests/conftest.py:

import pytest
from app.store import RedisAssignmentStore

@pytest.fixture
def store() -> RedisAssignmentStore:
    assignment_store = RedisAssignmentStore()
    assignment_store.reset()
    yield assignment_store
    assignment_store.reset()

How does the fixture isolate the test?

• The first reset removes dispatch keys left by an earlier run.
• The fixture yields one connected RedisAssignmentStore to the test.
• The final reset removes the race evidence after the test finishes.

• Save tests/conftest.py.

You should now see conftest.py in the tests folder. That file provides the shared store fixture.

Is the fixture unavailable to tests?

Confirm the filename is exactly conftest.py. Confirm it sits directly inside the same tests folder as the race test.

Help me diagnose a missing Pytest fixture.

Force both workers through the race

A threading.Barrier releases waiting threads only after both workers reach it. This removes timing luck from the demonstration.

Both workers target driver-001. Each worker uses a different ride ID so the reverse records reveal whether one driver gained two active rides.

• Create tests/test_race.py inside the tests folder.
• Paste this forced race test into tests/test_race.py:

from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from app.store import RedisAssignmentStore

def test_forced_race_creates_duplicate_active_assignment(
    store: RedisAssignmentStore,
) -> None:
    barrier = Barrier(2, timeout=5)
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(store.claim_unsafe, "driver-001", ride, barrier)
            for ride in ("ride-a", "ride-b")
        ]
        results = [future.result() for future in futures]
    successful = sum(result.success for result in results)
    rejected = sum(not result.success for result in results)
    duplicates = store.duplicate_active_assignments()
    print(f"successful={successful} rejected={rejected} duplicate_active={duplicates}")
    assert successful == 2
    assert rejected == 0
    assert duplicates == 1
    assert store.ride_assignment("ride-a") == "driver-001"
    assert store.ride_assignment("ride-b") == "driver-001"

How does the test force the failure?

• The thread pool starts exactly two competing workers.
• The barrier waits for both workers after their driver reads.
• The assertions require both unsafe claims to report success.
• The final assertions prove both reverse ride records point to driver-001.

• Save tests/test_race.py.

Your three race artifacts should now match the complete reference below.

✔️ Awesome, I've got everything!

Your unsafe store plus its isolated race test are ready. Save all three files before running the evidence check.

ⓧ I'd like to double check the full code

Compare each file with these complete references. Preserve the indentation inside RedisAssignmentStore.

from collections import Counter
from pathlib import Path
from threading import Barrier

import redis

from app.models import ClaimResult

class RedisAssignmentStore:
    def __init__(self, client: redis.Redis | None = None) -> None:
        self.client = client or redis.Redis(
            host="redis", port=6379, db=0, decode_responses=True
        )
        path = Path(__file__).parent / "lua" / "claim_driver.lua"
        self.claim_script = path.read_text(encoding="utf-8")

    @staticmethod
    def driver_key(driver_id: str) -> str:
        return f"dispatch:driver:{driver_id}:active"

    @staticmethod
    def ride_key(ride_id: str) -> str:
        return f"dispatch:ride:{ride_id}:active"

    def ping(self) -> bool:
        return bool(self.client.ping())

    def reset(self) -> None:
        keys = self.client.keys("dispatch:*")
        if keys:
            self.client.delete(*keys)

    def claim_unsafe(
        self, driver_id: str, ride_id: str, barrier: Barrier | None = None
    ) -> ClaimResult:
        existing = self.client.get(self.driver_key(driver_id))
        if existing is not None:
            return ClaimResult(False, driver_id, ride_id, existing)
        if barrier is not None:
            barrier.wait()
        self.client.set(self.driver_key(driver_id), ride_id)
        self.client.set(self.ride_key(ride_id), driver_id)
        return ClaimResult(True, driver_id, ride_id)

    def driver_assignment(self, driver_id: str) -> str | None:
        return self.client.get(self.driver_key(driver_id))

    def ride_assignment(self, ride_id: str) -> str | None:
        return self.client.get(self.ride_key(ride_id))

    def active_ride_assignments(self) -> dict[str, str]:
        assignments = {}
        for key in sorted(self.client.keys("dispatch:ride:*:active")):
            driver_id = self.client.get(key)
            if driver_id is not None:
                assignments[key.split(":")[2]] = driver_id
        return assignments

    def duplicate_active_assignments(self) -> int:
        counts = Counter(self.active_ride_assignments().values())
        return sum(max(0, count - 1) for count in counts.values())

What should this file contain?

This reference combines the Redis connection plus the unsafe claim path. It also includes the ride-record audit that detects hidden duplicate assignments.

import pytest
from app.store import RedisAssignmentStore

@pytest.fixture
def store() -> RedisAssignmentStore:
    assignment_store = RedisAssignmentStore()
    assignment_store.reset()
    yield assignment_store
    assignment_store.reset()

What should this fixture contain?

This reference resets dispatch keys before the test. It performs the same cleanup after the test completes.

from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from app.store import RedisAssignmentStore

def test_forced_race_creates_duplicate_active_assignment(
    store: RedisAssignmentStore,
) -> None:
    barrier = Barrier(2, timeout=5)
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(store.claim_unsafe, "driver-001", ride, barrier)
            for ride in ("ride-a", "ride-b")
        ]
        results = [future.result() for future in futures]
    successful = sum(result.success for result in results)
    rejected = sum(not result.success for result in results)
    duplicates = store.duplicate_active_assignments()
    print(f"successful={successful} rejected={rejected} duplicate_active={duplicates}")
    assert successful == 2
    assert rejected == 0
    assert duplicates == 1
    assert store.ride_assignment("ride-a") == "driver-001"
    assert store.ride_assignment("ride-b") == "driver-001"

What should this test prove?

This reference forces both workers past the same stale availability read. Its assertions treat two active ride records for one driver as the invariant violation.

Before you run the test, do you think both workers can report success after the barrier releases them?

• Run the forced race test inside the API service with this command:

docker compose exec api pytest -q -s tests/test_race.py

What should you see?

The evidence prints successful=2 rejected=0 duplicate_active=1. The test suite reports one passing test because this test successfully reproduced the intended failure.

Both workers claimed the driver. The two ride records prove that driver-001 now has two active rides.

That is the designed shortfall. Selection succeeded twice because the availability read and assignment writes formed separate correctness boundaries.

Did the race test fail or hang?

Check that both submitted calls receive the same barrier object. Check that barrier.wait() remains after the availability read.

If only one claim succeeds, confirm that the driver write still happens after the barrier. An earlier write lets the second worker observe the assignment.

Help me debug the forced assignment race.

You now have deterministic evidence of the assignment race. Next, you will close this correctness gap while preserving the winning assignment.

Claim Drivers Atomically and Benchmark

Your forced race proved that driver selection and assignment cannot be separate correctness boundaries. Both workers observed the same available driver before either worker recorded its claim.

This step moves the conditional writes into an atomic Redis Lua script. You will then benchmark the unsafe path against the atomic path with the same deterministic workload.

In this step, get ready to:

• Create a Lua claim that writes both assignment directions atomically.
• Connect atomic claims to the two-worker dispatch workload.
• Generate a measured reliability report for the concert spike.

Create the atomic Lua claim

A Lua script lets Redis read the current assignments before conditionally writing both keys in one operation. Redis executes the short script atomically.

• In the VS Code Explorer sidebar, create an `app/lua` folder inside `app`.
• Inside `app/lua`, create a file named `claim_driver.lua`.
• Add the atomic claim logic by copying this code into `app/lua/claim_driver.lua`:

local existing_ride = redis.call('GET', KEYS[1])
if existing_ride then
    return {0, existing_ride}
end
if redis.call('EXISTS', KEYS[2]) == 1 then
    return {0, redis.call('GET', KEYS[2])}
end
redis.call('SET', KEYS[1], ARGV[1])
redis.call('SET', KEYS[2], ARGV[2])
return {1, ARGV[1]}

What does this script enforce?

• The first lookup rejects a claim when the driver already has an active ride.
• The second lookup rejects a claim when the ride already has an active driver.
• The two `SET` calls create the driver-to-ride record plus the ride-to-driver record inside the same atomic execution.

• Save `app/lua/claim_driver.lua`.
• Confirm that `claim_driver.lua` appears inside the `app/lua` folder in the Explorer sidebar.

Cannot find the Lua file?

Check that `claim_driver.lua` is nested inside `app/lua`. A file created directly inside `app` will not match the path loaded by `RedisAssignmentStore`.

Ask for help locating the script with this debugging prompt.

The assignment store needs to load this script before it can invoke `EVAL`. The returned values also need to become a `ClaimResult` with the existing ride preserved after a rejection.

• Return to `app/store.py` in the VS Code editor.
• Add the `Path` import below the existing `Counter` import:

from pathlib import Path

Why load the script from a file?

`Path` gives the assignment store a location relative to `app/store.py`. This keeps the Redis logic in a dedicated Lua file that can be reviewed independently.

• Find the Redis client setup inside `RedisAssignmentStore.__init__()`.
• Add these lines below the `self.client` assignment:

        path = Path(__file__).parent / "lua" / "claim_driver.lua"
        self.claim_script = path.read_text(encoding="utf-8")

What gets loaded?

`path` points to the Lua file beside the Python package. `claim_script` holds its text so each atomic claim can send the same script to Redis.

• Find `claim_unsafe()` in `app/store.py`.
• Add `claim_atomic()` directly below `claim_unsafe()`:

    def claim_atomic(self, driver_id: str, ride_id: str) -> ClaimResult:
        result = self.client.execute_command(
            "EVAL",
            self.claim_script,
            2,
            self.driver_key(driver_id),
            self.ride_key(ride_id),
            ride_id,
            driver_id,
        )
        success = bool(result[0])
        return ClaimResult(
            success,
            driver_id,
            ride_id,
            None if success else str(result[1]),
        )

How does the Redis call work?

• `EVAL` sends the loaded script to Redis for one atomic execution.
• The value `2` declares the driver key plus the ride key as script keys.
• `ride_id` plus `driver_id` become the values the script writes through `ARGV`.
• A rejected claim keeps the winning ride identifier in `existing_ride_id`.

• Save `app/store.py`.

✔️ Awesome, I've got everything!

Your assignment store now retains the unsafe path for comparison. It also exposes an atomic claim backed by the Lua script.

ⓧ I'd like to double check the full code

from collections import Counter
from pathlib import Path
from threading import Barrier

import redis

from app.models import ClaimResult

class RedisAssignmentStore:
    def __init__(self, client: redis.Redis | None = None) -> None:
        self.client = client or redis.Redis(
            host="redis", port=6379, db=0, decode_responses=True
        )
        path = Path(__file__).parent / "lua" / "claim_driver.lua"
        self.claim_script = path.read_text(encoding="utf-8")

    @staticmethod
    def driver_key(driver_id: str) -> str:
        return f"dispatch:driver:{driver_id}:active"

    @staticmethod
    def ride_key(ride_id: str) -> str:
        return f"dispatch:ride:{ride_id}:active"

    def ping(self) -> bool:
        return bool(self.client.ping())

    def reset(self) -> None:
        keys = self.client.keys("dispatch:*")
        if keys:
            self.client.delete(*keys)

    def claim_unsafe(
        self, driver_id: str, ride_id: str, barrier: Barrier | None = None
    ) -> ClaimResult:
        existing = self.client.get(self.driver_key(driver_id))
        if existing is not None:
            return ClaimResult(False, driver_id, ride_id, existing)
        if barrier is not None:
            barrier.wait()
        self.client.set(self.driver_key(driver_id), ride_id)
        self.client.set(self.ride_key(ride_id), driver_id)
        return ClaimResult(True, driver_id, ride_id)

    def claim_atomic(self, driver_id: str, ride_id: str) -> ClaimResult:
        result = self.client.execute_command(
            "EVAL",
            self.claim_script,
            2,
            self.driver_key(driver_id),
            self.ride_key(ride_id),
            ride_id,
            driver_id,
        )
        success = bool(result[0])
        return ClaimResult(
            success,
            driver_id,
            ride_id,
            None if success else str(result[1]),
        )

    def driver_assignment(self, driver_id: str) -> str | None:
        return self.client.get(self.driver_key(driver_id))

    def ride_assignment(self, ride_id: str) -> str | None:
        return self.client.get(self.ride_key(ride_id))

    def active_ride_assignments(self) -> dict[str, str]:
        assignments = {}
        for key in sorted(self.client.keys("dispatch:ride:*:active")):
            driver_id = self.client.get(key)
            if driver_id is not None:
                assignments[key.split(":")[2]] = driver_id
        return assignments

    def duplicate_active_assignments(self) -> int:
        counts = Counter(self.active_ride_assignments().values())
        return sum(max(0, count - 1) for count in counts.values())

What should match?

The complete store loads `claim_driver.lua` during construction. Its unsafe methods plus invariant-audit methods remain unchanged for the before-and-after comparison.

The first atomic tests prove the write boundary directly. One test checks sequential rejection after a winner exists. The other test submits two claims concurrently.

• Inside the `tests` folder, create `test_atomic.py`.
• Add the imports plus the overwrite-protection test by copying this code:

from concurrent.futures import ThreadPoolExecutor
from app.store import RedisAssignmentStore

def test_failed_lua_claim_cannot_overwrite_existing_assignment(
    store: RedisAssignmentStore,
) -> None:
    winner = store.claim_atomic("driver-001", "ride-a")
    loser = store.claim_atomic("driver-001", "ride-b")
    assert winner.success is True
    assert loser.success is False
    assert loser.existing_ride_id == "ride-a"
    assert store.driver_assignment("driver-001") == "ride-a"
    assert store.ride_assignment("ride-a") == "driver-001"
    assert store.ride_assignment("ride-b") is None
    assert store.duplicate_active_assignments() == 0

What does the first test prove?

The winner creates both assignment directions. The rejected claim preserves the winner without creating a record for `ride-b`.

• Save `tests/test_atomic.py`.
• Before running the test, predict whether the losing claim can overwrite `ride-a`.
• Run the atomic test file from the repository terminal:

docker compose exec api pytest -q -s tests/test_atomic.py

What should you see?

The first atomic test passes. Its assertions confirm that the winner remains assigned while the losing ride has no reverse record.

Does the first atomic test fail?

Check that `claim_driver.lua` uses `KEYS[1]` for the driver record. Confirm that it returns before either `SET` call when that record exists.

Use this atomic-claim debugging prompt to trace the returned values.

• Add the concurrent atomic claim test below the first test:

def test_atomic_claim_allows_one_winner_and_one_rejection(
    store: RedisAssignmentStore,
) -> None:
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(store.claim_atomic, "driver-001", ride)
            for ride in ("ride-a", "ride-b")
        ]
        results = [future.result() for future in futures]
    assert sum(result.success for result in results) == 1
    assert sum(not result.success for result in results) == 1
    assert store.duplicate_active_assignments() == 0

Why submit two claims?

The thread pool creates direct contention for `driver-001`. Atomic execution permits one winner plus one rejection while preserving zero duplicate active assignments.

• Save `tests/test_atomic.py`.
• Run the atomic test file again:

docker compose exec api pytest -q -s tests/test_atomic.py

What should you see now?

Both atomic tests pass. The second test proves that direct contention produces one successful claim plus one rejected claim.

That locks down the critical boundary. A rejected competitor can no longer mutate the winning assignment.

Connect atomic claims to dispatch

A correct claim method still needs workload-level evidence. The dispatch runner must track outcomes separately so contention never gets confused with an invariant violation.

• Inside the `app` folder, create `dispatch.py`.
• Use the complete file reference below to add the dispatch workload.

✔️ Awesome, I've got everything!

Your dispatch runner ranks candidates before attempting claims. It records each outcome plus the latency observed for every rider.

ⓧ I'd like to double check the full code

• Replace the contents of `app/dispatch.py` with this complete file:

from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict
from math import ceil
from threading import Barrier
from time import perf_counter
from typing import Literal

from app.geo import Strategy, rank_candidate_drivers
from app.models import BatchMetrics, DispatchResult, Driver, RiderRequest
from app.store import RedisAssignmentStore

ClaimMode = Literal["unsafe", "atomic"]

def dispatch_one(
    rider: RiderRequest,
    drivers: list[Driver],
    traffic: dict[str, float],
    strategy: Strategy,
    mode: ClaimMode,
    store: RedisAssignmentStore,
    barrier: Barrier | None = None,
) -> DispatchResult:
    started = perf_counter()
    attempted = []
    rejected = 0
    for driver in rank_candidate_drivers(rider, drivers, traffic, strategy):
        attempted.append(driver.driver_id)
        claim = (
            store.claim_unsafe(driver.driver_id, rider.ride_id, barrier)
            if mode == "unsafe"
            else store.claim_atomic(driver.driver_id, rider.ride_id)
        )
        if claim.success:
            return DispatchResult(
                rider.ride_id,
                driver.driver_id,
                rejected,
                tuple(attempted),
                (perf_counter() - started) * 1000,
            )
        rejected += 1
    return DispatchResult(
        rider.ride_id,
        None,
        rejected,
        tuple(attempted),
        (perf_counter() - started) * 1000,
    )

def percentile(values: list[float], fraction: float) -> float:
    ordered = sorted(values)
    return ordered[max(0, ceil(fraction * len(ordered)) - 1)]

def count_strategy_disagreements(
    riders: list[RiderRequest], drivers: list[Driver], traffic: dict[str, float]
) -> int:
    total = 0
    for rider in riders:
        nearest = rank_candidate_drivers(rider, drivers, traffic, "nearest")
        eta = rank_candidate_drivers(rider, drivers, traffic, "eta")
        total += bool(nearest and eta and nearest[0].driver_id != eta[0].driver_id)
    return total

def run_batch(
    riders: list[RiderRequest],
    drivers: list[Driver],
    traffic: dict[str, float],
    strategy: Strategy,
    mode: ClaimMode,
    workers: int,
    store: RedisAssignmentStore,
) -> BatchMetrics:
    if workers != 2:
        raise ValueError("This experiment requires exactly two workers")
    if len(riders) % 2:
        raise ValueError("Use an even number of paired rider requests")
    store.reset()
    results = []
    started = perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as executor:
        for index in range(0, len(riders), 2):
            barrier = Barrier(2, timeout=5) if mode == "unsafe" else None
            futures = [
                executor.submit(
                    dispatch_one,
                    rider,
                    drivers,
                    traffic,
                    strategy,
                    mode,
                    store,
                    barrier,
                )
                for rider in riders[index : index + 2]
            ]
            results.extend(future.result() for future in futures)
    duration = perf_counter() - started
    attempts = Counter(
        driver_id for result in results for driver_id in result.attempted_driver_ids
    )
    latencies = [result.latency_ms for result in results]
    successful = sum(result.driver_id is not None for result in results)
    return BatchMetrics(
        mode,
        len(riders),
        workers,
        successful,
        sum(result.rejected_claims for result in results),
        len(riders) - successful,
        sum(max(0, count - 1) for count in attempts.values()),
        store.duplicate_active_assignments(),
        count_strategy_disagreements(riders, drivers, traffic),
        round(len(riders) / duration, 2),
        round(percentile(latencies, 0.50), 3),
        round(percentile(latencies, 0.95), 3),
    )

def compare_safety_paths(
    riders: list[RiderRequest],
    drivers: list[Driver],
    traffic: dict[str, float],
    workers: int,
    store: RedisAssignmentStore,
) -> dict[str, dict]:
    unsafe = run_batch(riders, drivers, traffic, "eta", "unsafe", workers, store)
    atomic = run_batch(riders, drivers, traffic, "eta", "atomic", workers, store)
    return {"unsafe": asdict(unsafe), "atomic": asdict(atomic)}

How is the workload measured?

`dispatch_one()` records attempted drivers plus rejected claims for one rider. It measures elapsed dispatch time from candidate ranking through the final outcome.

`run_batch()` applies the forced barrier only to unsafe pairs. `compare_safety_paths()` runs the ETA-aware unsafe workload before the atomic workload.

• Save `app/dispatch.py`.

Seeing an import error in dispatch.py?

Compare the imported model names with `app/models.py`. The dispatch file requires `BatchMetrics`, `DispatchResult`, `Driver`, plus `RiderRequest`.

Use this import debugging prompt if the module still fails to load.

The API exposes the same experiment through a benchmark endpoint. This makes the unsafe and atomic results available as structured JSON while the command-line report remains the evidence artifact.

• Return to `app/main.py`.
• Add the dispatch comparison import above the existing fixture import:

from app.dispatch import compare_safety_paths

What does the API import?

`compare_safety_paths()` runs both claim modes against equivalent deterministic fixtures. The endpoint can return the resulting metric dictionaries directly.

• Add the benchmark endpoint below `scenario()`:

@app.post("/benchmark")
def benchmark(requests: int = 40, drivers: int = 20) -> dict:
    driver_data = generate_drivers(drivers)
    rider_data = generate_riders(requests)
    return compare_safety_paths(
        rider_data,
        driver_data,
        traffic_conditions(driver_data, rider_data),
        2,
        store,
    )

What does the endpoint compare?

The endpoint creates the requested deterministic fixtures. It then runs the unsafe path plus the atomic path with exactly two workers.

• Save `app/main.py`.

✔️ Awesome, I've got everything!

Your API now exposes health, scenario, plus benchmark endpoints from the same local application.

ⓧ I'd like to double check the full code

from fastapi import FastAPI

from app.dispatch import compare_safety_paths
from app.fixtures import generate_drivers, generate_riders
from app.geo import regional_pricing, traffic_conditions
from app.store import RedisAssignmentStore

app = FastAPI(title="Ride Dispatch Reliability Lab")
store = RedisAssignmentStore()

@app.get("/health")
def health() -> dict:
    return {"status": "ok", "redis": store.ping()}

@app.get("/scenario")
def scenario(requests: int = 40, drivers: int = 20) -> dict:
    driver_data = generate_drivers(drivers)
    rider_data = generate_riders(requests)
    return {
        "requests": requests,
        "drivers": drivers,
        "regions": regional_pricing(driver_data, rider_data),
        "invariant": "each driver has at most one active ride assignment",
        "scope": "educational simulation, not Uber's proprietary algorithms",
    }

@app.post("/benchmark")
def benchmark(requests: int = 40, drivers: int = 20) -> dict:
    driver_data = generate_drivers(drivers)
    rider_data = generate_riders(requests)
    return compare_safety_paths(
        rider_data,
        driver_data,
        traffic_conditions(driver_data, rider_data),
        2,
        store,
    )

What should the full API contain?

`/health` proves Redis reachability. `/scenario` returns deterministic regional context. `/benchmark` compares both safety paths.

The batch test now checks metric separation. Successful assignments, rejected claims, unmatched riders, repeated attempts, plus duplicate active assignments each answer a different reliability question.

• Return to `tests/test_atomic.py`.
• Add these imports below the `ThreadPoolExecutor` import:

from app.dispatch import run_batch
from app.fixtures import generate_drivers, generate_riders
from app.geo import traffic_conditions

Why use deterministic fixtures?

The generated drivers plus riders repeat the same paired contention pattern. Traffic values come from fixture locations instead of live data.

• Add the batch metric test below the concurrent atomic test:

def test_batch_metrics_separate_rejections_and_unmatched(
    store: RedisAssignmentStore,
) -> None:
    drivers = generate_drivers(2)
    riders = generate_riders(4)
    traffic = traffic_conditions(drivers, riders)
    metrics = run_batch(riders, drivers, traffic, "eta", "atomic", 2, store)
    print(metrics)
    assert metrics.successful_assignments == 2
    assert metrics.unmatched_riders == 2
    assert metrics.rejected_claims > 0
    assert metrics.duplicate_assignment_attempts > 0
    assert metrics.duplicate_active_assignments == 0

What does this batch prove?

Two drivers can serve two riders. The remaining riders become unmatched after their candidate claims are rejected.

Rejected claims plus duplicate attempts show contention. Zero duplicate active assignments proves that contention never breaks the invariant.

• Save `tests/test_atomic.py`.

✔️ Awesome, I've got everything!

Your atomic test file now covers overwrite protection, direct contention, plus batch metric separation.

ⓧ I'd like to double check the full code

from concurrent.futures import ThreadPoolExecutor
from app.dispatch import run_batch
from app.fixtures import generate_drivers, generate_riders
from app.geo import traffic_conditions
from app.store import RedisAssignmentStore

def test_failed_lua_claim_cannot_overwrite_existing_assignment(
    store: RedisAssignmentStore,
) -> None:
    winner = store.claim_atomic("driver-001", "ride-a")
    loser = store.claim_atomic("driver-001", "ride-b")
    assert winner.success is True
    assert loser.success is False
    assert loser.existing_ride_id == "ride-a"
    assert store.driver_assignment("driver-001") == "ride-a"
    assert store.ride_assignment("ride-a") == "driver-001"
    assert store.ride_assignment("ride-b") is None
    assert store.duplicate_active_assignments() == 0

def test_atomic_claim_allows_one_winner_and_one_rejection(
    store: RedisAssignmentStore,
) -> None:
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(store.claim_atomic, "driver-001", ride)
            for ride in ("ride-a", "ride-b")
        ]
        results = [future.result() for future in futures]
    assert sum(result.success for result in results) == 1
    assert sum(not result.success for result in results) == 1
    assert store.duplicate_active_assignments() == 0

def test_batch_metrics_separate_rejections_and_unmatched(
    store: RedisAssignmentStore,
) -> None:
    drivers = generate_drivers(2)
    riders = generate_riders(4)
    traffic = traffic_conditions(drivers, riders)
    metrics = run_batch(riders, drivers, traffic, "eta", "atomic", 2, store)
    print(metrics)
    assert metrics.successful_assignments == 2
    assert metrics.unmatched_riders == 2
    assert metrics.rejected_claims > 0
    assert metrics.duplicate_assignment_attempts > 0
    assert metrics.duplicate_active_assignments == 0

What should the complete test file cover?

The three tests move from a sequential conflict to concurrent contention. The final test checks that every reliability outcome retains its own metric.

Before you run the completed tests, predict whether concurrency produces a rejected claim or a duplicate active assignment on the atomic path.

• Run the completed atomic test file:

docker compose exec api pytest -q -s tests/test_atomic.py

What should the completed test run show?

All three tests pass. The printed batch metrics include rejections plus duplicate attempts while `duplicate_active_assignments` remains zero.

Do the batch assertions fail?

Confirm that `run_batch()` resets Redis before each mode. Check that unsafe pairs receive a barrier while atomic pairs receive `None`.

Use this batch debugging prompt to trace each metric.

Your atomic path now survives direct contention plus a paired batch. The next check turns that correctness result into measured evidence from a larger local workload.

Benchmark the concert spike

The benchmark runs both safety paths against the same deterministic inputs. It records outcome counts separately from locally measured throughput plus latency percentiles.

Timing values depend on your machine. The report preserves your measurements without turning them into vendor performance claims.

• Inside the `scripts` folder, create `benchmark.py`.
• Use the complete file reference below to add the benchmark command.

✔️ Awesome, I've got everything!

Your benchmark command compares the unsafe plus atomic paths. It refuses to write a successful report if the atomic invariant fails.

ⓧ I'd like to double check the full code

• Replace the contents of `scripts/benchmark.py` with this complete file:

import argparse
import platform
import sys
from pathlib import Path

from app.dispatch import compare_safety_paths
from app.fixtures import generate_drivers, generate_riders
from app.geo import traffic_conditions
from app.store import RedisAssignmentStore

METRICS = (
    ("Successful assignments", "successful_assignments"),
    ("Rejected claims", "rejected_claims"),
    ("Unmatched riders", "unmatched_riders"),
    ("Duplicate assignment attempts", "duplicate_assignment_attempts"),
    ("Duplicate active assignments", "duplicate_active_assignments"),
    ("Nearest versus ETA disagreements", "strategy_disagreements"),
    ("Throughput, requests per second", "throughput_rps"),
    ("p50 dispatch latency, ms", "p50_latency_ms"),
    ("p95 dispatch latency, ms", "p95_latency_ms"),
)

def render_report(results: dict[str, dict], requests: int, drivers: int) -> str:
    rows = [
        f"| {label} | {results['unsafe'][key]} | {results['atomic'][key]} |"
        for label, key in METRICS
    ]
    return "\n".join(
        [
            "# Ride Dispatch Reliability Report",
            "",
            "## Scope",
            "",
            "This deterministic educational simulation does not reproduce Uber's proprietary algorithms.",
            "",
            "Redis is authoritative for this experiment. Production durability, crash recovery, failover, and multi-region coordination are outside scope.",
            "",
            "## Correctness Invariant",
            "",
            "Each driver can have at most one active ride assignment.",
            "",
            "## Workload",
            "",
            f"- Rider requests: {requests}",
            f"- Drivers: {drivers}",
            "- Competing workers: 2",
            "- Matching: deterministic ETA-aware ranking",
            "",
            "## Measured Before and After",
            "",
            "| Metric | Unsafe read then write | Atomic Redis Lua |",
            "| --- | ---: | ---: |",
            *rows,
            "",
            "## Interpretation",
            "",
            "Duplicate attempts measure contention. Rejected claims are expected atomic outcomes. Neither is an invariant violation.",
            "",
            "Duplicate active assignments are calculated from active ride records grouped by driver, so an overwritten driver key cannot hide two successful rides.",
            "",
            "Timing values were measured locally and are not vendor performance claims.",
            "",
            "## Environment",
            "",
            f"- Platform: {platform.platform()}",
            f"- Python: {sys.version.split()[0]}",
            "- Redis: local Docker Compose service",
            "",
        ]
    )

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--requests", type=int, required=True)
    parser.add_argument("--drivers", type=int, required=True)
    parser.add_argument("--workers", type=int, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    drivers = generate_drivers(args.drivers)
    riders = generate_riders(args.requests)
    results = compare_safety_paths(
        riders,
        drivers,
        traffic_conditions(drivers, riders),
        args.workers,
        RedisAssignmentStore(),
    )
    if results["atomic"]["duplicate_active_assignments"] != 0:
        raise SystemExit("Invariant failed: atomic duplicate active assignments detected")
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        render_report(results, args.requests, args.drivers), encoding="utf-8"
    )
    print(f"Wrote {args.output}")
    print(
        "Atomic result: "
        f"successful={results['atomic']['successful_assignments']}, "
        f"rejected={results['atomic']['rejected_claims']}, "
        f"unmatched={results['atomic']['unmatched_riders']}, "
        f"duplicate_attempts={results['atomic']['duplicate_assignment_attempts']}, "
        f"duplicate_active={results['atomic']['duplicate_active_assignments']}"
    )

if __name__ == "__main__":
    main()

How is the report protected?

The command generates deterministic fixtures before running both claim modes. It stops immediately if the atomic result contains any duplicate active assignment.

A successful run writes the Markdown report plus prints the atomic outcome counts. The throughput plus latency values come from your local execution.

• Save `scripts/benchmark.py`.

The repository README records the invariant, architecture, commands, evidence link, plus production scope boundary. It gives another engineer enough context to reproduce the experiment.

• Return to `README.md` in the repository.
• Use the complete reference below to update the project documentation.

✔️ Awesome, I've got everything!

Your README now explains how to reproduce the unsafe race, run the atomic benchmark, plus interpret the resulting metrics.

ⓧ I'd like to double check the full code

• Replace the contents of `README.md` with this complete file:

# Real-Time Ride Dispatch and Surge Pricing Engine

A deterministic local backend experiment that compares nearest-driver selection with ETA-aware matching, reproduces duplicate assignments, and fixes the race with Redis Lua.

## Correctness Invariant

Each driver can have at most one active ride assignment.

Redis is authoritative for this educational simulation. A successful claim creates both assignment directions. A rejected claim creates neither and cannot overwrite the winner.

## Architecture

~~~mermaid
flowchart LR
    F[Deterministic fixtures] --> H[H3 candidate search]
    H --> R[Nearest or ETA-aware ranking]
    R --> D[Two dispatch workers]
    D --> U[Unsafe read then write]
    D --> L[Atomic Redis Lua]
    U --> S[(Redis assignment store)]
    L --> S
    S --> I[Invariant audit]
    I --> B[Benchmark report]
    B --> A[FastAPI endpoints]
~~~

## What It Demonstrates

- H3 geographic partitioning and nearby-cell expansion
- Nearest-driver versus deterministic ETA-aware ranking
- Educational surge pricing bounded from 1.0x to 3.0x
- A forced race that creates actual duplicate active assignments
- Atomic rejection without overwriting an existing assignment
- Separate success, rejection, unmatched, contention, and invariant metrics
- Locally measured throughput and latency percentiles

## Run the Project

~~~bash
docker compose up --build -d
docker compose exec api pytest -q
~~~

Open `http://localhost:8000/docs`.

## Run the Standard Benchmark

~~~bash
docker compose exec api python -m scripts.benchmark --requests 40 --drivers 20 --workers 2 --output reports/reliability.md
~~~

## Run the 500-Request Reliability Mission

~~~bash
docker compose exec api python -m scripts.benchmark --requests 500 --drivers 250 --workers 2 --output reports/reliability.md
~~~

Measured evidence: [reports/reliability.md](reports/reliability.md)

## Metric Definitions

- Successful assignments: riders whose dispatch call returned a driver.
- Rejected claims: claims refused because a driver was already assigned.
- Unmatched riders: riders that exhausted every H3 candidate.
- Duplicate assignment attempts: repeated attempts targeting one driver.
- Duplicate active assignments: active ride records beyond one for a driver.

Rejected claims and duplicate attempts show contention. Only duplicate active assignments violate the invariant.

## Scope Boundary

This project uses deterministic locations, synthetic traffic, and a simple supply-demand multiplier. It does not reproduce or claim knowledge of Uber's proprietary algorithms.

Redis is authoritative only for this local experiment. Production durability, crash recovery, expiration, failover, fencing, regional isolation, and multi-region coordination are outside scope.

## Cleanup

~~~bash
docker compose down -v
~~~

What does the README communicate?

The architecture traces deterministic fixtures through H3 ranking, competing workers, Redis, invariant auditing, plus measured evidence.

The scope boundary keeps the reliability claim precise. It identifies production concerns that this local experiment intentionally excludes.

• Save `README.md`.

Before you run the benchmark, predict which path will record duplicate active assignments. Also predict whether the atomic path can still record rejected claims.

• Run the standard concert workload from the repository terminal:

docker compose exec api python -m scripts.benchmark --requests 40 --drivers 20 --workers 2 --output reports/reliability.md

What should the benchmark show?

The command writes `reports/reliability.md`. It prints locally measured atomic outcomes with zero duplicate active assignments.

The unsafe table records the race behavior. The atomic table can contain rejected claims because those rejections protect the invariant.

Does the benchmark stop or hang?

Confirm that both Docker services remain healthy. Check that `run_batch()` creates a new two-party barrier for each unsafe rider pair.

Use this benchmark debugging prompt to isolate the failing path.

• In the VS Code Explorer sidebar, select `reports/reliability.md`.
• Confirm that the workload lists `40` rider requests.
• Confirm that the workload lists `20` drivers.
• Compare successful assignments across the unsafe plus atomic columns.
• Compare rejected claims across the unsafe plus atomic columns.
• Compare unmatched riders across the unsafe plus atomic columns.
• Compare duplicate assignment attempts across the unsafe plus atomic columns.
• Confirm that the atomic duplicate active assignment count is `0`.

How should you read the metrics?

Duplicate assignment attempts measure repeated targeting of one driver. Rejected claims measure conflicts refused by the current assignment state.

Duplicate active assignments count extra active ride records grouped by driver. That final metric measures the invariant directly.

You now have measured evidence for both sides of the concurrency story. The unsafe path reproduces the violation while the atomic path preserves one active ride per driver.

Secret mission

Prove Reliability Under Competing Workers

Process 500 deterministic ride requests with 250 drivers through exactly two competing workers. Replace the standard report with measured evidence that the atomic path preserves one active ride per driver under sustained local contention.

Clean Up Your Resources

Clean Up Your Resources

Your local FastAPI and Redis lab runs through Docker Compose at no ongoing cost. Choose the cleanup level that matches your next move.

Resources you used:

• The api service that runs the FastAPI application.
• The redis service that stores local assignment records.
• The local Docker Compose network that connects both services.
• The redis-data named volume that stores local Redis data.

The generated reports/reliability.md report stays in the starter repository after runtime cleanup. It preserves your measured evidence from the 500-request workload.

Keep everything running

No action needed. Choose this option if you plan to keep testing the API or rerun the deterministic benchmark.

• Keep the api service running.
• Keep the redis service running.
• Use http://localhost:8000/docs to run the existing endpoints from the API documentation.
• Preserve reports/reliability.md as evidence from the completed workload.

Pause - I'll come back to this later

Pausing releases the resources used by the active processes while preserving the containers for later. Your benchmark report stays on disk.

• Stop both local services by running this command:

docker compose stop

What Does This Command Do?

This command stops both service containers. The containers remain available for a later restart.

The redis-data volume remains available. Your source files stay unchanged.

• Confirm the terminal shows both services entering a stopped state.

Services Still Running?

• Return to the terminal for the starter repository.
• Confirm Docker Desktop is running.

Help me diagnose why my local services did not stop.

Delete - I don't want to use this again

Deletion removes the local runtime created for the lab. The repository stays on disk with reports/reliability.md inside it.

Deleting Redis Data Is Permanent

Removing the volume can feel drastic. The command permanently deletes the local Redis assignment data.

Your source files remain in the repository. The measured reliability report remains there too.

• Remove the service containers and local runtime data by running this command:

docker compose down -v

What Does This Command Remove?

This command removes the api service container.

It removes the redis service container.

It removes the Compose network.

The -v flag removes the redis-data named volume.

• Confirm the terminal reports removal of both service containers.
• Confirm the terminal reports removal of the Compose network.
• Confirm the terminal reports removal of the named volume.
• Confirm reports/reliability.md remains in the starter repository.

Cleanup Did Not Complete?

• Return to the terminal for the starter repository.
• Confirm Docker Desktop is running.

Help me diagnose why the local project resources were not removed.

Nice Work!

Nice Work!

Excellent work! You built a deterministic ride-dispatch experiment that exposes a concurrency failure before proving the Redis-backed fix under measured contention.

You've learned how to:

• Built deterministic H3 candidate search that narrows each rider's driver pool to nearby cells. Compared nearest-driver selection with ETA-aware ranking. Calculated regional surge values bounded from 1.0x to 3.0x.
• Reproduced an actual duplicate active assignment with a forced two-worker race. Enforced one active ride per driver with atomic Redis Lua claims.
• Generated a measured before-and-after reliability report with outcome counts. Added locally measured throughput plus p50 and p95 latency percentiles. Kept contention metrics distinct from invariant violations.
• Secret Mission: Processed 500 deterministic ride requests through exactly two workers. Proved the atomic path produced zero duplicate active assignments.

Ready to quiz yourself?