Serialize per-submission ops via a row lock

submit_ce/api/file_store.py

@@ -97,6 +97,24 @@ def store_source_package(self, submission_id: str, content: SubmitFile, chunk_si
         Returns `list[FileStatus]`"""
         pass
 
+    @abstractmethod
+    def rebuild_source_package(self, submission_id: str) -> None:
+        """Regenerate the canonical source tar.gz from the current
+        contents of `src/`.
+
+        Callers must hold the per-submission lock for this submission
+        (see `SubmitApi.lock_submission`) — either by going through
+        `api.save()` (which already holds the row lock) or by wrapping
+        the call in `with api.lock_submission(submission_id):`.
+        Without that, a concurrent upload could land between this
+        rebuild and the downstream consumer of the package.
+
+        Used by `CompileApiService` before posting to tex2pdf-api so
+        that the package handed to the compiler reflects the live
+        `src/`, not whatever `store_source_package` last wrote.
+        """
+        pass
+
     @abstractmethod
     def get_source_package_checksum(self, submission_id: str) -> str:
         """Get the checksum of the source package for a submission."""

submit_ce/api/submit.py

@@ -60,8 +60,9 @@
 """
 
 from abc import ABC, abstractmethod
+from contextlib import AbstractContextManager
 from datetime import datetime
-from typing import Tuple, List, Optional
+from typing import Tuple, List, Optional, Union
 
 from submit_ce.api.compile_service import CompileService
 from submit_ce.domain import Submission, Event, License
@@ -154,6 +155,27 @@ def get_file_store(self) -> SubmissionFileStore:
         """
         ...
 
+    @abstractmethod
+    def lock_submission(
+        self, submission_id: Union[int, str]
+    ) -> AbstractContextManager[None]:
+        """Acquire an exclusive lock on a submission for the duration of
+        the `with` block.
+
+        Use this from controller code paths that mutate submission state
+        directly via the file store (e.g. review.py) and therefore
+        bypass `save()`. Events dispatched through `save()` already run
+        under this lock and must NOT re-enter it from a separate
+        session.
+
+        Raises
+        ------
+        :class:`.SubmissionLocked`
+            If the lock cannot be acquired within the configured wait
+            time (e.g. MySQL `innodb_lock_wait_timeout`).
+        """
+        ...
+
     @abstractmethod
     def get_compiler(self) -> CompileService:
         """

submit_ce/domain/exceptions.py

@@ -28,5 +28,21 @@ class SaveError(RuntimeError):
     """Failed to persist event state."""
 
 
+class SubmissionLocked(RuntimeError):
+    """The submission row is locked by another in-flight operation.
+
+    Intentionally NOT a subclass of `SaveError` so that the existing
+    `except SaveError` handlers in UI controllers do not swallow it
+    and convert it into HTTP 500. A dedicated Flask error handler
+    surfaces this as HTTP 409.
+    """
+
+    def __init__(self, submission_id: int | str | None = None) -> None:
+        self.submission_id = submission_id
+        super().__init__(
+            f"submission {submission_id} is locked by another operation"
+        )
+
+
 class NothingToDo(RuntimeError):
     """There is nothing to do."""

submit_ce/implementations/__init__.py

@@ -1,5 +1,7 @@
+from contextlib import contextmanager
 from datetime import datetime
-from typing import Optional, Tuple, List, IO
+from io import BytesIO
+from typing import Iterator, Optional, Tuple, List, IO, Union
 from pathlib import Path
 
 from arxiv.files import FileObj
@@ -82,6 +84,10 @@
     def store_source_package(self, submission_id: str, content: SubmitFile, chunk_size: int) -> list[FileStatus]:
         raise RuntimeError("Not stored, this is from a NullFileStore")
 
+    def rebuild_source_package(self, submission_id: str) -> None:
+        # NullFileStore has nothing to rebuild from.
+        pass
+
     def get_source_package_checksum(self, submission_id: str) -> str:
         return ""
 
@@ -252,3 +258,8 @@
             submission = event.apply(submission)
 
         return submission, events
+
+    @contextmanager
+    def lock_submission(self, submission_id: Union[int, str]) -> Iterator[None]:
+        # NullImplementation has no real persistence; the lock is a no-op.
+        yield

submit_ce/implementations/compile/compile_api_service.py

@@ -107,15 +107,18 @@ def start_preflight(self,
            &dest=gs://arxiv-sync-test-01/api-test/junk.json'
         '''
 
-        # This tgz may be older than the files in the src dir, 
-        #   since submit 2.0 does not yet regenerate after file changes.
-        # We could:
-        #   - update tex2pdf-api to build from the src dir
-        #   - update tex2pdf-api to support rezip
-        #   - check file dates and rezip locally, maybe on user request
-        source_path = current_app.api.get_file_store().get_full_source_package_path(submission.submission_id)
-
-        preflight_path = current_app.api.get_file_store().get_full_preflight_package_path(submission.submission_id)
+        # Rebuild the source tar from the live src/ so the package
+        # handed to tex2pdf-api reflects the most recent uploads /
+        # deletes, not whatever store_source_package last wrote.
+        # We are inside event.execute(), which runs under the row lock
+        # taken by api.save()'s _load(..., lock_row=True); no concurrent
+        # upload can mutate src/ between the rebuild and tex2pdf-api
+        # fetching the package.
+        file_store = current_app.api.get_file_store()
+        file_store.rebuild_source_package(submission.submission_id)
+        source_path = file_store.get_full_source_package_path(submission.submission_id)
+
+        preflight_path = file_store.get_full_preflight_package_path(submission.submission_id)
 
         query_params = {
             'source': source_path,

submit_ce/implementations/file_store/gs_file_store.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from pathlib import Path
 from typing import IO, List, Optional
+import gzip
 import json
 import io
 import logging
@@ -134,6 +135,51 @@ def store_source_file(self,
         blob.reload()
         return self._blob_to_file_status(submission_id, blob)
 
+    @override
+    def rebuild_source_package(self, submission_id: str) -> None:
+        """Regenerate `{submission_id}.tar.gz` from the current `src/`.
+
+        Caller must hold the per-submission lock. Output is
+        byte-deterministic for identical input: members are sorted by
+        GCS object name, gzip header carries mtime=0, and TarInfo
+        fields other than name/size are zeroed.
+        """
+        src_dir = self._source_path(submission_id)
+        list_prefix = str(src_dir).rstrip("/") + "/"
+        package_path = str(self._source_package_path(submission_id))
+        package_blob = self.bucket.blob(package_path)
+
+        blobs = sorted(
+            (
+                b
+                for b in self.bucket.client.list_blobs(
+                    self.bucket, prefix=list_prefix
+                )
+                if not b.name.endswith("/") and b.size is not None
+            ),
+            key=lambda b: b.name,
+        )
+
+        buf = io.BytesIO()
+        # gzip mtime=0 + sorted listing + zeroed TarInfo fields keep the
+        # archive byte-stable across rebuilds of identical input.
+        with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz, \
+             tarfile.open(fileobj=gz, mode="w") as tar:
+            for blob in blobs:
+                relname = blob.name[len(list_prefix):]
+                info = tarfile.TarInfo(name=relname)
+                info.size = blob.size
+                info.mtime = 0
+                info.uid = info.gid = 0
+                info.uname = info.gname = ""
+                info.mode = 0o644
+                data = blob.download_as_bytes()
+                tar.addfile(info, io.BytesIO(data))
+        buf.seek(0)
+        package_blob.upload_from_file(
+            buf, content_type="application/gzip"
+        )
+
     @override
     def store_source_package(self,
                      submission_id: str,

submit_ce/implementations/file_store/tests/test_gs_file_store.py

@@ -255,3 +255,75 @@ def test_artifact_delete(store, sub_id, upload_artifact, resource, content):
     upload_artifact(sub_id, resource, content)
     getattr(store, f'delete_{resource}')(sub_id)
     assert not getattr(store, f'does_{resource}_exist')(sub_id)
+
+
+# ---------------------------------------------------------------------------
+# rebuild_source_package: deterministic tar built from live src/
+# ---------------------------------------------------------------------------
+
+def _read_package(store, sub_id) -> dict[str, bytes]:
+    """Download the canonical source tar.gz and return {member: bytes}."""
+    blob = store.bucket.blob(str(store._source_package_path(sub_id)))
+    raw = blob.download_as_bytes()
+    members: dict[str, bytes] = {}
+    with tarfile.open(fileobj=BytesIO(raw), mode="r:gz") as tar:
+        for m in tar.getmembers():
+            f = tar.extractfile(m)
+            members[m.name] = f.read() if f else b""
+    return members
+
+
+def test_rebuild_source_package_contains_live_src(store, sub_id):
+    store.store_source_file(sub_id, FakeFile("a.tex", b"AAA"), chunk_size=4096)
+    store.store_source_file(sub_id, FakeFile("b.tex", b"BBB"), chunk_size=4096)
+
+    store.rebuild_source_package(sub_id)
+
+    members = _read_package(store, sub_id)
+    assert members == {"a.tex": b"AAA", "b.tex": b"BBB"}
+
+
+def test_rebuild_source_package_reflects_subsequent_changes(store, sub_id):
+    store.store_source_file(sub_id, FakeFile("a.tex", b"AAA"), chunk_size=4096)
+    store.rebuild_source_package(sub_id)
+    assert _read_package(store, sub_id) == {"a.tex": b"AAA"}
+
+    # Add a file, then a delete; rebuild must reflect both.
+    store.store_source_file(sub_id, FakeFile("c.tex", b"CCC"), chunk_size=4096)
+    store.delete_source_file(sub_id, "a.tex")
+    store.rebuild_source_package(sub_id)
+
+    assert _read_package(store, sub_id) == {"c.tex": b"CCC"}
+
+
+def test_rebuild_source_package_is_byte_deterministic(store, sub_id):
+    store.store_source_file(sub_id, FakeFile("a.tex", b"AAA"), chunk_size=4096)
+    store.store_source_file(sub_id, FakeFile("b.tex", b"BBB"), chunk_size=4096)
+
+    store.rebuild_source_package(sub_id)
+    pkg_blob = store.bucket.blob(str(store._source_package_path(sub_id)))
+    first = pkg_blob.download_as_bytes()
+
+    store.rebuild_source_package(sub_id)
+    second = pkg_blob.download_as_bytes()
+
+    assert first == second, "rebuild must be byte-deterministic for identical src/"
+
+
+def test_rebuild_source_package_no_directory_placeholders(store, sub_id):
+    # Create a nested file; the listing may include the directory blob
+    # on some uploads. Either way, the tar must not contain a "subdir/"
+    # entry with size=0 — that would be a directory placeholder, and
+    # extracting it can break downstream consumers.
+    store.store_source_file(
+        sub_id, FakeFile("subdir/nested.tex", b"NESTED"), chunk_size=4096
+    )
+    store.rebuild_source_package(sub_id)
+
+    blob = store.bucket.blob(str(store._source_package_path(sub_id)))
+    raw = blob.download_as_bytes()
+    with tarfile.open(fileobj=BytesIO(raw), mode="r:gz") as tar:
+        names = [m.name for m in tar.getmembers()]
+        for m in tar.getmembers():
+            assert not m.name.endswith("/"), f"unexpected directory entry: {m.name!r}"
+    assert "subdir/nested.tex" in names

submit_ce/implementations/legacy_implementation/__init__.py

@@ -1,6 +1,7 @@
 import logging
+from contextlib import contextmanager
 from datetime import datetime, UTC
-from typing import Optional, List, Tuple, Callable
+from typing import Iterator, Optional, List, Tuple, Callable, Union
 from sqlalchemy.exc import OperationalError, ProgrammingError
 from typing_extensions import override
 
@@ -24,11 +25,25 @@
 from ...domain.event.base import Event, EventWithSideEffect
 from ...domain.util import get_tzaware_utc_now
 
-from ...domain.event import CreateSubmission
-from ...domain.exceptions import NoSuchSubmission, NothingToDo
+from ...domain.event import CreateSubmission, UploadFiles
+from ...domain.exceptions import NoSuchSubmission, NothingToDo, SubmissionLocked
 from . import db
 
 
+# MySQL error code for ER_LOCK_WAIT_TIMEOUT. Surfaces when an InnoDB
+# row lock cannot be acquired within `innodb_lock_wait_timeout`.
+_MYSQL_LOCK_WAIT_TIMEOUT = 1205
+
+
+def _is_lock_wait_timeout(exc: OperationalError) -> bool:
+    """True if the OperationalError originates from MySQL errno 1205."""
+    orig = getattr(exc, "orig", None)
+    if orig is None:
+        return False
+    args = getattr(orig, "args", ())
+    return bool(args) and args[0] == _MYSQL_LOCK_WAIT_TIMEOUT
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -99,12 +114,50 @@
         stmt = select(Submission).where(Submission.submission_id == int(submission_id))
         if lock_row:  # row will be locked until .commit() or use .flush() to get auto inc ids without unlocking
             stmt = stmt.with_for_update()
-        submission = session.scalars(stmt).first()
+        try:
+            submission = session.scalars(stmt).first()
+        except OperationalError as exc:
+            if lock_row and _is_lock_wait_timeout(exc):
+                raise SubmissionLocked(submission_id) from exc
+            raise
         if not submission:
             raise NoSuchSubmission()
         else:
             return (to_submission(submission), db.get_events(session, submission_id))
 
+    @override
+    @contextmanager
+    def lock_submission(self, submission_id: Union[int, str]) -> Iterator[None]:
+        """Acquire SELECT ... FOR UPDATE on the submission row for the
+        duration of the `with` block.
+
+        Use only from controller code paths that mutate submission state
+        directly (e.g. review.py); paths going through `save()` already
+        hold this lock and must not re-enter from a separate session.
+        """
+        with self.get_session() as session:
+            try:
+                # `.unique()` is required because the `Submission` ORM has
+                # joined eager loads on collections; without it
+                # `scalar_one()` raises InvalidRequestError. Same pattern
+                # as `load_submissions_for_user`.
+                session.execute(
+                    select(Submission)
+                        .where(Submission.submission_id == int(submission_id))
+                        .with_for_update()
+                ).unique().scalar_one()
+            except OperationalError as exc:
+                session.rollback()
+                if _is_lock_wait_timeout(exc):
+                    raise SubmissionLocked(submission_id) from exc
+                raise
+            try:
+                yield
+                session.commit()
+            except Exception:
+                session.rollback()
+                raise
+
 
     @override
     def save(self, *events: Event, submission_id: Optional[str] = None) -> Tuple[Submission, List[Event]]: