fix(dataset): lock the If-Match validator's transaction-id read

docs/static/resources/openapi.json

@@ -24039,7 +24039,7 @@
             }
           },
           {
-            "description": "Optional optimistic-concurrency guard. Pass the ``ETag`` returned by a prior read of this dataset; the update is rejected with 412 if the dataset has changed since.",
+            "description": "Optional optimistic-concurrency guard. Pass the ``ETag`` returned by a prior read of this dataset; the update is rejected with 412 if the dataset has changed since, or with a retryable 409 if a concurrent save won a lock race (see those responses for which action each calls for).",
             "in": "header",
             "name": "If-Match",
             "schema": {
@@ -24121,6 +24121,21 @@
           "404": {
             "$ref": "#/components/responses/404"
           },
+          "409": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "properties": {
+                    "message": {
+                      "type": "string"
+                    }
+                  },
+                  "type": "object"
+                }
+              }
+            },
+            "description": "A concurrent save on this dataset won a database lock race, so this conditional request could not be applied. The client's ``If-Match`` token is NOT proven stale: unlike a 412, the correct action is to retry the SAME request (with backoff; a few attempts, then surface the conflict) rather than refetch the entity for a new token."
+          },
           "412": {
             "content": {
               "application/json": {
@@ -24134,7 +24149,7 @@
                 }
               }
             },
-            "description": "The dataset changed since the version identified by the request's ``If-Match`` header; the update was not applied."
+            "description": "The dataset changed since the version identified by the request's ``If-Match`` header; the update was not applied. The token is stale: refetch the entity to pick up the current version before retrying \u2014 retrying the same request unchanged will fail again (contrast the retryable 409 above)."
           },
           "422": {
             "$ref": "#/components/responses/422"

superset/daos/version.py

@@ -30,6 +30,7 @@
 
 from superset.versioning.queries import (
     current_live_transaction_id,
+    current_live_transaction_id_locked,
     current_live_version_uuid,
     current_version_info,
     current_version_number,
@@ -67,6 +68,9 @@ class VersionDAO:
     current_version_number = staticmethod(current_version_number)
     current_version_info = staticmethod(current_version_info)
     current_live_transaction_id = staticmethod(current_live_transaction_id)
+    current_live_transaction_id_locked = staticmethod(
+        current_live_transaction_id_locked
+    )
     current_live_version_uuid = staticmethod(current_live_version_uuid)
     list_change_records_batch = staticmethod(list_change_records_batch)
     list_versions = staticmethod(list_versions)

superset/datasets/api.py

@@ -32,9 +32,10 @@
 from flask_babel import gettext as _, ngettext
 from jinja2.exceptions import TemplateError
 from marshmallow import ValidationError
+from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm.exc import MultipleResultsFound
 
-from superset import event_logger, is_feature_enabled, security_manager
+from superset import db, event_logger, is_feature_enabled, security_manager
 from superset.commands.dataset.create import CreateDatasetCommand
 from superset.commands.dataset.delete import DeleteDatasetCommand
 from superset.commands.dataset.duplicate import DuplicateDatasetCommand
@@ -112,11 +113,13 @@
     concurrency_token_from,
     current_entity_version_info,
     entity_concurrency_token,
+    EntityVersionInfo,
     get_version_endpoint,
     list_versions_endpoint,
     lock_entity_for_update,
     restore_version_endpoint,
 )
+from superset.versioning.db_errors import is_lock_contention_error
 from superset.versioning.etag import (
     is_conditional_write,
     raise_for_stale_write,
@@ -571,6 +574,28 @@ def post(self) -> Response:
             logger.exception("Unexpected error in DatasetRestApi.post")
             return self.response_500(message="Fatal error")
 
+    def _lock_contention_response(self) -> Response:
+        """The shared retryable 409 for a conditional save losing a lock race.
+
+        In the direct lock-acquisition and validator-read catches, the
+        rollback is LOAD-BEARING for lock-wait-timeout: with
+        ``innodb_rollback_on_timeout`` OFF (the MySQL default) a 1205
+        rolls back only the failing STATEMENT -- the transaction is still
+        alive and still holds any entity row lock already acquired, and
+        this rollback is what releases it. For a deadlock (1213) InnoDB
+        already rolled the transaction back and this clears the aborted
+        session before responding. The decorated command-failure path
+        already rolls back before raising; this shared cleanup is a
+        harmless no-op on that path, not an additional required rollback.
+        """
+        db.session.rollback()  # pylint: disable=consider-using-transaction
+        return self.response(
+            409,
+            message=_(
+                "Another save is in progress for this dataset. Retry the same request."
+            ),
+        )
+
     @expose("/<pk>", methods=("PUT",))
     @protect()
     @statsd_metrics
@@ -579,7 +604,7 @@ def post(self) -> Response:
         log_to_statsd=False,
     )
     @requires_json
-    def put(self, pk: int) -> Response:
+    def put(self, pk: int) -> Response:  # noqa: C901
         """Update a dataset.
         ---
         put:
@@ -598,9 +623,11 @@ def put(self, pk: int) -> Response:
               type: string
             name: If-Match
             description: >-
-              Optional optimistic-concurrency guard. Pass the ``ETag`` returned
-              by a prior read of this dataset; the update is rejected with 412
-              if the dataset has changed since.
+              Optional optimistic-concurrency guard. Pass the ``ETag``
+              returned by a prior read of this dataset; the update is
+              rejected with 412 if the dataset has changed since, or
+              with a retryable 409 if a concurrent save won a lock race
+              (see those responses for which action each calls for).
           requestBody:
             description: Dataset schema
             required: true
@@ -677,10 +704,30 @@ def put(self, pk: int) -> Response:
               $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
+            409:
+              description: >-
+                A concurrent save on this dataset won a database lock
+                race, so this conditional request could not be applied.
+                The client's ``If-Match`` token is NOT proven stale:
+                unlike a 412, the correct action is to retry the SAME
+                request (with backoff; a few attempts, then surface the
+                conflict) rather than refetch the entity for a new
+                token.
+              content:
+                application/json:
+                  schema:
+                    type: object
+                    properties:
+                      message:
+                        type: string
             412:
               description: >-
                 The dataset changed since the version identified by the
-                request's ``If-Match`` header; the update was not applied.
+                request's ``If-Match`` header; the update was not
+                applied. The token is stale: refetch the entity to pick
+                up the current version before retrying — retrying the
+                same request unchanged will fail again (contrast the
+                retryable 409 above).
               content:
                 application/json:
                   schema:
@@ -706,22 +753,50 @@ def put(self, pk: int) -> Response:
 
         # Serialise conditional saves on this dataset: the guard below reads
         # the live version, the command writes, and the two must not interleave
-        # with another request's. Only a conditional save pays for the lock; an
+        # with another request's. Only a conditional save pays for the locks; an
         # unconditional PUT behaves exactly as it did before the guard existed.
-        # (On MySQL REPEATABLE READ the version read below is still a plain
-        # consistent read and can predate the lock; see the caveats on
-        # lock_entity_for_update.)
-        if is_conditional_write():
-            # Hold the locked entity for the rest of the request: the
-            # session's identity map references clean objects weakly, so
-            # discarding the return value could let the refreshed object
-            # be collected and the command's find_by_id re-read a stale
-            # row on MySQL REPEATABLE READ (see lock_entity_for_update).
-            _locked_entity = lock_entity_for_update(SqlaTable, pk)
+        # The validator's transaction id uses a locking read. The displayed
+        # version number and lazy child reads can still observe an older
+        # MySQL REPEATABLE READ snapshot; see lock_entity_for_update.
+        conditional: bool = is_conditional_write()
+        if conditional:
+            # Blocking here is the serialisation doing its job; LOSING the
+            # race at this acquisition (deadlock, or lock-wait timeout
+            # after ``innodb_lock_wait_timeout``) is the most common
+            # contention outcome of all three lock points and maps to the
+            # same retryable 409 -- not an uncaught 500. The locked entity
+            # is HELD for the rest of the request: the session's identity
+            # map references clean objects weakly, so discarding the
+            # return value could let the refreshed object be collected and
+            # the command's find_by_id re-read a stale row on MySQL
+            # REPEATABLE READ (see lock_entity_for_update).
+            try:
+                _locked_entity: SqlaTable | None = lock_entity_for_update(SqlaTable, pk)
+            except OperationalError as ex:
+                if not is_lock_contention_error(ex):
+                    raise
+                return self._lock_contention_response()
 
         # Live version identifiers before the update (empty + query-free when
-        # ``ENABLE_VERSIONING_CAPTURE`` is off).
-        old_info = current_entity_version_info(SqlaTable, pk)
+        # ``ENABLE_VERSIONING_CAPTURE`` is off). On the conditional path the
+        # live transaction id is read under an exclusive row lock: a plain
+        # read is served from the request's REPEATABLE READ snapshot on MySQL
+        # and can miss a concurrent commit, letting a stale If-Match token
+        # pass the guard. A lock race lost at that read (deadlock / lock
+        # wait) proves concurrent CONTENTION, not that this request's token
+        # is stale — so it maps to a retryable 409, and the client should
+        # retry the SAME request. (A deadlock at Continuum's version-row
+        # insert inside the command is classified by the
+        # DatasetUpdateFailedError handler below via ``__cause__`` and
+        # returns the same 409.)
+        try:
+            old_info: EntityVersionInfo = current_entity_version_info(
+                SqlaTable, pk, lock_for_stale_check=conditional
+            )
+        except OperationalError as ex:
+            if not (conditional and is_lock_contention_error(ex)):
+                raise
+            return self._lock_contention_response()
 
         try:
             raise_for_stale_write(concurrency_token_from(old_info))
@@ -796,6 +871,16 @@ def put(self, pk: int) -> Response:
             )
             response = self.response_422(message=str(ex))
         except DatasetUpdateFailedError as ex:
+            # The gap lock the conditional path's locking read takes on a
+            # zero-live-row range can deadlock against another conditional
+            # writer at Continuum's version-row INSERT inside the command;
+            # on_error chains the driver error as __cause__, and the update
+            # transaction has rolled back. (The post-commit override_columns
+            # refresh raises its own exception type and cannot reach this
+            # branch.) Same retryable classification as the read-point
+            # handler above: the token is not proven stale.
+            if conditional and is_lock_contention_error(ex.__cause__):
+                return self._lock_contention_response()
             logger.error(
                 "Error updating model %s: %s",
                 self.__class__.__name__,

superset/translations/messages.pot

@@ -17279,3 +17279,7 @@ msgstr ""
 
 msgid "Show all"
 msgstr ""
+
+#: superset/datasets/api.py
+msgid "Another save is in progress for this dataset. Retry the same request."
+msgstr ""

superset/versioning/api_helpers.py

@@ -94,12 +94,26 @@ def current_entity_version_info(
     model_cls: type[Model],
     entity_id: int | None,
     entity_uuid: UUID | None = None,
+    *,
+    lock_for_stale_check: bool = False,
 ) -> EntityVersionInfo:
     """Resolve the live version number, transaction id, and version uuid.
 
     Returns an empty (all-``None``) record and issues *no* queries when
     capture is disabled. When *entity_uuid* is not supplied it is resolved
     with a single ``SELECT uuid`` rather than loading the whole entity row.
+
+    With ``lock_for_stale_check`` the live ``transaction_id`` — the input
+    to the ``If-Match`` token — is read under an exclusive row lock so it
+    reflects committed state even on MySQL REPEATABLE READ, where a plain
+    read is served from the request's pre-lock snapshot (see
+    :func:`~superset.versioning.queries.current_live_transaction_id_locked`
+    for the mechanism, lock-strength rationale, and residuals). Only
+    conditional writes should pay for the lock. The displayed version
+    *number* still comes from a plain aggregate read: a concurrently
+    committed row can leave it one behind in response metadata, but the
+    guard itself never consults it. The entity-uuid resolution also stays
+    a plain read: the uuid is immutable for the life of the row.
     """
     if entity_id is None or not _capture_enabled():
         return EntityVersionInfo()
@@ -117,9 +131,15 @@ def current_entity_version_info(
         )
     if entity_uuid is None:
         return EntityVersionInfo()
+    version: int | None
+    transaction_id: int | None
     version, transaction_id = VersionDAO.current_version_info(
         model_cls, entity_id, entity_uuid
     )
+    if lock_for_stale_check:
+        transaction_id = VersionDAO.current_live_transaction_id_locked(
+            model_cls, entity_id, entity_uuid
+        )
     version_uuid = (
         VersionDAO.derive_version_uuid(entity_uuid, transaction_id)
         if transaction_id is not None
@@ -227,10 +247,10 @@ def lock_entity_for_update(
     undoing the refresh.
 
     The refresh covers the entity row itself. Lazy-loaded child
-    collections (columns, metrics) are still plain consistent reads
-    afterwards, as is the version-info read behind the ``If-Match``
-    comparison -- on MySQL REPEATABLE READ both can still observe the
-    pre-lock snapshot.
+    collections (columns, metrics) and the displayed version number remain
+    plain consistent reads and can observe the pre-lock snapshot on MySQL
+    REPEATABLE READ. The conditional ``If-Match`` comparison instead uses
+    a locking read of the live transaction id.
 
     Postgres (READ COMMITTED) and SQLite are not exposed to the staleness,
     and the refresh is harmless there. (Version restore has the same

superset/versioning/db_errors.py

@@ -34,6 +34,81 @@
 
 from sqlalchemy.exc import DBAPIError
 
+#: MySQL/MariaDB deadlock and lock-wait-timeout error codes.
+_MYSQL_LOCK_CONTENTION: tuple[int, ...] = (1213, 1205)
+
+#: PostgreSQL SQLSTATEs: serialization_failure, deadlock_detected,
+#: lock_not_available.
+_PG_LOCK_CONTENTION: tuple[str, ...] = ("40001", "40P01", "55P03")
+
+#: Contention phrases for drivers that carry no structured code at all
+#: (SQLite). Matched against the DRIVER's own message only -- never a
+#: SQLAlchemy wrapper's rendering, which appends the statement and its
+#: bound parameters.
+_LOCK_CONTENTION_PHRASES: tuple[str, ...] = (
+    "deadlock",
+    "lock wait timeout",
+    "database is locked",
+    "database table is locked",
+)
+
+
+def is_lock_contention_error(exc: BaseException | None) -> bool:
+    """Whether *exc* is a database deadlock / lock-wait failure.
+
+    A write that loses a lock race has, by definition, interleaved with a
+    concurrent writer. It does NOT prove the caller's ``If-Match`` token
+    stale, so response-mapping callers classify it as a retryable
+    conflict (409, retry the same request) rather than a 500 -- or a 412,
+    whose refetch-the-token guidance would be wrong here.
+
+    Classification reads the DRIVER DIAGNOSTIC, never the SQLAlchemy
+    wrapper's rendered text: ``str(DBAPIError)`` appends the failing
+    statement and its bound parameters, so an unrelated failure whose
+    SQL or values merely contain a lock word (a chart named "deadlock
+    analysis", a filter on a column called ``lock_wait_timeout``) would
+    be answered "retry the same request" forever.
+
+    A structured code is AUTHORITATIVE where the driver supplies one: a
+    MySQL errno or a PostgreSQL SQLSTATE that is not in the contention
+    sets ends classification at ``False`` rather than falling through to
+    a keyword match. Only code-less drivers reach the phrase check, and
+    then only against their own message. Accepts ``None`` (an exception
+    with no ``__cause__``) and errors with empty driver args without
+    raising.
+    """
+    if exc is None:
+        return False
+
+    orig: object = getattr(exc, "orig", None)
+    if orig is None:
+        if isinstance(exc, DBAPIError):
+            # A wrapper whose driver error was not preserved carries no
+            # diagnostic to classify -- and its own text is the
+            # statement-and-parameters rendering this function must not
+            # read. Unclassifiable is not retryable.
+            return False
+        # Not a wrapper: a raw driver exception is its own diagnostic.
+        diagnostic: object = exc
+    else:
+        diagnostic = orig
+
+    args: tuple[object, ...] | None = getattr(diagnostic, "args", None)
+    if args and isinstance(args[0], int):
+        # MySQL-family errno: structured and therefore decisive.
+        return args[0] in _MYSQL_LOCK_CONTENTION
+
+    sqlstate: str | None = getattr(diagnostic, "pgcode", None) or getattr(
+        diagnostic, "sqlstate", None
+    )
+    if sqlstate is not None:
+        # PostgreSQL SQLSTATE: structured and therefore decisive.
+        return sqlstate in _PG_LOCK_CONTENTION
+
+    text: str = str(diagnostic).lower()
+    return any(phrase in text for phrase in _LOCK_CONTENTION_PHRASES)
+
+
 #: PostgreSQL SQLSTATE for "relation does not exist" (undefined_table).
 _PG_UNDEFINED_TABLE = "42P01"