tests: implement concurrency and load tests for inventory and orders

.env

@@ -37,5 +37,11 @@ RATE_LIMIT_TTL_SECONDS=1
 IDEMPOTENT_KEY_LIFETIME_SEC=86400
 RESERVE_TIMEOUT_MINUTES=15
 #db_engine_layer
-POOL_SIZE=50
-MAX_OVERFLOW=100
+POOL_SIZE=200
+MAX_OVERFLOW=200
+#prometheus
+PROMETHEUS_HOST=prometheus_fairdrop
+PROMETHEUS_PORT=9090
+#grafana
+GRAFANA_HOST=grafana_fairdrop
+GRAFANA_PORT=3000

.gitignore

@@ -17,3 +17,4 @@ wheels/
 .vscode/
 
 # Local env
+.coverage

Makefile

@@ -1,10 +1,10 @@
 HOST ?= http://localhost:8080
 
-.PHONY: stress-test oversell-test
+.PHONY: stress-test oversell-test orders-test
 
 stress-test:
 	uv run locust \
-		-f locustfile.py \
+		-f load_tests/locustfile.py \
 		--headless \
 		--users 500 \
 		--spawn-rate 100 \
@@ -14,7 +14,16 @@ stress-test:
 oversell-test:
 	DB_HOST=localhost uv run python scripts/seed_oversell_product.py
 	uv run locust \
-		-f locustfile_oversell.py \
+		-f load_tests/locustfile_oversell.py \
+		--headless \
+		--users 100 \
+		--spawn-rate 50 \
+		--run-time 60s \
+		--host $(HOST)
+
+orders-test:
+	uv run locust \
+		-f load_tests/locustfile_orders.py \
 		--headless \
 		--users 100 \
 		--spawn-rate 50 \

app/services/inventory/tasks.py

@@ -27,7 +27,7 @@ async def release_expired_reservations(ctx: dict) -> None:
                 select(Reservation).with_for_update().where(Reservation.id == res_id)
             )
             reservation = res_result.scalar_one_or_none()
-            if reservation is None:
+            if reservation is None or reservation.status != OrderStatus.PENDING:
                 continue
             prod_result = await session.execute(
                 select(Product)

app/services/orders/models.py

@@ -4,7 +4,7 @@
 from uuid import UUID, uuid4
 
 from sqlalchemy import CheckConstraint, Enum, ForeignKey, Numeric, Text
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy.sql import func
 from sqlalchemy.types import DateTime, Integer, String
 
@@ -41,6 +41,10 @@ class Order(Base):
         DateTime, server_default=func.now(), onupdate=func.now()
     )
 
+    items: Mapped[list['OrderItem']] = relationship(
+        'OrderItem', back_populates='order', cascade='all, delete-orphan'
+    )
+
     __table_args__ = (
         CheckConstraint('total_amount >= 0', name='check_total_amount_non_negative'),
     )
@@ -52,6 +56,8 @@ class OrderItem(Base):
     id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
     order_id: Mapped[UUID] = mapped_column(ForeignKey('orders.id'), nullable=False)
     product_id: Mapped[UUID] = mapped_column(ForeignKey('products.id'), nullable=False)
+
+    order: Mapped['Order'] = relationship('Order', back_populates='items')
     product_name: Mapped[str] = mapped_column(String(), nullable=False)
     quantity: Mapped[int] = mapped_column(Integer, nullable=False)
     price: Mapped[Decimal] = mapped_column(

app/services/orders/service.py

@@ -60,6 +60,7 @@ async def create_order_from_reservation(
     session.add(create_order_item)
     reservation.order_id = create_order.id
     await session.commit()
+    await session.refresh(create_order, attribute_names=['items'])
     return create_order
 
 

app/shared/decorators.py

@@ -4,6 +4,7 @@
 from typing import Any
 
 from fastapi import HTTPException, Request, Response, status
+from fastapi.encoders import jsonable_encoder
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 
@@ -44,7 +45,8 @@ async def wrapper(*args: Any, **kwargs: Any) -> Response | Any:
             if isinstance(response, BaseModel):
                 json_str_to_cache = response.model_dump_json()
             else:
-                json_str_to_cache = json.dumps(response)
+                json_data = jsonable_encoder(response)
+                json_str_to_cache = json.dumps(json_data)
             await redis_client.setex(redis_key, ttl_seconds, json_str_to_cache)
             return response
 

docker-compose.yaml

@@ -95,9 +95,9 @@ services:
 
   prometheus:
     image: prom/prometheus:latest
-    container_name: prometheus_fairdrop
+    container_name: ${PROMETHEUS_HOST}
     ports:
-      - '9090:9090'
+      - '${PROMETHEUS_PORT}:9090'
     volumes:
       - ./infrastructure/prometheus/prometheus.yaml:/etc/prometheus/prometheus.yml:ro
       - fairdrop-prometheus:/prometheus
@@ -106,9 +106,9 @@ services:
 
   grafana:
     image: grafana/grafana:latest
-    container_name: grafana_fairdrop
+    container_name: ${GRAFANA_HOST}
     ports:
-      - '3000:3000'
+      - '${GRAFANA_PORT}:3000'
     volumes:
       - fairdrop-grafana:/var/lib/grafana
     depends_on:

locustfile.py → load_tests/locustfile.py

@@ -2,7 +2,7 @@
 
 from locust import between, task
 
-from locust_base import BaseUser
+from load_tests.locust_base import BaseUser
 
 TARGET_PRODUCT_ID = '5995fa75-07c7-4b55-82b7-6bfbb52948b8'
 

load_tests/locustfile_orders.py

@@ -0,0 +1,42 @@
+from http import HTTPStatus
+
+from locust import between, task
+
+from load_tests.locust_base import BaseUser
+
+TARGET_PRODUCT_ID = '5995fa75-07c7-4b55-82b7-6bfbb52948b8'
+
+
+class OrderLoadUser(BaseUser):
+    wait_time = between(1, 3)
+
+    @task
+    def reserve_and_order(self) -> None:
+        if not self.access_token:
+            return
+        with self.client.post(
+            '/api/v1/inventory/reserve',
+            headers=self.auth_headers,
+            json={'product_id': TARGET_PRODUCT_ID, 'quantity': 1},
+            catch_response=True,
+        ) as response:
+            if response.status_code not in (
+                HTTPStatus.OK,
+                HTTPStatus.CREATED,
+                HTTPStatus.CONFLICT,
+                HTTPStatus.TOO_MANY_REQUESTS,
+                HTTPStatus.BAD_REQUEST,
+            ):
+                response.failure(f'Reserve failed: {response.status_code}')
+                return
+            if response.status_code not in (HTTPStatus.OK, HTTPStatus.CREATED):
+                return
+            reservation_id = response.json()['id']
+        with self.client.post(
+            '/api/v1/orders/',
+            headers=self.auth_headers,
+            json={'reservation_id': reservation_id},
+            catch_response=True,
+        ) as response:
+            if response.status_code not in (HTTPStatus.CREATED, HTTPStatus.OK):
+                response.failure(f'Order failed: {response.status_code}')

locustfile_oversell.py → load_tests/locustfile_oversell.py

@@ -2,7 +2,7 @@
 
 from locust import between, task
 
-from locust_base import BaseUser
+from load_tests.locust_base import BaseUser
 
 OVERSELL_PRODUCT_ID = '3fe44185-589a-4703-b640-40df8d7ea67f'
 

nginx/nginx.conf

@@ -47,6 +47,7 @@ server {
         # limit_req zone=public_limit burst=50 nodelay; # uncomment if you want to use double layer white list
         # limit_req zone=b2b_limit burst=2000 nodelay; # uncomment if you want to use double layer white list
         limit_req zone=global_limit burst=50 nodelay;
+        limit_req_status 429;
         proxy_pass http://app:8000;
         proxy_set_header Host $host;
         proxy_set_header X-Real-IP $remote_addr;

pyproject.toml

@@ -32,6 +32,7 @@ dev = [
     'pre-commit>=4.5.1',
     'pytest>=9.0.2',
     'pytest-asyncio>=1.3.0',
+    'pytest-cov>=7.0.0',
     'ruff>=0.15.0',
 ]
 
@@ -55,7 +56,7 @@ python_classes = ['Test*']
 python_functions = ['test_*']
 pythonpath = ['.']
 asyncio_mode = 'auto'
-asyncio_default_fixture_loop_scope = 'session'
+asyncio_default_fixture_loop_scope = 'function'
 
 [tool.mypy]
 python_version = '3.11'

tests/test_arq_expiry.py

@@ -0,0 +1,62 @@
+import asyncio
+import datetime
+from uuid import uuid4
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from app.services.inventory.models import Product, Reservation
+from app.services.inventory.tasks import release_expired_reservations
+from app.services.orders.models import OrderStatus
+from app.services.user.models import User
+
+
+async def test_arq_concurrent_expiry_no_double_return(
+    db_engine: AsyncEngine, db_session_factory: async_sessionmaker[AsyncSession]
+) -> None:
+    async with db_session_factory() as clean_up_session:
+        await clean_up_session.execute(
+            delete(Reservation).where(Reservation.status == OrderStatus.PENDING)
+        )
+        await clean_up_session.commit()
+    async with db_session_factory() as setup_session:
+        user = User(id=uuid4(), email=f'test_{uuid4()}@mail.com', password_hash='foo')
+        product = Product(
+            id=uuid4(), name='Test Plate carrier', price=100.0, qty_available=0
+        )
+        setup_session.add(user)
+        setup_session.add(product)
+        await setup_session.commit()
+        for _ in range(10):
+            reservation = Reservation(
+                qty_reserved=1,
+                status=OrderStatus.PENDING,
+                idempotency_key=str(uuid4()),
+                expires_at=datetime.datetime.now(datetime.UTC)
+                - datetime.timedelta(minutes=60),
+                user_id=user.id,
+                product_id=product.id,
+            )
+            setup_session.add(reservation)
+        product_id = product.id
+        await setup_session.commit()
+    worker_session_factory = async_sessionmaker(
+        bind=db_engine,
+        class_=AsyncSession,
+        expire_on_commit=True,
+    )
+    ctx = {'session_maker': worker_session_factory}
+    await asyncio.gather(
+        release_expired_reservations(ctx),
+        release_expired_reservations(ctx),
+    )
+    async with db_session_factory() as session:
+        reservations = await session.execute(
+            select(Product.qty_available).where(Product.id == product_id)
+        )
+        assert reservations.scalar_one() == 10
+        reservations = await session.execute(
+            select(Reservation.status).where(Reservation.product_id == product_id)
+        )
+        statuses = reservations.scalars().all()
+        assert all(s == OrderStatus.EXPIRED for s in statuses)