diff --git a/submit_ce/api/file_store.py b/submit_ce/api/file_store.py
index 70ed54ac..7752f4cd 100644
--- a/submit_ce/api/file_store.py
+++ b/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."""
diff --git a/submit_ce/api/submit.py b/submit_ce/api/submit.py
index 9f18c5fc..2a3058c0 100644
--- a/submit_ce/api/submit.py
+++ b/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:
         """
diff --git a/submit_ce/domain/exceptions.py b/submit_ce/domain/exceptions.py
index 9a4961ed..fba1c004 100644
--- a/submit_ce/domain/exceptions.py
+++ b/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."""
diff --git a/submit_ce/implementations/__init__.py b/submit_ce/implementations/__init__.py
index 0ec6036c..0943d27b 100644
--- a/submit_ce/implementations/__init__.py
+++ b/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_file(self, submission_id: str,
     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 @@ def save(self, *events: Event, submission_id: Optional[str] = None) -> Tuple[Sub
             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
diff --git a/submit_ce/implementations/compile/compile_api_service.py b/submit_ce/implementations/compile/compile_api_service.py
index 5d6432c8..8bc9259e 100644
--- a/submit_ce/implementations/compile/compile_api_service.py
+++ b/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,
diff --git a/submit_ce/implementations/file_store/gs_file_store.py b/submit_ce/implementations/file_store/gs_file_store.py
index 7cb1929b..584445b1 100644
--- a/submit_ce/implementations/file_store/gs_file_store.py
+++ b/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,
diff --git a/submit_ce/implementations/file_store/tests/test_gs_file_store.py b/submit_ce/implementations/file_store/tests/test_gs_file_store.py
index eacd93fb..35ae8829 100644
--- a/submit_ce/implementations/file_store/tests/test_gs_file_store.py
+++ b/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
diff --git a/submit_ce/implementations/legacy_implementation/__init__.py b/submit_ce/implementations/legacy_implementation/__init__.py
index 3967f66d..295e91f9 100644
--- a/submit_ce/implementations/legacy_implementation/__init__.py
+++ b/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 @@ def _load(self, session: SqlalchemySession, submission_id: str, lock_row: bool =
         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]]:
diff --git a/submit_ce/implementations/legacy_implementation/tests/test_lock_submission.py b/submit_ce/implementations/legacy_implementation/tests/test_lock_submission.py
new file mode 100644
index 00000000..e5bac07a
--- /dev/null
+++ b/submit_ce/implementations/legacy_implementation/tests/test_lock_submission.py
@@ -0,0 +1,234 @@
+"""Unit tests for the submission row lock plumbing.
+
+What's exercised here:
+
+- `_is_lock_wait_timeout` correctly identifies MySQL errno 1205.
+- `LegacySubmitImplementation.lock_submission(...)` enters/exits
+  cleanly for an existing submission against a SQLite test DB
+  (SQLite's `SELECT ... FOR UPDATE` is a no-op syntactically but
+  the context manager's commit/rollback/finally machinery still
+  needs to be exercised).
+- The SQLAlchemy session is rolled back when the `with` body raises.
+- `OperationalError(errno=1205)` is translated to
+  `SubmissionLocked`.
+- `NullImplementation.lock_submission(...)` is a no-op.
+- `PubsubEventSubmitImplementation.lock_submission(...)` delegates
+  to its inner_api.
+
+What is NOT exercised:
+
+- Real MySQL contention (two sessions competing for the same row)
+  and the `innodb_lock_wait_timeout` path. SQLite is the test
+  backend; real lock contention has to be verified manually or in
+  an integration environment with MySQL.
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy.exc import OperationalError
+
+from submit_ce.domain.exceptions import SubmissionLocked
+from submit_ce.implementations import NullImplementation
+from submit_ce.implementations.legacy_implementation import (
+    LegacySubmitImplementation,
+    _is_lock_wait_timeout,
+    _MYSQL_LOCK_WAIT_TIMEOUT,
+)
+from submit_ce.implementations.pubsub import PubsubEventSubmitImplementation
+
+
+# ---------------------------------------------------------------------------
+# _is_lock_wait_timeout
+# ---------------------------------------------------------------------------
+
+def _make_operational_error(errno: int) -> OperationalError:
+    orig = type("DBAPIError", (Exception,), {})()
+    orig.args = (errno, "lock wait timeout exceeded")
+    return OperationalError("SELECT ...", {}, orig)
+
+
+def test_is_lock_wait_timeout_true_for_1205():
+    exc = _make_operational_error(_MYSQL_LOCK_WAIT_TIMEOUT)
+    assert _is_lock_wait_timeout(exc) is True
+
+
+def test_is_lock_wait_timeout_false_for_other_errno():
+    exc = _make_operational_error(1234)
+    assert _is_lock_wait_timeout(exc) is False
+
+
+def test_is_lock_wait_timeout_false_when_no_orig():
+    exc = OperationalError("SELECT ...", {}, None)
+    assert _is_lock_wait_timeout(exc) is False
+
+
+def test_is_lock_wait_timeout_false_when_orig_has_no_args():
+    orig = type("Bare", (Exception,), {})()
+    exc = OperationalError("SELECT ...", {}, orig)
+    assert _is_lock_wait_timeout(exc) is False
+
+
+# ---------------------------------------------------------------------------
+# SubmissionLocked is NOT a SaveError subclass
+# ---------------------------------------------------------------------------
+
+def test_submission_locked_not_subclass_of_save_error():
+    """Critical invariant: existing `except SaveError` handlers in UI
+    controllers must NOT swallow SubmissionLocked, or it would surface
+    as 500 instead of 409."""
+    from submit_ce.domain.exceptions import SaveError
+    assert not issubclass(SubmissionLocked, SaveError)
+
+
+# ---------------------------------------------------------------------------
+# LegacySubmitImplementation.lock_submission against SQLite
+# ---------------------------------------------------------------------------
+
+def _make_legacy_impl(session_factory):
+    """Build a minimal LegacySubmitImplementation around an in-memory
+    SQLAlchemy session factory. Compiler / store don't matter for
+    lock_submission tests."""
+    return LegacySubmitImplementation(
+        store=MagicMock(),
+        compiler=MagicMock(),
+        get_session=session_factory,
+    )
+
+
+def test_lock_submission_enters_and_exits_cleanly():
+    """`lock_submission` must commit on clean exit. Verified with a
+    mock session because SQLite's SELECT ... FOR UPDATE is a no-op,
+    so a "real" integration test against the test DB would not
+    exercise the lock behaviour any more thoroughly than this mock.
+    Real MySQL contention is out of scope for unit tests."""
+    fake_session = MagicMock()
+    fake_session.execute.return_value.scalar_one.return_value = MagicMock()
+    fake_session_cm = MagicMock()
+    fake_session_cm.__enter__.return_value = fake_session
+    fake_session_cm.__exit__.return_value = None
+
+    impl = _make_legacy_impl(lambda: fake_session_cm)
+
+    with impl.lock_submission(42):
+        pass  # acquire + release with no body — must not raise.
+
+    fake_session.commit.assert_called_once()
+    fake_session.rollback.assert_not_called()
+
+
+def test_lock_submission_rolls_back_on_exception():
+    """If the `with` body raises, the session is rolled back and the
+    exception propagates."""
+    fake_session = MagicMock()
+    fake_session.execute.return_value.scalar_one.return_value = MagicMock()
+    fake_session_cm = MagicMock()
+    fake_session_cm.__enter__.return_value = fake_session
+    fake_session_cm.__exit__.return_value = None
+
+    impl = _make_legacy_impl(lambda: fake_session_cm)
+
+    sentinel = RuntimeError("body failed")
+    with pytest.raises(RuntimeError) as ei:
+        with impl.lock_submission(42):
+            raise sentinel
+    assert ei.value is sentinel
+    fake_session.rollback.assert_called_once()
+    fake_session.commit.assert_not_called()
+
+
+def test_lock_submission_translates_errno_1205_to_submission_locked():
+    """Mock the session so SELECT FOR UPDATE raises OperationalError
+    errno 1205; verify SubmissionLocked is raised."""
+    fake_session = MagicMock()
+    fake_session.execute.side_effect = _make_operational_error(
+        _MYSQL_LOCK_WAIT_TIMEOUT
+    )
+    # Make `with self.get_session() as session` work.
+    fake_session_cm = MagicMock()
+    fake_session_cm.__enter__.return_value = fake_session
+    fake_session_cm.__exit__.return_value = None
+
+    impl = _make_legacy_impl(lambda: fake_session_cm)
+
+    with pytest.raises(SubmissionLocked):
+        with impl.lock_submission(42):
+            pass
+
+    fake_session.rollback.assert_called()
+
+
+def test_lock_submission_propagates_non_1205_operational_error():
+    """A non-lock-wait OperationalError must propagate as-is."""
+    fake_session = MagicMock()
+    fake_session.execute.side_effect = _make_operational_error(2002)
+    fake_session_cm = MagicMock()
+    fake_session_cm.__enter__.return_value = fake_session
+    fake_session_cm.__exit__.return_value = None
+
+    impl = _make_legacy_impl(lambda: fake_session_cm)
+
+    with pytest.raises(OperationalError):
+        with impl.lock_submission(42):
+            pass
+
+
+def test_load_lock_row_translates_errno_1205():
+    """`_load(..., lock_row=True)` must also translate errno 1205 so
+    `api.save(...)` paths surface SubmissionLocked, not OperationalError."""
+    fake_session = MagicMock()
+    fake_session.scalars.side_effect = _make_operational_error(
+        _MYSQL_LOCK_WAIT_TIMEOUT
+    )
+    impl = _make_legacy_impl(MagicMock())
+
+    with pytest.raises(SubmissionLocked):
+        impl._load(fake_session, "42", lock_row=True)
+
+
+def test_load_without_lock_row_does_not_translate_errno_1205():
+    """Without lock_row=True, errno 1205 is not a "lock wait" scenario
+    we expect — let the original exception propagate."""
+    fake_session = MagicMock()
+    fake_session.scalars.side_effect = _make_operational_error(
+        _MYSQL_LOCK_WAIT_TIMEOUT
+    )
+    impl = _make_legacy_impl(MagicMock())
+
+    with pytest.raises(OperationalError):
+        impl._load(fake_session, "42", lock_row=False)
+
+
+# ---------------------------------------------------------------------------
+# NullImplementation: no-op context manager
+# ---------------------------------------------------------------------------
+
+def test_null_implementation_lock_submission_is_no_op():
+    # NullImplementation is currently missing `healthy()` so it can't
+    # be instantiated as the full abstract API (pre-existing issue,
+    # not introduced here). We invoke the unbound method directly to
+    # verify the no-op shape without going through `__init__`.
+    cm = NullImplementation.lock_submission(MagicMock(), 42)
+    cm.__enter__()
+    cm.__exit__(None, None, None)  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# PubsubEventSubmitImplementation: delegates
+# ---------------------------------------------------------------------------
+
+def test_pubsub_lock_submission_delegates_to_inner_api():
+    inner = MagicMock()
+    inner.lock_submission.return_value.__enter__.return_value = None
+    inner.lock_submission.return_value.__exit__.return_value = None
+
+    publisher = MagicMock()
+    impl = PubsubEventSubmitImplementation(
+        publisher=publisher, topic="t", inner_api=inner
+    )
+
+    with impl.lock_submission(42):
+        pass
+
+    inner.lock_submission.assert_called_once_with(42)
diff --git a/submit_ce/implementations/pubsub/__init__.py b/submit_ce/implementations/pubsub/__init__.py
index 6080c75d..f9a2c1ab 100644
--- a/submit_ce/implementations/pubsub/__init__.py
+++ b/submit_ce/implementations/pubsub/__init__.py
@@ -1,7 +1,8 @@
 """submit-ce API implementation that sends pubsub events."""
 
+from contextlib import AbstractContextManager
 from datetime import datetime
-from typing import Optional, Tuple, List
+from typing import Optional, Tuple, List, Union
 import logging
 
 from google.cloud import pubsub_v1
@@ -64,6 +65,12 @@ def next_freeze_time(self, reference: Optional[datetime] = None) -> datetime:
     def healthy(self) -> tuple[bool,str]:
         return self.inner_api.healthy()
 
+    def lock_submission(
+        self, submission_id: Union[int, str]
+    ) -> AbstractContextManager[None]:
+        # Lock state lives in the inner API's database; delegate.
+        return self.inner_api.lock_submission(submission_id)
+
     @staticmethod
     def serialize_msg(*events: Event) -> bytes:
         """Serialize events to `bytes` to send as pubsub message."""
diff --git a/submit_ce/ui/controllers/new/review.py b/submit_ce/ui/controllers/new/review.py
index 007cab9b..dd7bd323 100644
--- a/submit_ce/ui/controllers/new/review.py
+++ b/submit_ce/ui/controllers/new/review.py
@@ -212,11 +212,13 @@ def _update_preflight(params: MultiDict, submission_id: str, workspace: Workspac
     if not has_changes:
         return False
 
-    file_store = current_app.api.get_file_store()
-    file_store.delete_preflight(submission_id)
-    file_store.store_user_decisions(submission_id, new_decisions)
-    for path in files_to_delete:
-        file_store.delete_source_file(submission_id, path)
+    api = current_app.api
+    file_store = api.get_file_store()
+    with api.lock_submission(submission_id):
+        file_store.delete_preflight(submission_id)
+        file_store.store_user_decisions(submission_id, new_decisions)
+        for path in files_to_delete:
+            file_store.delete_source_file(submission_id, path)
 
     return True
 
@@ -248,17 +250,23 @@ def _load_or_create_preflight(submission_id: str, params: MultiDict, session: Se
     preflight_data = _get_preflight_data(submission_id)
     zzrm_data = None
     if preflight_data is None:
-        file_store = current_app.api.get_file_store()
+        api = current_app.api
+        file_store = api.get_file_store()
         # if there is no preflight, then there wouldn't be a user decisions file.
         #   check if there is a zzrm, and use that as initial user decisions.
         zzrm_data = _get_zzrm_data(workspace, submission_id)
-        if zzrm_data is not None:
-            zzrm_data = dm.convert_zzrm_to_user_decisions(zzrm_data)
-            file_store.store_user_decisions(submission_id, zzrm_data)
-            file_store.delete_source_file(submission_id, '00README.json')
-
-        # user_decisions + preflight + compile logs -> directives.json
-        file_store.delete_directives(submission_id)
+        # Lock around the direct file_store mutations only. The
+        # subsequent start_preflight() goes through api.save(), which
+        # takes its own row lock from a separate session — re-entering
+        # the lock from here would deadlock on the same row.
+        with api.lock_submission(submission_id):
+            if zzrm_data is not None:
+                zzrm_data = dm.convert_zzrm_to_user_decisions(zzrm_data)
+                file_store.store_user_decisions(submission_id, zzrm_data)
+                file_store.delete_source_file(submission_id, '00README.json')
+
+            # user_decisions + preflight + compile logs -> directives.json
+            file_store.delete_directives(submission_id)
 
         start_preflight(params, session, submission_id, token)
         preflight_data = _get_preflight_data(submission_id)
diff --git a/submit_ce/ui/factory.py b/submit_ce/ui/factory.py
index 4e52d1b2..ab6f04e9 100644
--- a/submit_ce/ui/factory.py
+++ b/submit_ce/ui/factory.py
@@ -4,11 +4,13 @@
 
 from flask.logging import default_handler
 from flask import Flask, request
+from werkzeug.exceptions import Conflict
 
 from arxiv.base import Base
 from arxiv.config import settings as base_settings
 from arxiv import db
 
+from submit_ce.domain.exceptions import SubmissionLocked
 from .auth import request_auth
 from .config import settings
 from . import backend, filters
@@ -58,4 +60,14 @@ def check_auth():
     def shutdown_session(exception=None):
         db.Session.remove()
 
+    @app.errorhandler(SubmissionLocked)
+    def _handle_submission_locked(exc: SubmissionLocked):
+        # Return (do not raise) the Conflict so Flask renders it
+        # through its standard HTTPException path. Raising here would
+        # turn the original exception into a fresh unhandled one.
+        return Conflict(description=(
+            "Another operation is in progress for this submission; "
+            "please retry."
+        ))
+
     return app
diff --git a/submit_ce/ui/tests/test_file_store_lock_guard.py b/submit_ce/ui/tests/test_file_store_lock_guard.py
new file mode 100644
index 00000000..71122536
--- /dev/null
+++ b/submit_ce/ui/tests/test_file_store_lock_guard.py
@@ -0,0 +1,211 @@
+"""Guard test: every direct file_store mutation in UI controllers must
+happen inside `with api.lock_submission(...)`.
+
+This is a structural check. Direct file-store writes/deletes bypass
+`api.save()` and so do not pick up the row lock that `save()` would
+take. Without the wrapper, two browsers on the same submission can
+race a write against an in-flight compile/preflight/directives.
+
+The check targets only `submit_ce/ui/controllers/**.py`. It does NOT
+target:
+
+- `submit_ce/domain/event/file.py` and other event-side mutators,
+  which run inside `event.execute()` already covered by `save()`'s
+  lock.
+- `submit_ce/implementations/compile/compile_api_service.py`, whose
+  `store_preview`-style writes happen inside `start_compile` /
+  `start_preflight` invoked from `Start*.execute()`, again under
+  `save()`.
+- Test code and the file_store implementations themselves.
+
+A naive regex would miss aliased / fluent forms. The check uses an
+AST walk that tracks names assigned from `*.get_file_store()` and
+also recognises the fluent `*.get_file_store().store_*(...)` form.
+"""
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+from typing import Iterable, Set
+
+import pytest
+
+
+CONTROLLERS_ROOT = (
+    Path(__file__).resolve().parents[2] / "ui" / "controllers"
+)
+
+# Conventional aliases for the file store. Calls on a name matching
+# one of these are tracked even if we can't see the assignment (e.g.
+# the alias was set in an enclosing scope or passed in as a parameter).
+CONVENTIONAL_NAMES: Set[str] = {"file_store", "fstore", "store"}
+
+
+def _iter_controller_files() -> Iterable[Path]:
+    for path in CONTROLLERS_ROOT.rglob("*.py"):
+        # Skip the controllers' own test subdir and __pycache__.
+        parts = set(path.parts)
+        if "tests" in parts or "__pycache__" in parts:
+            continue
+        yield path
+
+
+def _attach_parents(tree: ast.AST) -> None:
+    for node in ast.walk(tree):
+        for child in ast.iter_child_nodes(node):
+            child.parent = node  # type: ignore[attr-defined]
+
+
+def _is_file_store_factory_call(node: ast.AST) -> bool:
+    """True if `node` is a Call whose .func is an Attribute with
+    attr == 'get_file_store' (e.g. `current_app.api.get_file_store()`,
+    `api.get_file_store()`, `self.api.get_file_store()`)."""
+    return (
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "get_file_store"
+    )
+
+
+def _collect_aliases(scope: ast.AST) -> Set[str]:
+    """Names within `scope` assigned (somewhere) from `*.get_file_store()`."""
+    aliases: Set[str] = set()
+    for node in ast.walk(scope):
+        if isinstance(node, ast.Assign) and _is_file_store_factory_call(node.value):
+            for target in node.targets:
+                if isinstance(target, ast.Name):
+                    aliases.add(target.id)
+        elif isinstance(node, ast.AnnAssign) and _is_file_store_factory_call(node.value):
+            if isinstance(node.target, ast.Name):
+                aliases.add(node.target.id)
+    return aliases
+
+
+def _is_tracked_receiver(receiver: ast.AST, aliases: Set[str]) -> bool:
+    if isinstance(receiver, ast.Name) and (
+        receiver.id in aliases or receiver.id in CONVENTIONAL_NAMES
+    ):
+        return True
+    if _is_file_store_factory_call(receiver):
+        return True
+    return False
+
+
+def _inside_lock_submission_with(node: ast.AST) -> bool:
+    """Walk parent links to see if `node` is inside a `with` block
+    whose context manager is a Call of an Attribute with
+    attr == 'lock_submission'."""
+    current = getattr(node, "parent", None)
+    while current is not None:
+        if isinstance(current, (ast.With, ast.AsyncWith)):
+            for item in current.items:
+                ctx = item.context_expr
+                if (
+                    isinstance(ctx, ast.Call)
+                    and isinstance(ctx.func, ast.Attribute)
+                    and ctx.func.attr == "lock_submission"
+                ):
+                    return True
+        current = getattr(current, "parent", None)
+    return False
+
+
+def _find_violations(path: Path) -> list[str]:
+    source = path.read_text(encoding="utf-8")
+    tree = ast.parse(source, filename=str(path))
+    _attach_parents(tree)
+    aliases = _collect_aliases(tree)
+
+    violations: list[str] = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if not isinstance(func, ast.Attribute):
+            continue
+        if not (func.attr.startswith("store_") or func.attr.startswith("delete_")):
+            continue
+        if not _is_tracked_receiver(func.value, aliases):
+            continue
+        if _inside_lock_submission_with(node):
+            continue
+        violations.append(
+            f"{path}:{node.lineno}: {ast.unparse(node)}"
+        )
+    return violations
+
+
+def test_ui_controller_file_store_mutations_are_locked():
+    """Every file_store.store_*/delete_* call in a UI controller must
+    be wrapped in `with api.lock_submission(...)`."""
+    all_violations: list[str] = []
+    for path in _iter_controller_files():
+        all_violations.extend(_find_violations(path))
+
+    assert not all_violations, (
+        "Direct file_store mutations in UI controllers must be inside "
+        "`with api.lock_submission(submission_id):`. Found:\n  "
+        + "\n  ".join(all_violations)
+    )
+
+
+def test_guard_does_not_fire_on_event_side_mutators(tmp_path: Path):
+    """Negative control: the same patterns in domain/event/file.py
+    are covered by api.save()'s lock and must not be flagged.
+
+    We assert this by running the guard's checker over that file
+    directly and verifying it returns no violations only because the
+    file is not under the controllers root — i.e. the guard's scope
+    excludes it by construction."""
+    event_file = (
+        Path(__file__).resolve().parents[2]
+        / "domain" / "event" / "file.py"
+    )
+    assert event_file.exists(), "expected domain/event/file.py to exist"
+    # The file is not under CONTROLLERS_ROOT, so the controller walk
+    # skips it. _find_violations would flag every call site if applied
+    # directly — that's the point: scope, not behaviour, exempts it.
+    assert CONTROLLERS_ROOT not in event_file.parents
+
+
+def test_guard_catches_a_synthetic_ungated_mutation(tmp_path: Path):
+    """Positive control: synthesise an ungated mutation in a fake
+    controller file and verify the checker reports it."""
+    fake = tmp_path / "fake_controller.py"
+    fake.write_text(
+        "from flask import current_app\n"
+        "def handler(submission_id):\n"
+        "    file_store = current_app.api.get_file_store()\n"
+        "    file_store.delete_source_file(submission_id, 'main.tex')\n"
+    )
+    violations = _find_violations(fake)
+    assert any("delete_source_file" in v for v in violations), violations
+
+
+def test_guard_accepts_wrapped_mutation(tmp_path: Path):
+    """Positive control: same mutation wrapped in `lock_submission`
+    must not be flagged."""
+    fake = tmp_path / "fake_controller_ok.py"
+    fake.write_text(
+        "from flask import current_app\n"
+        "def handler(submission_id):\n"
+        "    api = current_app.api\n"
+        "    file_store = api.get_file_store()\n"
+        "    with api.lock_submission(submission_id):\n"
+        "        file_store.delete_source_file(submission_id, 'main.tex')\n"
+    )
+    violations = _find_violations(fake)
+    assert violations == [], violations
+
+
+def test_guard_handles_fluent_chain(tmp_path: Path):
+    """Fluent call current_app.api.get_file_store().delete_directives(...)
+    outside of a lock_submission block must be flagged."""
+    fake = tmp_path / "fake_controller_fluent.py"
+    fake.write_text(
+        "from flask import current_app\n"
+        "def handler(submission_id):\n"
+        "    current_app.api.get_file_store().delete_directives(submission_id)\n"
+    )
+    violations = _find_violations(fake)
+    assert any("delete_directives" in v for v in violations), violations
diff --git a/submit_ce/ui/tests/test_submission_locked_handler.py b/submit_ce/ui/tests/test_submission_locked_handler.py
new file mode 100644
index 00000000..2e89e95b
--- /dev/null
+++ b/submit_ce/ui/tests/test_submission_locked_handler.py
@@ -0,0 +1,65 @@
+"""Test that the Flask app converts `SubmissionLocked` into HTTP 409.
+
+The error handler is registered in `submit_ce.ui.factory.create_web_app`.
+Raising `SubmissionLocked` from inside a request (or from inside a view
+function, an `api.save(...)` call, or `api.lock_submission(...)`) must
+produce HTTP 409 Conflict, not 500 Internal Server Error.
+
+A dedicated `SubmissionLocked` exception that is intentionally NOT a
+subclass of `SaveError` keeps the existing `except SaveError: raise
+InternalServerError` blocks in the controllers from swallowing it
+into a 500. That invariant is asserted in
+`test_lock_submission.test_submission_locked_not_subclass_of_save_error`.
+"""
+from __future__ import annotations
+
+from flask import Flask
+from werkzeug.exceptions import Conflict
+
+from submit_ce.domain.exceptions import SaveError, SubmissionLocked
+
+
+def _app_with_locked_handler() -> Flask:
+    """Build a tiny Flask app with just the handler we care about,
+    avoiding the full create_web_app() (which needs DB + auth setup).
+    Mirrors the registration in factory.py:create_web_app."""
+    app = Flask(__name__)
+
+    @app.errorhandler(SubmissionLocked)
+    def _handle_submission_locked(exc):
+        return Conflict(description=(
+            "Another operation is in progress for this submission; "
+            "please retry."
+        ))
+
+    @app.route("/raise-locked")
+    def _raise_locked():
+        raise SubmissionLocked(42)
+
+    @app.route("/raise-save-error")
+    def _raise_save_error():
+        raise SaveError("some save failed")
+
+    return app
+
+
+def test_submission_locked_becomes_409():
+    app = _app_with_locked_handler()
+    client = app.test_client()
+
+    resp = client.get("/raise-locked")
+    assert resp.status_code == 409, resp.data
+    assert b"Another operation is in progress" in resp.data
+
+
+def test_save_error_is_not_caught_by_submission_locked_handler():
+    """Sanity-check: a SaveError must NOT be swallowed by the
+    SubmissionLocked handler. With Flask's default config, an
+    unhandled SaveError bubbles up to a 500."""
+    app = _app_with_locked_handler()
+    # propagate exceptions instead of converting to 500, so we can
+    # assert the exception type cleanly.
+    app.testing = False
+    client = app.test_client()
+    resp = client.get("/raise-save-error")
+    assert resp.status_code == 500