Fix background job execution and CI/CD Docker test

- Root cause: _run_job created a new asyncio event loop that conflicted
  with SQLAlchemy's async engine (Future attached to different loop)
- Fix: added sync engine/session factory for background tasks (thread pool)
- Fix: explicit db.commit() before background task starts so the new
  sync connection can see the job record
- CI: Docker test now starts a postgres container instead of running
  standalone (app needs postgres for startup)
- Console: stops polling after 10 consecutive 404s (stale job IDs)
- Added psycopg2-binary for sync PostgreSQL access

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: eprifti

.github/workflows/ci.yml

@@ -87,17 +87,37 @@ jobs:
 
       - name: Test Docker image
         run: |
-          docker run -d --name test-app -p 8000:8000 predomicsapp-web:latest
+          # Start a postgres container for the app
+          docker network create test-net
+          docker run -d --name test-db --network test-net \
+            -e POSTGRES_USER=predomics \
+            -e POSTGRES_PASSWORD=predomics \
+            -e POSTGRES_DB=predomics \
+            postgres:16-alpine
+          # Wait for postgres
+          for i in $(seq 1 15); do
+            if docker exec test-db pg_isready -U predomics > /dev/null 2>&1; then
+              echo "Postgres ready on attempt $i"
+              break
+            fi
+            sleep 2
+          done
+          # Start app with postgres connection
+          docker run -d --name test-app --network test-net -p 8000:8000 \
+            -e PREDOMICS_DATABASE_URL=postgresql+asyncpg://predomics:predomics@test-db:5432/predomics \
+            predomicsapp-web:latest
           for i in $(seq 1 30); do
             if curl -sf http://localhost:8000/health; then
-              echo " Health check passed on attempt $i"
-              docker stop test-app
+              echo "Health check passed on attempt $i"
+              docker stop test-app test-db
+              docker network rm test-net
               exit 0
             fi
             echo "Waiting for app to start (attempt $i/30)..."
             sleep 2
           done
           echo "Health check failed after 30 attempts"
           docker logs test-app
-          docker stop test-app
+          docker stop test-app test-db
+          docker network rm test-net
           exit 1

backend/app/core/database.py

@@ -1,7 +1,8 @@
 """Async SQLAlchemy database setup."""
 
+from sqlalchemy import create_engine as create_sync_engine
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
-from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import DeclarativeBase, sessionmaker
 
 from .config import settings
 
@@ -10,9 +11,15 @@ class Base(DeclarativeBase):
     pass
 
 
+# Async engine — used by FastAPI request handlers
 engine = create_async_engine(settings.database_url, echo=settings.debug)
 async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
 
+# Sync engine — used by background tasks (which run in thread pools)
+_sync_url = settings.database_url.replace("+asyncpg", "").replace("+aiosqlite", "")
+sync_engine = create_sync_engine(_sync_url, echo=settings.debug)
+sync_session_factory = sessionmaker(sync_engine, expire_on_commit=False)
+
 
 async def get_db():
     """FastAPI dependency that yields a database session."""

backend/app/routers/analysis.py

@@ -9,7 +9,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..core.database import get_db, async_session_factory
+from ..core.database import get_db, sync_session_factory
 from ..core.deps import get_current_user
 from ..models.db_models import User, Project, Job
 from ..models.schemas import (
@@ -60,10 +60,11 @@ async def run_analysis(
         p = storage.get_dataset_path(project_id, ytest_dataset_id)
         ytest_path = str(p) if p else ""
 
-    # Create job record in database
+    # Create job record in database — commit immediately so the background
+    # task (which uses a separate sync DB connection) can see the job.
     job = Job(project_id=project_id, status="pending", config=config.model_dump())
     db.add(job)
-    await db.flush()
+    await db.commit()
 
     job_dir = Path(storage.settings.project_dir) / project_id / "jobs" / job.id
     job_dir.mkdir(parents=True, exist_ok=True)
@@ -86,8 +87,11 @@ async def run_analysis(
 
 
 def _run_job(job_id: str, project_id: str, param_path: str) -> None:
-    """Execute gpredomics in a subprocess so we can capture Rust log output."""
-    import asyncio
+    """Execute gpredomics in a subprocess so we can capture Rust log output.
+
+    Runs in a Starlette thread pool — uses synchronous DB sessions to avoid
+    event-loop conflicts with the async engine.
+    """
     import os
     import subprocess
     import sys
@@ -96,86 +100,75 @@ def _run_job(job_id: str, project_id: str, param_path: str) -> None:
     log_path = job_dir / "console.log"
     results_path = job_dir / "results.json"
 
-    async def _execute():
-        # Mark job as running
-        async with async_session_factory() as db:
-            result = await db.execute(select(Job).where(Job.id == job_id))
-            job = result.scalar_one()
-            job.status = "running"
-            await db.commit()
-
-        try:
-            if not engine.HAS_ENGINE:
-                # Mock mode — no subprocess needed
-                mock_results = engine._mock_results()
-                storage.save_job_result(project_id, job_id, mock_results)
-                with open(log_path, "w") as lf:
-                    lf.write("[mock] Engine not available, using mock results\n")
-
-                async with async_session_factory() as db:
-                    result = await db.execute(select(Job).where(Job.id == job_id))
-                    job = result.scalar_one()
-                    job.status = "completed"
-                    job.results_path = str(results_path)
-                    job.completed_at = datetime.now(timezone.utc)
-                    await db.commit()
-                return
-
-            # Run worker subprocess — stream output line-by-line for live console
-            worker_module = "app.services.worker"
-            env = os.environ.copy()
-            env["PYTHONUNBUFFERED"] = "1"
-
-            proc = subprocess.Popen(
-                [sys.executable, "-u", "-m", worker_module, param_path, str(results_path)],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                cwd=str(Path(__file__).resolve().parents[2]),  # backend/
-                env=env,
-            )
-
-            # Read output line-by-line and flush to log file immediately
-            with open(log_path, "w") as lf:
-                for line in proc.stdout:
-                    lf.write(line.decode("utf-8", errors="replace"))
-                    lf.flush()
-
-            proc.wait()
+    # Mark job as running
+    with sync_session_factory() as db:
+        job = db.execute(select(Job).where(Job.id == job_id)).scalar_one()
+        job.status = "running"
+        db.commit()
 
-            if proc.returncode != 0:
-                error_msg = log_path.read_text()[-500:] if log_path.exists() else "Unknown error"
-                raise RuntimeError(f"Worker exited with code {proc.returncode}: {error_msg}")
+    try:
+        if not engine.HAS_ENGINE:
+            # Mock mode — no subprocess needed
+            mock_results = engine._mock_results()
+            storage.save_job_result(project_id, job_id, mock_results)
+            with open(log_path, "w") as lf:
+                lf.write("[mock] Engine not available, using mock results\n")
 
-            async with async_session_factory() as db:
-                result = await db.execute(select(Job).where(Job.id == job_id))
-                job = result.scalar_one()
+            with sync_session_factory() as db:
+                job = db.execute(select(Job).where(Job.id == job_id)).scalar_one()
                 job.status = "completed"
                 job.results_path = str(results_path)
                 job.completed_at = datetime.now(timezone.utc)
-                await db.commit()
+                db.commit()
+            return
+
+        # Run worker subprocess — stream output line-by-line for live console
+        worker_module = "app.services.worker"
+        env = os.environ.copy()
+        env["PYTHONUNBUFFERED"] = "1"
+
+        proc = subprocess.Popen(
+            [sys.executable, "-u", "-m", worker_module, param_path, str(results_path)],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            cwd=str(Path(__file__).resolve().parents[2]),  # backend/
+            env=env,
+        )
 
-            logger.info("Job %s completed", job_id)
+        # Read output line-by-line and flush to log file immediately
+        with open(log_path, "w") as lf:
+            for line in proc.stdout:
+                lf.write(line.decode("utf-8", errors="replace"))
+                lf.flush()
 
-        except Exception as e:
-            # Append error to log
-            with open(log_path, "a") as lf:
-                lf.write(f"\n[error] {e}\n")
+        proc.wait()
 
-            async with async_session_factory() as db:
-                result = await db.execute(select(Job).where(Job.id == job_id))
-                job = result.scalar_one()
-                job.status = "failed"
-                job.error_message = str(e)
-                job.completed_at = datetime.now(timezone.utc)
-                await db.commit()
+        if proc.returncode != 0:
+            error_msg = log_path.read_text()[-500:] if log_path.exists() else "Unknown error"
+            raise RuntimeError(f"Worker exited with code {proc.returncode}: {error_msg}")
 
-            logger.error("Job %s failed: %s", job_id, e)
+        with sync_session_factory() as db:
+            job = db.execute(select(Job).where(Job.id == job_id)).scalar_one()
+            job.status = "completed"
+            job.results_path = str(results_path)
+            job.completed_at = datetime.now(timezone.utc)
+            db.commit()
 
-    loop = asyncio.new_event_loop()
-    try:
-        loop.run_until_complete(_execute())
-    finally:
-        loop.close()
+        logger.info("Job %s completed", job_id)
+
+    except Exception as e:
+        # Append error to log
+        with open(log_path, "a") as lf:
+            lf.write(f"\n[error] {e}\n")
+
+        with sync_session_factory() as db:
+            job = db.execute(select(Job).where(Job.id == job_id)).scalar_one()
+            job.status = "failed"
+            job.error_message = str(e)
+            job.completed_at = datetime.now(timezone.utc)
+            db.commit()
+
+        logger.error("Job %s failed: %s", job_id, e)
 
 
 @router.get("/{project_id}/jobs/{job_id}/logs")

backend/pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "pandas>=1.3",
     "sqlalchemy[asyncio]>=2.0",
     "asyncpg>=0.29",
+    "psycopg2-binary>=2.9",
     "python-jose[cryptography]>=3.3",
     "bcrypt>=4.0",
     "pyyaml>=6.0",

backend/tests/test_api.py

@@ -16,11 +16,12 @@
 os.environ["PREDOMICS_PROJECT_DIR"] = os.path.join(_tmp, "projects")
 os.environ["PREDOMICS_UPLOAD_DIR"] = os.path.join(_tmp, "uploads")
 os.environ["PREDOMICS_SAMPLE_DIR"] = os.path.join(_tmp, "samples")
-os.environ["PREDOMICS_DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
+_db_path = os.path.join(_tmp, "test.db")
+os.environ["PREDOMICS_DATABASE_URL"] = f"sqlite+aiosqlite:///{_db_path}"
 os.environ["PREDOMICS_SECRET_KEY"] = "test-secret-key"
 
 from app.main import app  # noqa: E402
-from app.core.database import engine, Base, get_db, async_session_factory  # noqa: E402
+from app.core.database import engine, Base, get_db, async_session_factory, sync_engine  # noqa: E402
 from app.services import engine as ml_engine  # noqa: E402
 from app.services import storage  # noqa: E402
 
@@ -30,10 +31,13 @@ async def db_session():
     """Create tables and yield a test database session."""
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
+    # Also create tables for the sync engine (used by background tasks)
+    Base.metadata.create_all(sync_engine)
     async with async_session_factory() as session:
         yield session
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.drop_all)
+    Base.metadata.drop_all(sync_engine)
 
 
 @pytest_asyncio.fixture

frontend/src/components/ConsolePanel.vue

@@ -27,9 +27,12 @@ const jobStatus = ref('pending')
 const consoleEl = ref(null)
 let pollTimer = null
 
+let errorCount = 0
+
 async function pollLogs() {
   try {
     const { data } = await axios.get(`/api/analysis/${props.projectId}/jobs/${props.jobId}/logs`)
+    errorCount = 0
     logContent.value = data.log
     jobStatus.value = data.status
 
@@ -46,7 +49,13 @@ async function pollLogs() {
       emit('failed', props.jobId)
     }
   } catch (e) {
-    console.error('Failed to poll logs:', e)
+    errorCount++
+    if (errorCount >= 10) {
+      logContent.value = '[error] Job not found. Close the console and re-launch the analysis.'
+      jobStatus.value = 'failed'
+      stopPolling()
+      emit('failed', props.jobId)
+    }
   }
 }