From 837d31fa68b29f1a60fa14ba9de33cc0b03b7732 Mon Sep 17 00:00:00 2001 From: Jas Kalayan Date: Fri, 16 Oct 2026 21:16:07 +0100 Subject: [PATCH 1/3] feat: recover submission workflows across workers --- biosimdb_interface/form/workflows.py | 40 +++++++++++++++++++++------- 1 file changed, 31 insertions(+), 9 deletions(-) diff --git a/biosimdb_interface/form/workflows.py b/biosimdb_interface/form/workflows.py index 0a68ca3..154d7b4 100644 --- a/biosimdb_interface/form/workflows.py +++ b/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: From f481743c90928ad72cb01309ab08c9023f3e8ee5 Mon Sep 17 00:00:00 2001 From: Jas Kalayan Date: Fri, 16 Oct 2026 21:17:01 +0100 Subject: [PATCH 2/3] feat: include multiple workers for Gunicorn --- docker/Dockerfile | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docker/Dockerfile b/docker/Dockerfile index 775cae7..0d4a0a6 100644 --- a/docker/Dockerfile +++ b/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()'"] From b1f4833cefa6b6841e79610407d8aee80891b3ae Mon Sep 17 00:00:00 2001 From: Jas Kalayan Date: Fri, 16 Oct 2026 21:21:23 +0100 Subject: [PATCH 3/3] tests(workflows): include further tests for workflows --- tests/test_form/test_workflows.py | 66 +++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 tests/test_form/test_workflows.py diff --git a/tests/test_form/test_workflows.py b/tests/test_form/test_workflows.py new file mode 100644 index 0000000..f1fccc7 --- /dev/null +++ b/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)