feat(tests): add optional PostgreSQL integration test target

docker-compose.test.yml

@@ -1,17 +1,45 @@
+# Test Infrastructure — PostgreSQL + Redis for Integration Tests
+#
+# Usage:
+#   docker-compose -f docker-compose.test.yml up -d
+#   TEST_DATABASE_URL=postgresql+asyncpg://test_user:test_pass@localhost:5433/test_db \
+#     REDIS_URL=redis://localhost:6380/0 \
+#     pytest tests/ -v
+#   docker-compose -f docker-compose.test.yml down -v
+
 services:
-  db:
-    image: postgres:16-alpine
+  test-db:
+    image: postgres:15-alpine
+    container_name: ytprocessor-test-db
     environment:
-      POSTGRES_DB: test_db
       POSTGRES_USER: test_user
       POSTGRES_PASSWORD: test_pass
+      POSTGRES_DB: test_db
+    ports:
+      - "127.0.0.1:5433:5432"
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U test_user -d test_db"]
       interval: 5s
       timeout: 5s
       retries: 5
-    networks:
-      - ytprocessor-network
+    deploy:
+      resources:
+        limits:
+          memory: 512M
+          cpus: "0.5"
 
-networks:
-  ytprocessor-network:
+  test-redis:
+    image: redis:7-alpine
+    container_name: ytprocessor-test-redis
+    ports:
+      - "127.0.0.1:6380:6379"
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+    deploy:
+      resources:
+        limits:
+          memory: 128M
+          cpus: "0.25"

pyproject.toml

@@ -202,13 +202,15 @@ dependency-groups = ["test"]
 # Matrix configuration for testing different Python versions and configurations
 [[tool.hatch.envs.test.matrix]]
 python = ["3.12"]
-db = ["sqlite"]
+db = ["sqlite", "postgres"]
 
 [tool.hatch.envs.test.scripts]
 unit = "python -c \"import glob,os;[os.remove(f) for f in glob.glob('.coverage.*')+['.coverage'] if os.path.exists(f)]\" && pytest tests/ --ignore=tests/test_api -v --cov=app --cov-report=xml --cov-report=term-missing -n auto"
 integration = "python -c \"import glob,os;[os.remove(f) for f in glob.glob('.coverage.*')+['.coverage'] if os.path.exists(f)]\" && pytest tests/test_api/ -v --cov=app --cov-report=xml --cov-report=term-missing -n auto"
 # Run all tests (both unit and integration)
 all = "python -c \"import glob,os;[os.remove(f) for f in glob.glob('.coverage.*')+['.coverage'] if os.path.exists(f)]\" && pytest tests/ -v --cov=app --cov-report=xml --cov-report=term-missing -n auto"
+# PostgreSQL integration tests (requires docker-compose.test.yml services)
+postgres = "python -c \"import glob,os;[os.remove(f) for f in glob.glob('.coverage.*')+['.coverage'] if os.path.exists(f)]\" && TEST_DATABASE_URL=postgresql+asyncpg://test_user:test_pass@localhost:5433/test_db pytest tests/ -v --cov=app --cov-report=xml --cov-report=term-missing -n auto"
 # Coverage scripts
 cov = "pytest tests/ -v --cov=app --cov-report=xml --cov-report=term-missing"
 "cov-report" = "coverage report --format=terminal --show-missing"

tests/conftest.py

@@ -4,13 +4,21 @@
 os.environ["TESTING"] = "1"
 os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only-not-for-production-use-32chars"
 
-# Determine unique database URL per xdist worker to avoid race conditions
-_worker_id = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
-_test_db_path = os.path.abspath(f"test_{_worker_id}.db")
-_test_db_url = f"sqlite+aiosqlite:///{_test_db_path}"
+# Support running integration tests against real PostgreSQL via docker-compose.test.yml.
+# Usage:
+#   TEST_DATABASE_URL=postgresql+asyncpg://test_user:test_pass@localhost:5433/test_db \
+#     pytest tests/ -v
+# If unset, fallback to per-worker SQLite for fast parallel unit tests.
+_test_db_url = os.environ.get("TEST_DATABASE_URL")
+_using_postgres = _test_db_url is not None
+
+if not _using_postgres:
+    # Determine unique database URL per xdist worker to avoid race conditions
+    _worker_id = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
+    _test_db_path = os.path.abspath(f"test_{_worker_id}.db")
+    _test_db_url = f"sqlite+aiosqlite:///{_test_db_path}"
 
 # Force reconfigure the database URL before any app imports
-# This ensures the app uses SQLite instead of PostgreSQL
 import app.config  # noqa: E402
 
 app.config.settings.database_url = _test_db_url
@@ -28,16 +36,18 @@
 import app.models  # noqa: E402
 from app.database import Base, get_db  # noqa: E402
 
-# Now import app - it will use the SQLite URL we set above
+# Now import app - it will use the database URL we set above
 from app.main import app as fastapi_app  # noqa: E402
 
 TEST_DATABASE_URL = _test_db_url
 
+_engine_kwargs = {"poolclass": NullPool}
+if not _using_postgres:
+    _engine_kwargs["connect_args"] = {"check_same_thread": False}
 
 test_engine = create_async_engine(
     TEST_DATABASE_URL,
-    connect_args={"check_same_thread": False},
-    poolclass=NullPool,
+    **_engine_kwargs,
 )
 
 # Use async_sessionmaker for proper async session support

tests/test_worker/test_graceful_shutdown.py

@@ -197,7 +197,7 @@ async def test_graceful_shutdown_during_brpop(self):
             patch("worker.main.stop_health_server"),
             patch("worker.main.sync_outbox_to_queue", new_callable=AsyncMock),
             patch("worker.main.cleanup_expired_jobs", new_callable=AsyncMock, return_value=0),
-            patch("worker.main.reset_stuck_jobs", new_callable=AsyncMock, return_value=0),
+            patch("worker.main.requeue_stuck_jobs", new_callable=AsyncMock, return_value=0),
             patch("worker.main.write_health_async", new_callable=AsyncMock),
             patch("worker.main.update_worker_state"),
         ):

tests/test_worker/test_main.py

@@ -293,7 +293,7 @@ async def test_main_proceeds_when_redis_and_db_connected(self):
             patch("worker.main.write_health_async", new_callable=AsyncMock),
             patch("worker.main.sync_outbox_to_queue", new_callable=AsyncMock),
             patch("worker.main.cleanup_expired_jobs", new_callable=AsyncMock, return_value=0),
-            patch("worker.main.reset_stuck_jobs", new_callable=AsyncMock, return_value=0),
+            patch("worker.main.requeue_stuck_jobs", new_callable=AsyncMock, return_value=0),
         ):
             # Should complete without raising
             await main()
@@ -330,7 +330,7 @@ async def recording_ping():
             patch("worker.main.write_health_async", new_callable=AsyncMock),
             patch("worker.main.sync_outbox_to_queue", new_callable=AsyncMock),
             patch("worker.main.cleanup_expired_jobs", new_callable=AsyncMock, return_value=0),
-            patch("worker.main.reset_stuck_jobs", new_callable=AsyncMock, return_value=0),
+            patch("worker.main.requeue_stuck_jobs", new_callable=AsyncMock, return_value=0),
         ):
             await main()
 
@@ -370,7 +370,7 @@ async def recording_execute(stmt):
             patch("worker.main.write_health_async", new_callable=AsyncMock),
             patch("worker.main.sync_outbox_to_queue", new_callable=AsyncMock),
             patch("worker.main.cleanup_expired_jobs", new_callable=AsyncMock, return_value=0),
-            patch("worker.main.reset_stuck_jobs", new_callable=AsyncMock, return_value=0),
+            patch("worker.main.requeue_stuck_jobs", new_callable=AsyncMock, return_value=0),
         ):
             await main()
 

worker/main.py

@@ -18,7 +18,8 @@
     update_worker_state,
     write_health_async,
 )
-from worker.processor import process_next_job, reset_stuck_jobs, sync_outbox_to_queue
+from worker.processor import process_next_job, sync_outbox_to_queue
+from worker.zombie_sweeper import requeue_stuck_jobs
 from worker.queue import redis_client
 
 # Initialize structured logging
@@ -110,7 +111,9 @@ async def cleanup_expired_jobs() -> int:
                     await db.delete(job)
                     cleanup_count += 1
                 except Exception as db_err:
-                    logger.warning("failed_to_delete_db_row", job_id=job.id, error=str(db_err), exc_info=True)
+                    logger.warning(
+                        "failed_to_delete_db_row", job_id=job.id, error=str(db_err), exc_info=True
+                    )
 
         # Batch commit after processing all jobs
         try:
@@ -165,6 +168,12 @@ async def main() -> None:
     cleanup_interval_minutes: int = int(os.environ.get("CLEANUP_INTERVAL_MINUTES", "5"))
     cleanup_interval = timedelta(minutes=cleanup_interval_minutes)
     last_cleanup = datetime.now(UTC) - cleanup_interval
+
+    # Outbox sync runs independently of cleanup for lower latency (default: 30s)
+    outbox_sync_interval_seconds: int = int(os.environ.get("OUTBOX_SYNC_INTERVAL_SECONDS", "30"))
+    outbox_sync_interval = timedelta(seconds=outbox_sync_interval_seconds)
+    last_outbox_sync = datetime.now(UTC) - outbox_sync_interval
+
     heartbeat_counter = 0
     heartbeat_interval = (
         10  # Write heartbeat every 10 iterations (~20 seconds, since brpop_timeout=2)
@@ -230,16 +239,26 @@ async def main() -> None:
             await asyncio.sleep(1)
 
         now = datetime.now(UTC)
+
+        # Independent outbox sync (30s default) — lower latency than cleanup
+        if now - last_outbox_sync >= outbox_sync_interval:
+            try:
+                synced = await sync_outbox_to_queue()
+                if synced > 0:
+                    logger.info("outbox_sync_completed", synced=synced)
+                last_outbox_sync = now
+            except Exception as e:
+                logger.error("outbox_sync_error", error=str(e))
+
         if now - last_cleanup >= cleanup_interval:
             try:
-                # Sync outbox to queue during cleanup (handles crash recovery)
-                await sync_outbox_to_queue()
                 cleanup_count = await cleanup_expired_jobs()
-                stuck_count = await reset_stuck_jobs(timeout_minutes=10)
+                # Zombie sweeper: requeue jobs stuck in processing (SIGKILL/OOM recovery)
+                stuck_count = await requeue_stuck_jobs(timeout_minutes=15)
                 logger.info(
                     "cleanup_cycle_completed",
                     expired_jobs_cleaned=cleanup_count,
-                    stuck_jobs_reset=stuck_count,
+                    stuck_jobs_requeued=stuck_count,
                 )
                 last_cleanup = now
                 update_worker_state(last_cleanup=last_cleanup.isoformat())

worker/zombie_sweeper.py

@@ -4,18 +4,24 @@
 It polls for jobs that have been stuck in 'PROCESSING' status for too long
 and requeues them as 'pending' instead of marking them as failed.
 
+Uses the transactional outbox pattern for crash-safe requeueing:
+the status update and outbox entry are committed atomically,
+and the outbox relay handles Redis enqueue asynchronously.
+
 Poll interval: 5 minutes
 Timeout: 15 minutes stuck in 'PROCESSING' = zombie
 """
 
+import json
+import uuid
 from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import select, update
 
 from app.database import get_async_session_factory
 from app.logging_config import get_logger
 from app.models.download_job import DownloadJob
-from worker.queue import redis_client
+from app.models.outbox import Outbox
 
 logger = get_logger(__name__)
 
@@ -29,6 +35,11 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int:
     Instead of marking as 'failed', we requeue them as 'pending' so they
     can be retried by another worker.
 
+    Uses the transactional outbox pattern to avoid the dual-write problem:
+    the status update and outbox entry are committed atomically.
+    If Redis is unavailable, the outbox relay will enqueue the job
+    when connectivity is restored.
+
     Args:
         timeout_minutes: Jobs stuck in PROCESSING for longer than this are requeued.
 
@@ -54,6 +65,18 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int:
         requeued_count = 0
         for job in stuck_jobs:
             try:
+                # Create outbox entry atomically with status update.
+                # Prevents the dual-write problem: if Redis is down,
+                # the outbox relay will enqueue when it recovers.
+                outbox_entry = Outbox(
+                    id=uuid.uuid4(),
+                    job_id=job.id,
+                    event_type="zombie_recovery",
+                    payload=json.dumps({"recovered_at": datetime.now(UTC).isoformat()}),
+                    status="pending",
+                )
+                db.add(outbox_entry)
+
                 await db.execute(
                     update(DownloadJob)
                     .where(DownloadJob.id == job.id)
@@ -62,7 +85,6 @@ async def requeue_stuck_jobs(timeout_minutes: int = 15) -> int:
                         updated_at=datetime.now(UTC),
                     )
                 )
-                await redis_client.lpush("download_queue", str(job.id))
                 requeued_count += 1
                 logger.info(
                     "zombie_job_requeued",