Support single/split deployments, optimize image processing, and expand test coverage

.env.example

@@ -1,5 +1,12 @@
 # Example .env file for analytics_service
 
+# Docker Compose deployment mode toggle — read automatically by the `docker
+# compose` CLI (not the app itself). "split" = 3 separate containers
+# (analytics-web/consumer/worker, fault-isolated, default). "single" = one
+# container running all three as separate processes (see run_all.sh).
+# Only one should be active at a time — both bind host port 8000.
+COMPOSE_PROFILES=split
+
 # Logging Configuration
 LOG_DIR=logs
 LOG_LEVEL=INFO
@@ -18,6 +25,15 @@ ENVIRONMENT=development
 # Database Configuration
 DATABASE_URL=postgresql://postgres:postgres@localhost:5432/temporal
 RESET_DB=false
+# asyncpg pool size. Must stay comfortably UNDER your Postgres instance's own
+# max_connections (with headroom for other clients/replicas) — it is NOT a
+# "raise it for more throughput" knob. Load-tested: 10 stalls hard around
+# ~150-200 concurrent events, but 100 against a 100-connection Postgres
+# instance caused mass "sorry, too many clients already" failures — worse
+# than the stall. See WORKER_MAX_CONCURRENT_ACTIVITIES below for the setting
+# that actually protects this pool from an unpredictable event burst.
+DATABASE_POOL_MIN_SIZE=10
+DATABASE_POOL_MAX_SIZE=50
 
 # Orchestration Mode: 'real-time' or 'batch'
 PROCESSING_MODE=real-time
@@ -27,6 +43,19 @@ BATCH_SIZE=100
 # Temporal Configuration
 TEMPORAL_HOST=localhost:7233
 TEMPORAL_QUEUE=analytics-processing-queue
+# Caps concurrent activity execution regardless of event burst size — the
+# actual protection for the DB pool. Keep at or below DATABASE_POOL_MAX_SIZE.
+WORKER_MAX_CONCURRENT_ACTIVITIES=40
+
+# Image processing (deface_blur_activity) concurrency — tune against actual
+# server specs (CPU cores, available memory). Download/upload are cheap I/O;
+# face-blur spawns a real subprocess with a fresh ONNX model load per call and
+# is the memory-expensive step — keep BLUR_CONCURRENCY_LIMIT small and sized
+# against available memory, not CPU count or WORKER_MAX_CONCURRENT_ACTIVITIES.
+# IMAGE_EXECUTOR_MAX_WORKERS defaults to max(4, cpu_count*2) if unset.
+IMAGE_EXECUTOR_MAX_WORKERS=16
+PER_SUBMISSION_IMAGE_CONCURRENCY=3
+BLUR_CONCURRENCY_LIMIT=2
 
 # API Authentication — single shared Bearer token checked via secrets.compare_digest
 # in app/api/deps.py. Required (no default) — the app will not start without it.

app/config.py

@@ -1,6 +1,7 @@
 import json
+import os
 from typing import Dict, Any, List
-from pydantic import Field, field_validator
+from pydantic import Field, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 class Settings(BaseSettings):
@@ -12,6 +13,13 @@ class Settings(BaseSettings):
 
     # Database Configuration
     DATABASE_URL: str = Field(default="postgresql://postgres:postgres@localhost:5432/temporal")
+    # asyncpg connection pool bounds — this is the real concurrency ceiling for
+    # DB-touching activities (insert/update submission, llm_logs, etc). Sized too
+    # small and high-concurrency batches (e.g. 200+ simultaneous real-time
+    # submissions) stall almost entirely on pool.acquire() rather than failing
+    # fast, since activities queue for a connection instead of erroring out.
+    DATABASE_POOL_MIN_SIZE: int = Field(default=2, gt=0)
+    DATABASE_POOL_MAX_SIZE: int = Field(default=10, gt=0)
 
     # Orchestration Mode: 'real-time' or 'batch'
     PROCESSING_MODE: str = Field(default="real-time")
@@ -33,6 +41,24 @@ class Settings(BaseSettings):
     # Temporal Configuration
     TEMPORAL_HOST: str = Field(default="localhost:7233")
     TEMPORAL_QUEUE: str = Field(default="analytics-processing-queue")
+    # Caps how many activities the worker runs simultaneously, regardless of how
+    # many workflows are started/queued — this is what actually protects the DB
+    # pool from an unpredictable event burst. Without a cap, every incoming
+    # event immediately becomes a concurrent DB-touching activity (load-tested:
+    # a 500-event burst instantly saturated a 100-connection Postgres instance).
+    # Anything beyond this limit waits safely in Temporal's own task queue
+    # instead of piling onto Postgres. Keep at or below DATABASE_POOL_MAX_SIZE.
+    WORKER_MAX_CONCURRENT_ACTIVITIES: int = Field(default=40, gt=0)
+
+    # Image processing (deface_blur_activity) concurrency — tune these against
+    # actual server specs (CPU cores, available memory), not whatever machine
+    # they were load-tested on. Download/upload are cheap I/O; face-blur spawns
+    # a real subprocess with a fresh ONNX model load every call and is the
+    # memory-expensive step — see deface_blur_activity.py for why these are
+    # treated as separate concerns rather than one concurrency number.
+    IMAGE_EXECUTOR_MAX_WORKERS: int = Field(default_factory=lambda: max(4, (os.cpu_count() or 4) * 2), gt=0)
+    PER_SUBMISSION_IMAGE_CONCURRENCY: int = Field(default=3, gt=0)
+    BLUR_CONCURRENCY_LIMIT: int = Field(default=2, gt=0)
 
     # API Authentication — single shared Bearer token, checked via
     # secrets.compare_digest in app/api/deps.py. Required (no default): the app
@@ -130,6 +156,18 @@ def validate_database_url(cls, v: str) -> str:
             return v.replace("postgresql+asyncpg://", "postgresql://")
         return v
 
+    @model_validator(mode="after")
+    def validate_pool_bounds(self) -> "Settings":
+        # Without this, an inconsistent env config loads without error and only
+        # fails later at asyncpg.create_pool() with a generic ValueError that
+        # doesn't name the misconfigured variables.
+        if self.DATABASE_POOL_MIN_SIZE > self.DATABASE_POOL_MAX_SIZE:
+            raise ValueError(
+                f"DATABASE_POOL_MIN_SIZE ({self.DATABASE_POOL_MIN_SIZE}) must not exceed "
+                f"DATABASE_POOL_MAX_SIZE ({self.DATABASE_POOL_MAX_SIZE})."
+            )
+        return self
+
     @field_validator("PROCESS_CONFIG_STORY", "PROCESS_CONFIG_DISCUSSION")
     @classmethod
     def validate_process_config_json(cls, v: str, info) -> str:

app/database/db.py

@@ -27,32 +27,43 @@ async def initialize_schema(self) -> None:
             raise RuntimeError("Database pool is not initialized. Call connect() first.")
 
         async with self.pool.acquire() as conn:
-            if settings.RESET_DB:
-                if settings.ENVIRONMENT.lower().strip() == "development":
-                    await conn.execute("DROP SCHEMA IF EXISTS public CASCADE;")
-                    await conn.execute("CREATE SCHEMA public;")
-                    logger.warning("Database schema reset requested; dropped and recreated public schema.")
-                else:
-                    logger.error(
-                        f"RESET_DB=True was requested but ENVIRONMENT={settings.ENVIRONMENT!r} is not "
-                        "'development' — refusing to drop the schema. Set ENVIRONMENT=development if "
-                        "this is intentional."
-                    )
-
-            schema_sql = SCHEMA_FILE.read_text(encoding="utf-8")
-            schema_sql = schema_sql.replace("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ")
-            schema_sql = schema_sql.replace("CREATE INDEX ", "CREATE INDEX IF NOT EXISTS ")
-            await conn.execute(schema_sql)
-
-            # Always run the seed script to keep prompts in sync with seed_prompts.sql
-            seed_sql = SEED_PROMPTS_FILE.read_text(encoding="utf-8")
-            await conn.execute(seed_sql)
-
-            # Always run the themes seed script to seed initial approved taxonomies
-            seed_themes_sql = SEED_THEMES_FILE.read_text(encoding="utf-8")
-            await conn.execute(seed_themes_sql)
-
-            logger.info("Database schema initialized successfully.")
+            # web/consumer/worker each run as separate processes (run_all.sh) and
+            # every one of them calls connect()/initialize_schema() independently
+            # at startup — the asyncio.Lock below only serializes coroutines within
+            # one process, so without a cross-process lock they race on the same
+            # CREATE TABLE/TYPE DDL and one hits a Postgres catalog collision (e.g.
+            # duplicate key on pg_type) even with "IF NOT EXISTS", since the
+            # existence check and creation aren't atomic across concurrent sessions.
+            await conn.execute("SELECT pg_advisory_lock(727384910)")
+            try:
+                if settings.RESET_DB:
+                    if settings.ENVIRONMENT.lower().strip() == "development":
+                        await conn.execute("DROP SCHEMA IF EXISTS public CASCADE;")
+                        await conn.execute("CREATE SCHEMA public;")
+                        logger.warning("Database schema reset requested; dropped and recreated public schema.")
+                    else:
+                        logger.error(
+                            f"RESET_DB=True was requested but ENVIRONMENT={settings.ENVIRONMENT!r} is not "
+                            "'development' — refusing to drop the schema. Set ENVIRONMENT=development if "
+                            "this is intentional."
+                        )
+
+                schema_sql = SCHEMA_FILE.read_text(encoding="utf-8")
+                schema_sql = schema_sql.replace("CREATE TABLE ", "CREATE TABLE IF NOT EXISTS ")
+                schema_sql = schema_sql.replace("CREATE INDEX ", "CREATE INDEX IF NOT EXISTS ")
+                await conn.execute(schema_sql)
+
+                # Always run the seed script to keep prompts in sync with seed_prompts.sql
+                seed_sql = SEED_PROMPTS_FILE.read_text(encoding="utf-8")
+                await conn.execute(seed_sql)
+
+                # Always run the themes seed script to seed initial approved taxonomies
+                seed_themes_sql = SEED_THEMES_FILE.read_text(encoding="utf-8")
+                await conn.execute(seed_themes_sql)
+
+                logger.info("Database schema initialized successfully.")
+            finally:
+                await conn.execute("SELECT pg_advisory_unlock(727384910)")
 
     async def connect(self) -> None:
         """
@@ -71,8 +82,8 @@ async def connect(self) -> None:
             try:
                 self.pool = await asyncpg.create_pool(
                     dsn=settings.DATABASE_URL,
-                    min_size=2,
-                    max_size=10
+                    min_size=settings.DATABASE_POOL_MIN_SIZE,
+                    max_size=settings.DATABASE_POOL_MAX_SIZE
                 )
                 await self.initialize_schema()
                 logger.info("Database connection pool established successfully.")

app/services/classifier.py

@@ -5,6 +5,7 @@
 and approved themes, then computes cosine similarity to find the best match.
 """
 import logging
+import threading
 from typing import Dict, Any, List, Optional, Tuple
 
 import numpy as np
@@ -17,16 +18,32 @@
 
 # Module-level model cache — loaded once per worker process
 _model: Optional[SentenceTransformer] = None
+# Real OS-thread lock, not asyncio.Lock — this is called from separate threads
+# via asyncio.to_thread (build_theme_embeddings/get_theme_similarities), not
+# concurrently within one event loop.
+_model_lock = threading.Lock()
 
 
 def _get_model() -> SentenceTransformer:
-    """Lazily load the sentence transformer model (cached at module level)."""
+    """
+    Lazily load the sentence transformer model (cached at module level).
+    Thread-safe: double-checked locking, since multiple concurrent activities
+    call this from separate OS threads. Without the lock, several threads can
+    all see _model is None at once and each construct their own
+    SentenceTransformer concurrently — that's not safe (load-tested: it
+    corrupts PyTorch's internal state with "NotImplementedError: Cannot copy
+    out of meta tensor; no data" under concurrent load). The un-locked fast
+    path below keeps the common case (already loaded) lock-free.
+    """
     global _model
-    if _model is None:
-        model_name = settings.EMBEDDING_MODEL_NAME
-        logger.info(f"Loading SentenceTransformer model '{model_name}'...")
-        _model = SentenceTransformer(model_name)
-        logger.info(f"SentenceTransformer model '{model_name}' loaded successfully.")
+    if _model is not None:
+        return _model
+    with _model_lock:
+        if _model is None:
+            model_name = settings.EMBEDDING_MODEL_NAME
+            logger.info(f"Loading SentenceTransformer model '{model_name}'...")
+            _model = SentenceTransformer(model_name)
+            logger.info(f"SentenceTransformer model '{model_name}' loaded successfully.")
     return _model
 
 

app/temporal/csv_processing_activity.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import logging
+import threading
 import uuid
 from datetime import datetime
 from typing import Any, Dict, List, Optional
@@ -18,16 +19,25 @@
 logger = logging.getLogger("analytics_service.temporal.csv_processing_activity")
 
 _producer: Optional[Producer] = None
+# Real OS-thread lock, not asyncio.Lock — _get_producer() is called from
+# separate threads via asyncio.to_thread (_push_rows_sync), not concurrently
+# within one event loop. Without this, concurrent CsvProcessingWorkflow child
+# workflows could each construct their own Producer at once (see the identical
+# fix + reasoning in app/services/classifier.py's _get_model()).
+_producer_lock = threading.Lock()
 
 
 def _get_producer() -> Producer:
     global _producer
-    if _producer is None:
-        _producer = Producer({
-            "bootstrap.servers": settings.KAFKA_BOOTSTRAP_SERVERS,
-            "acks": "all",
-            "enable.idempotence": True,
-        })
+    if _producer is not None:
+        return _producer
+    with _producer_lock:
+        if _producer is None:
+            _producer = Producer({
+                "bootstrap.servers": settings.KAFKA_BOOTSTRAP_SERVERS,
+                "acks": "all",
+                "enable.idempotence": True,
+            })
     return _producer