Support submissions across multiple gunicorn workers

biosimdb_interface/form/workflows.py

@@ -2,13 +2,14 @@
 
 from __future__ import annotations
 
+import os
 import shutil
 import threading
 import time
 import uuid
 from dataclasses import dataclass, field
 
-from .utils import make_upload_tmpdir
+from flask import current_app
 
 
 class WorkflowNotFound(Exception):
@@ -42,19 +43,24 @@ def _validate_id(workflow_id: str | None) -> str:
         except (ValueError, TypeError, AttributeError) as exc:
             raise WorkflowNotFound("Invalid upload session.") from exc
 
+    @staticmethod
+    def _dir_for(workflow_id: str) -> str:
+        """Deterministic directory so every gunicorn worker finds the same files."""
+        root = current_app.config["UPLOAD_FOLDER"]
+        os.makedirs(root, exist_ok=True)
+        return os.path.join(root, f"biosimdb_submission_{workflow_id}")
+
     def reset(self, workflow_id: str) -> Workflow:
         """Delete only this workflow's old directory and create a new one."""
         workflow_id = self._validate_id(workflow_id)
 
         with self._lock:
-            old = self._workflows.pop(workflow_id, None)
-            if old:
-                shutil.rmtree(old.tmpdir, ignore_errors=True)
-
-            workflow = Workflow(
-                workflow_id=workflow_id,
-                tmpdir=make_upload_tmpdir("biosimdb_submission_"),
-            )
+            self._workflows.pop(workflow_id, None)
+            tmpdir = self._dir_for(workflow_id)
+            shutil.rmtree(tmpdir, ignore_errors=True)
+            os.makedirs(tmpdir)
+
+            workflow = Workflow(workflow_id=workflow_id, tmpdir=tmpdir)
             self._workflows[workflow_id] = workflow
             return workflow
 
@@ -64,6 +70,18 @@ def get(self, workflow_id: str | None) -> Workflow:
 
         with self._lock:
             workflow = self._workflows.get(workflow_id)
+            if not workflow:
+                # Created by another worker process: recover from disk.
+                tmpdir = self._dir_for(workflow_id)
+                if os.path.isdir(tmpdir):
+                    mtime = os.path.getmtime(tmpdir)
+                    workflow = Workflow(
+                        workflow_id=workflow_id,
+                        tmpdir=tmpdir,
+                        created_at=mtime,
+                        updated_at=mtime,
+                    )
+                    self._workflows[workflow_id] = workflow
             if not workflow:
                 raise WorkflowNotFound(
                     "This upload session no longer exists. Please extract again."
@@ -76,6 +94,10 @@ def get(self, workflow_id: str | None) -> Workflow:
                 )
 
             workflow.updated_at = time.time()
+            try:
+                os.utime(workflow.tmpdir)  # share activity with other workers
+            except OSError:
+                pass
             return workflow
 
     def delete(self, workflow_id: str | None) -> None:

docker/Dockerfile

@@ -92,4 +92,6 @@ ENV PORT=${PORT} \
 
 EXPOSE ${PORT}
 
-CMD ["sh", "-c", "gunicorn --bind 0.0.0.0:$PORT --timeout 0 'biosimdb_interface:create_app()'"]
+CMD ["sh", "-c", "gunicorn --bind 0.0.0.0:$PORT --worker-class gthread --workers 2 --threads 4 --timeout 0 'biosimdb_interface:create_app()'"]
+
+# CMD ["sh", "-c", "gunicorn --bind 0.0.0.0:$PORT --worker-class gthread --workers 2 --threads 4 --timeout 0 --log-level debug --access-logfile - --error-logfile - --capture-output 'biosimdb_interface:create_app()'"]

tests/test_form/test_workflows.py

@@ -0,0 +1,66 @@
+import os
+import uuid
+
+import pytest
+
+from biosimdb_interface.form.workflows import WorkflowNotFound, WorkflowStore
+
+
+@pytest.fixture
+def store(app, tmp_path):
+    app.config["UPLOAD_FOLDER"] = str(tmp_path)
+    with app.app_context():
+        yield WorkflowStore(ttl_seconds=60)
+
+
+def test_reset_creates_directory(store):
+    workflow_id = str(uuid.uuid4())
+    workflow = store.reset(workflow_id)
+    assert os.path.isdir(workflow.tmpdir)
+    assert workflow_id in workflow.tmpdir
+
+
+def test_invalid_id_raises(store):
+    with pytest.raises(WorkflowNotFound):
+        store.get("not-a-uuid")
+
+
+def test_unknown_id_raises(store):
+    with pytest.raises(WorkflowNotFound):
+        store.get(str(uuid.uuid4()))
+
+
+def test_get_recovers_workflow_created_by_another_worker(store):
+    workflow_id = str(uuid.uuid4())
+    created = store.reset(workflow_id)
+
+    other_worker = WorkflowStore(ttl_seconds=60)
+    recovered = other_worker.get(workflow_id)
+
+    assert recovered.tmpdir == created.tmpdir
+
+
+def test_expired_workflow_is_deleted(store):
+    workflow_id = str(uuid.uuid4())
+    workflow = store.reset(workflow_id)
+    workflow.updated_at -= 120
+
+    with pytest.raises(WorkflowNotFound):
+        store.get(workflow_id)
+    assert not os.path.exists(workflow.tmpdir)
+
+
+def test_delete_removes_directory(store):
+    workflow = store.reset(str(uuid.uuid4()))
+    store.delete(workflow.workflow_id)
+    assert not os.path.exists(workflow.tmpdir)
+
+
+def test_cleanup_expired_removes_only_expired(store):
+    old = store.reset(str(uuid.uuid4()))
+    new = store.reset(str(uuid.uuid4()))
+    old.updated_at -= 120
+
+    assert store.cleanup_expired() == 1
+    assert not os.path.exists(old.tmpdir)
+    assert os.path.isdir(new.tmpdir)