From a53f3fecdf8162076c09ad7a672971ac8699519c Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Wed, 9 Sep 2026 10:37:48 -0600 Subject: [PATCH 1/8] fix(dataset): read the If-Match validator's transaction id under FOR SHARE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The conditional-write guard compared the client's If-Match token against a version-info lookup made with plain consistent reads. On MySQL/InnoDB REPEATABLE READ those are served from the snapshot pinned by the request's first read, so a version row committed by a concurrent writer between that first read and this request's row lock stayed invisible — the stale token matched, raise_for_stale_write passed, and the second writer overwrote the first: the lost update the lock + If-Match pair exists to prevent, surviving on MySQL only. Validator-layer sibling of the entity refresh in apache/superset#44027. Add current_live_transaction_id_for_share — a plain live-row query (not the aggregate: locking clauses and aggregates don't combine reliably across dialects) under with_for_update(read=True), exempt from the RR snapshot without blocking readers — and thread lock_for_stale_check through current_entity_version_info; the dataset PUT passes it only for conditional writes. The displayed version number stays a plain aggregate read: a concurrent commit can leave response metadata one behind, but the guard never consults it. sc-120050 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW --- superset/daos/version.py | 4 + superset/datasets/api.py | 9 +- superset/versioning/api_helpers.py | 15 +++ superset/versioning/queries.py | 35 +++++++ .../conditional_token_lock_tests.py | 71 ++++++++++++++ .../versioning/test_version_info_locking.py | 94 +++++++++++++++++++ 6 files changed, 226 insertions(+), 2 deletions(-) create mode 100644 tests/integration_tests/versioning/conditional_token_lock_tests.py create mode 100644 tests/unit_tests/versioning/test_version_info_locking.py diff --git a/superset/daos/version.py b/superset/daos/version.py index 8f92d7162912..98fe16ed18ce 100644 --- a/superset/daos/version.py +++ b/superset/daos/version.py @@ -30,6 +30,7 @@ from superset.versioning.queries import ( current_live_transaction_id, + current_live_transaction_id_for_share, 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_for_share = staticmethod( + current_live_transaction_id_for_share + ) current_live_version_uuid = staticmethod(current_live_version_uuid) list_change_records_batch = staticmethod(list_change_records_batch) list_versions = staticmethod(list_versions) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index 9e2ae21a0abe..1b5d7357dbd1 100644 --- a/superset/datasets/api.py +++ b/superset/datasets/api.py @@ -720,8 +720,13 @@ def put(self, pk: int) -> Response: _locked_entity = lock_entity_for_update(SqlaTable, pk) # 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 FOR SHARE: 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. + old_info = current_entity_version_info( + SqlaTable, pk, lock_for_stale_check=is_conditional_write() + ) try: raise_for_stale_write(concurrency_token_from(old_info)) diff --git a/superset/versioning/api_helpers.py b/superset/versioning/api_helpers.py index dfe0e2d63a72..932bcdeb01aa 100644 --- a/superset/versioning/api_helpers.py +++ b/superset/versioning/api_helpers.py @@ -93,12 +93,23 @@ 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 ``FOR SHARE`` 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_for_share`). + 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. """ if entity_id is None or not _capture_enabled(): return EntityVersionInfo() @@ -119,6 +130,10 @@ def current_entity_version_info( 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_for_share( + model_cls, entity_id, entity_uuid + ) version_uuid = ( VersionDAO.derive_version_uuid(entity_uuid, transaction_id) if transaction_id is not None diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index 91cc5073dcc0..0a7063f949d1 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -224,6 +224,41 @@ def current_version_number( return version_number +def current_live_transaction_id_for_share( + model_cls: type[Model], entity_id: int, entity_uuid: UUID +) -> int | None: + """Return the live row's ``transaction_id`` via a locking read. + + The conditional-write (``If-Match``) guard must compare the client's + token against *committed* state. A plain consistent read is served + from the transaction's REPEATABLE READ snapshot on MySQL/InnoDB + (pinned by the request's earlier auth queries), so a version row + committed by a concurrent writer between this request's first read + and its row lock stays invisible — the stale token then matches and + the 412 the guard exists to raise is missed. ``FOR SHARE`` + (``with_for_update(read=True)``) makes this a locking read, which is + exempt from the snapshot and returns current committed data, without + blocking other readers of the same version row. + + Deliberately a plain row query, not the aggregate + :func:`current_version_info` — locking clauses do not combine with + aggregates reliably across dialects, and the guard only needs the + live ``transaction_id``. Renders no ``FOR SHARE`` on SQLite, which + serialises writers anyway. + """ + ver_cls = version_class(model_cls) + return ( + db.session.query(ver_cls.transaction_id) + .filter(identity_filter(ver_cls, entity_id, entity_uuid)) + .filter( + ver_cls.end_transaction_id.is_(None), + ver_cls.operation_type != OPERATION_DELETE, + ) + .with_for_update(read=True) + .scalar() + ) + + def current_live_transaction_id( model_cls: type[Model], entity_id: int, entity_uuid: UUID ) -> int | None: diff --git a/tests/integration_tests/versioning/conditional_token_lock_tests.py b/tests/integration_tests/versioning/conditional_token_lock_tests.py new file mode 100644 index 000000000000..93a409c8a126 --- /dev/null +++ b/tests/integration_tests/versioning/conditional_token_lock_tests.py @@ -0,0 +1,71 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""sc-120050: real-database proof for the FOR SHARE validator read. + +The statement-shape pins live in +tests/unit_tests/versioning/test_version_info_locking.py; the MySQL +REPEATABLE READ staleness itself cannot flip on Postgres (READ COMMITTED +takes a fresh snapshot per statement) and needs an interleaved read view +to reproduce on MySQL, so -- mirroring the honest-scope precedent of the +entity-lock fix -- these tests prove the locking read executes correctly +against a real backend and agrees with the plain read on the quiet path. +""" + +import pytest + +from superset import db +from superset.daos.version import VersionDAO +from superset.models.slice import Slice +from superset.versioning.api_helpers import current_entity_version_info +from tests.integration_tests.base_tests import SupersetTestCase +from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa: F401 + load_birth_names_dashboard_with_slices, + load_birth_names_data, +) + + +@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices") +class TestConditionalTokenLockingRead(SupersetTestCase): + def _versioned_chart(self) -> Slice: + """A chart with at least one version row (a real committed save).""" + chart = db.session.query(Slice).filter(Slice.slice_name == "Boys").one() + chart.description = "sc-120050 probe" + db.session.commit() + return chart + + def test_for_share_read_agrees_with_plain_read_on_quiet_path(self) -> None: + """Without concurrent writers both reads name the same live row, + and the FOR SHARE clause executes on the real backend (the + dialect-syntax half of the guard the unit pins cannot cover).""" + chart = self._versioned_chart() + try: + plain = VersionDAO.current_live_transaction_id(Slice, chart.id, chart.uuid) + locked = VersionDAO.current_live_transaction_id_for_share( + Slice, chart.id, chart.uuid + ) + assert plain is not None + assert locked == plain + + info = current_entity_version_info( + Slice, chart.id, chart.uuid, lock_for_stale_check=True + ) + assert info.transaction_id == plain + finally: + db.session.rollback() + chart = db.session.query(Slice).get(chart.id) + chart.description = None + db.session.commit() diff --git a/tests/unit_tests/versioning/test_version_info_locking.py b/tests/unit_tests/versioning/test_version_info_locking.py new file mode 100644 index 000000000000..113624070e72 --- /dev/null +++ b/tests/unit_tests/versioning/test_version_info_locking.py @@ -0,0 +1,94 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""sc-120050: the If-Match validator reads committed state under lock. + +On MySQL/InnoDB REPEATABLE READ a plain consistent read is served from the +snapshot pinned by the request's first read, so the version-info lookup +behind ``raise_for_stale_write`` could miss a concurrently committed +version row -- a stale client token then matched and the 412 was skipped, +recreating the lost update the lock + If-Match pair exists to prevent +(the validator-layer sibling of the entity fix in apache/superset#44027). +These are the dialect-independent statement-shape guards: the +conditional-path transaction-id read must carry ``with_for_update(read=True)`` +(a locking read is exempt from the snapshot) and +``current_entity_version_info`` must thread the flag -- with the default +(every GET path) never invoking the locking read at all. The real-database +proof lives in +tests/integration_tests/versioning/conditional_token_lock_tests.py. +""" + +from unittest.mock import MagicMock +from uuid import uuid4 + +from pytest_mock import MockerFixture + +from superset.versioning.api_helpers import current_entity_version_info +from superset.versioning.queries import current_live_transaction_id_for_share + + +def _mock_queries_db(mocker: MockerFixture) -> MagicMock: + db = MagicMock() + mocker.patch("superset.versioning.queries.db", new=db) + mocker.patch("superset.versioning.queries.version_class", new=MagicMock()) + return db + + +def test_for_share_read_is_a_locking_row_read(mocker: MockerFixture) -> None: + """The conditional-path read chains with_for_update(read=True). + + Dropping the flag (or the whole locking clause) fails here on every + backend -- the analogue of the shape pins on the entity lock. + """ + db = _mock_queries_db(mocker) + + current_live_transaction_id_for_share(MagicMock(), 1, uuid4()) + + query = db.session.query.return_value + filtered = query.filter.return_value.filter.return_value + filtered.with_for_update.assert_called_once_with(read=True) + filtered.with_for_update.return_value.scalar.assert_called_once_with() + + +def test_version_info_threads_the_lock_flag(mocker: MockerFixture) -> None: + """lock_for_stale_check=True swaps in the FOR SHARE transaction id.""" + dao = MagicMock() + dao.current_version_info.return_value = (0, 5) + dao.current_live_transaction_id_for_share.return_value = 6 + mocker.patch("superset.versioning.api_helpers.VersionDAO", new=dao) + mocker.patch("superset.versioning.api_helpers._capture_enabled", return_value=True) + entity_uuid = uuid4() + + info = current_entity_version_info( + MagicMock(), 1, entity_uuid, lock_for_stale_check=True + ) + + dao.current_live_transaction_id_for_share.assert_called_once() + # The guard's input is the locked value, not the snapshot aggregate's. + assert info.transaction_id == 6 + + +def test_version_info_defaults_to_plain_reads(mocker: MockerFixture) -> None: + """Without the flag (every GET path) the locking read never runs.""" + dao = MagicMock() + dao.current_version_info.return_value = (0, 5) + mocker.patch("superset.versioning.api_helpers.VersionDAO", new=dao) + mocker.patch("superset.versioning.api_helpers._capture_enabled", return_value=True) + + info = current_entity_version_info(MagicMock(), 1, uuid4()) + + dao.current_live_transaction_id_for_share.assert_not_called() + assert info.transaction_id == 5 From 99cc727d1f4e5b9c5ced675ecafc007c2fc02365 Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Wed, 9 Sep 2026 10:51:56 -0600 Subject: [PATCH 2/8] fix(dataset): harden the locked validator read per review round MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three-lens findings on the FOR SHARE design, all addressed: - The lock is now EXCLUSIVE (with_for_update()): this transaction later closes the very row it reads (Continuum sets end_transaction_id at commit), so holding a shared lock first invited InnoDB's shared-to-exclusive upgrade deadlock whenever anything queued for the row in between (e.g. retention pruning). Plain MVCC readers are not blocked by either strength, and same-entity writers are already serialised by the entity lock, so exclusivity costs nothing. Renamed to current_live_transaction_id_locked accordingly. - The plain-row read is ordered-and-limited so a defensively tolerated multi-open-row state (which the aggregate's max absorbs) degrades the same way instead of raising MultipleResultsFound on every PUT. - MySQL's gap lock on the zero-live-row range leaves a rare cross-entity deadlock at Continuum's insert; a lock race lost there IS a concurrent interleave, so the PUT path maps lock-contention errors (is_lock_contention_error) to the same 412 retry semantics as a stale token instead of a 500. Residuals documented on the query helper. - Test holes closed: the endpoint wiring is pinned (the 412 unit test asserts lock_for_stale_check=True reaches the helper — previously deleting the kwarg at the call site survived every test), the guard's real input is pinned (derive_version_uuid must consume the LOCKED id), and the zero-live-row branch has an integration test. lock flag is keyword-only; is_conditional_write() hoisted; open-PR citations reworded intent-shaped; cleanup restores the captured original via session.get. sc-120050 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW --- superset/daos/version.py | 6 +- superset/datasets/api.py | 41 +++++++++--- superset/versioning/api_helpers.py | 38 ++++++++--- superset/versioning/queries.py | 45 +++++++++---- .../conditional_token_lock_tests.py | 63 +++++++++++++++---- tests/unit_tests/datasets/api_tests.py | 6 +- .../versioning/test_version_info_locking.py | 47 ++++++++------ 7 files changed, 183 insertions(+), 63 deletions(-) diff --git a/superset/daos/version.py b/superset/daos/version.py index 98fe16ed18ce..1acc22da0e13 100644 --- a/superset/daos/version.py +++ b/superset/daos/version.py @@ -30,7 +30,7 @@ from superset.versioning.queries import ( current_live_transaction_id, - current_live_transaction_id_for_share, + current_live_transaction_id_locked, current_live_version_uuid, current_version_info, current_version_number, @@ -68,8 +68,8 @@ 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_for_share = staticmethod( - current_live_transaction_id_for_share + 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) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index 1b5d7357dbd1..e228fa5482d9 100644 --- a/superset/datasets/api.py +++ b/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 @@ -113,6 +114,7 @@ current_entity_version_info, entity_concurrency_token, get_version_endpoint, + is_lock_contention_error, list_versions_endpoint, lock_entity_for_update, restore_version_endpoint, @@ -579,7 +581,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: @@ -706,12 +708,13 @@ 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(): + conditional = is_conditional_write() + if conditional: # 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 @@ -721,12 +724,30 @@ def put(self, pk: int) -> Response: # Live version identifiers before the update (empty + query-free when # ``ENABLE_VERSIONING_CAPTURE`` is off). On the conditional path the - # live transaction id is read under FOR SHARE: 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. - old_info = current_entity_version_info( - SqlaTable, pk, lock_for_stale_check=is_conditional_write() - ) + # 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 on that read (deadlock / lock + # wait) IS a concurrent interleave, so it maps to the same 412 retry + # semantics as a stale token. + try: + old_info = 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 + # Not a unit of work: the transaction is already dead (deadlock + # rollback); this clears the aborted session before responding. + db.session.rollback() # pylint: disable=consider-using-transaction + return self.response( + 412, + message=_( + "The dataset was changed by another user or browser tab " + "after you opened it. Reopen it to pick up the latest " + "version, then reapply your changes." + ), + ) try: raise_for_stale_write(concurrency_token_from(old_info)) diff --git a/superset/versioning/api_helpers.py b/superset/versioning/api_helpers.py index 932bcdeb01aa..bf53f335bb41 100644 --- a/superset/versioning/api_helpers.py +++ b/superset/versioning/api_helpers.py @@ -93,6 +93,7 @@ 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. @@ -102,14 +103,16 @@ def current_entity_version_info( 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 ``FOR SHARE`` 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_for_share`). - Only conditional writes should pay for the lock. The displayed version + 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. + 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() @@ -131,7 +134,7 @@ def current_entity_version_info( model_cls, entity_id, entity_uuid ) if lock_for_stale_check: - transaction_id = VersionDAO.current_live_transaction_id_for_share( + transaction_id = VersionDAO.current_live_transaction_id_locked( model_cls, entity_id, entity_uuid ) version_uuid = ( @@ -202,6 +205,27 @@ def entity_concurrency_token( ) or unversioned_entity_token(entity_uuid) +def is_lock_contention_error(ex: Exception) -> bool: + """Whether *ex* is a database deadlock / lock-wait failure. + + A conditional write that loses a lock race (e.g. MySQL's gap-lock + deadlock on the version table, error 1213, or a lock wait timeout, + 1205; Postgres serialization/deadlock SQLSTATEs 40001/40P01) has, by + definition, interleaved with a concurrent writer -- exactly the + situation ``If-Match`` exists to surface. Callers map it to the same + retry response as a stale token rather than a 500. + """ + orig = getattr(ex, "orig", None) + code = getattr(orig, "args", [None])[0] if orig is not None else None + if code in (1213, 1205): + return True + sqlstate = getattr(orig, "pgcode", None) + if sqlstate in ("40001", "40P01"): + return True + text = str(ex).lower() + return "deadlock" in text or "lock wait timeout" in text + + def lock_entity_for_update( model_cls: type[Model], entity_id: int | None ) -> Model | None: diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index 0a7063f949d1..a7431702a27e 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -224,27 +224,46 @@ def current_version_number( return version_number -def current_live_transaction_id_for_share( +def current_live_transaction_id_locked( model_cls: type[Model], entity_id: int, entity_uuid: UUID ) -> int | None: - """Return the live row's ``transaction_id`` via a locking read. + """Return the live row's ``transaction_id`` via an exclusive locking read. The conditional-write (``If-Match``) guard must compare the client's token against *committed* state. A plain consistent read is served from the transaction's REPEATABLE READ snapshot on MySQL/InnoDB (pinned by the request's earlier auth queries), so a version row committed by a concurrent writer between this request's first read - and its row lock stays invisible — the stale token then matches and - the 412 the guard exists to raise is missed. ``FOR SHARE`` - (``with_for_update(read=True)``) makes this a locking read, which is - exempt from the snapshot and returns current committed data, without - blocking other readers of the same version row. + and its row lock stays invisible -- the stale token then matches and + the 412 the guard exists to raise is missed. A locking read is exempt + from the snapshot and returns current committed data. + + The lock is exclusive (``with_for_update()``), not shared: this + transaction later closes the very row it reads here (Continuum's + validity strategy sets ``end_transaction_id`` at commit), and holding + a shared lock first invites InnoDB's shared-to-exclusive upgrade + deadlock whenever anything else queues for the row in between. Plain + MVCC readers are not blocked by either lock strength, and writers to + the same entity are already serialised by the entity row lock taken + first, so exclusivity here costs nothing. + + Residual, documented rather than removed: on MySQL a locking read + over an empty range (an entity with no live version row yet) takes a + gap lock, and two concurrent conditional writers whose version rows + share a primary-key gap can deadlock at Continuum's insert. The PUT + path maps lock-contention errors on this read to the same 412 retry + semantics a stale token gets -- a deadlock loser *is* a concurrent + interleave. The lock also briefly blocks retention pruning of this + entity's version rows for the duration of the request transaction. Deliberately a plain row query, not the aggregate - :func:`current_version_info` — locking clauses do not combine with - aggregates reliably across dialects, and the guard only needs the - live ``transaction_id``. Renders no ``FOR SHARE`` on SQLite, which - serialises writers anyway. + :func:`current_version_info` -- locking clauses do not combine with + aggregates (Postgres rejects the combination outright), and the guard + only needs the live ``transaction_id``. Ordered-and-limited so a + defensively tolerated multi-open-row state (which the aggregate's + ``max`` absorbs) degrades the same way instead of raising + ``MultipleResultsFound``. Renders the dialect's locking clause + (``FOR UPDATE`` / none on SQLite, which serialises writers anyway). """ ver_cls = version_class(model_cls) return ( @@ -254,7 +273,9 @@ def current_live_transaction_id_for_share( ver_cls.end_transaction_id.is_(None), ver_cls.operation_type != OPERATION_DELETE, ) - .with_for_update(read=True) + .order_by(ver_cls.transaction_id.desc()) + .limit(1) + .with_for_update() .scalar() ) diff --git a/tests/integration_tests/versioning/conditional_token_lock_tests.py b/tests/integration_tests/versioning/conditional_token_lock_tests.py index 93a409c8a126..467545604beb 100644 --- a/tests/integration_tests/versioning/conditional_token_lock_tests.py +++ b/tests/integration_tests/versioning/conditional_token_lock_tests.py @@ -20,12 +20,15 @@ tests/unit_tests/versioning/test_version_info_locking.py; the MySQL REPEATABLE READ staleness itself cannot flip on Postgres (READ COMMITTED takes a fresh snapshot per statement) and needs an interleaved read view -to reproduce on MySQL, so -- mirroring the honest-scope precedent of the -entity-lock fix -- these tests prove the locking read executes correctly -against a real backend and agrees with the plain read on the quiet path. +to reproduce on MySQL. So -- the same honest-scope approach the sibling +entity-lock PRs take -- these tests prove the locking read executes +correctly against a real backend, agrees with the plain read on the +quiet path, and resolves the no-live-row case the way the unversioned +token path expects. """ import pytest +import sqlalchemy as sa from superset import db from superset.daos.version import VersionDAO @@ -40,21 +43,29 @@ @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices") class TestConditionalTokenLockingRead(SupersetTestCase): - def _versioned_chart(self) -> Slice: + def _versioned_chart(self) -> tuple[Slice, str | None]: """A chart with at least one version row (a real committed save).""" chart = db.session.query(Slice).filter(Slice.slice_name == "Boys").one() + original = chart.description chart.description = "sc-120050 probe" db.session.commit() - return chart + return chart, original - def test_for_share_read_agrees_with_plain_read_on_quiet_path(self) -> None: + def _restore(self, chart_id: int, original: str | None) -> None: + db.session.rollback() + chart = db.session.get(Slice, chart_id) + assert chart is not None + chart.description = original + db.session.commit() + + def test_locked_read_agrees_with_plain_read_on_quiet_path(self) -> None: """Without concurrent writers both reads name the same live row, - and the FOR SHARE clause executes on the real backend (the + and the locking clause executes on the real backend (the dialect-syntax half of the guard the unit pins cannot cover).""" - chart = self._versioned_chart() + chart, original = self._versioned_chart() try: plain = VersionDAO.current_live_transaction_id(Slice, chart.id, chart.uuid) - locked = VersionDAO.current_live_transaction_id_for_share( + locked = VersionDAO.current_live_transaction_id_locked( Slice, chart.id, chart.uuid ) assert plain is not None @@ -65,7 +76,35 @@ def test_for_share_read_agrees_with_plain_read_on_quiet_path(self) -> None: ) assert info.transaction_id == plain finally: - db.session.rollback() - chart = db.session.query(Slice).get(chart.id) - chart.description = None + self._restore(chart.id, original) + + def test_no_live_row_resolves_to_none_under_lock(self) -> None: + """The zero-live-row branch (lazy baselines) stays well-behaved. + + An entity whose version rows are gone (or never existed -- + baselines are written lazily) must yield None from the locked + read, so current_entity_version_info reports no version_uuid and + the guard falls back to the unversioned token, exactly as the + plain path does. This is also the branch where MySQL takes a gap + lock (see current_live_transaction_id_locked's residual note). + """ + chart, original = self._versioned_chart() + try: + db.session.execute( + sa.text("DELETE FROM slices_version WHERE id = :id"), + {"id": chart.id}, + ) db.session.commit() + + locked = VersionDAO.current_live_transaction_id_locked( + Slice, chart.id, chart.uuid + ) + assert locked is None + + info = current_entity_version_info( + Slice, chart.id, chart.uuid, lock_for_stale_check=True + ) + assert info.version_uuid is None + assert info.entity_uuid == chart.uuid + finally: + self._restore(chart.id, original) diff --git a/tests/unit_tests/datasets/api_tests.py b/tests/unit_tests/datasets/api_tests.py index 002ae6d8d1aa..de05851e2ea4 100644 --- a/tests/unit_tests/datasets/api_tests.py +++ b/tests/unit_tests/datasets/api_tests.py @@ -468,7 +468,7 @@ def test_put_dataset_rejects_stale_if_match( version_uuid="new", entity_uuid=dataset.uuid, ), - ): + ) as version_info: response = client.put( f"/api/v1/dataset/{dataset.id}", json={"description": "from a stale tab"}, @@ -476,6 +476,10 @@ def test_put_dataset_rejects_stale_if_match( ) assert response.status_code == 412 + # Pins the endpoint wiring: a conditional write must ask for the + # locking validator read (sc-120050) -- without this, dropping the + # lock_for_stale_check kwarg at the call site survives every test. + assert version_info.call_args.kwargs["lock_for_stale_check"] is True assert response.headers["ETag"] == '"new"' db.session.expire(dataset) assert dataset.description is None diff --git a/tests/unit_tests/versioning/test_version_info_locking.py b/tests/unit_tests/versioning/test_version_info_locking.py index 113624070e72..5a083c7ad65a 100644 --- a/tests/unit_tests/versioning/test_version_info_locking.py +++ b/tests/unit_tests/versioning/test_version_info_locking.py @@ -20,11 +20,13 @@ snapshot pinned by the request's first read, so the version-info lookup behind ``raise_for_stale_write`` could miss a concurrently committed version row -- a stale client token then matched and the 412 was skipped, -recreating the lost update the lock + If-Match pair exists to prevent -(the validator-layer sibling of the entity fix in apache/superset#44027). -These are the dialect-independent statement-shape guards: the -conditional-path transaction-id read must carry ``with_for_update(read=True)`` -(a locking read is exempt from the snapshot) and +recreating the lost update the lock + If-Match pair exists to prevent. +(The entity read one layer down has the same staleness class; the fix +proposed in apache/superset#44027 addresses it there.) These are the +dialect-independent statement-shape guards: the conditional-path +transaction-id read must be an exclusive locking read (exempt from the +snapshot; exclusive so this transaction's own later close of the row +cannot deadlock on a shared-to-exclusive upgrade) and ``current_entity_version_info`` must thread the flag -- with the default (every GET path) never invoking the locking read at all. The real-database proof lives in @@ -37,7 +39,7 @@ from pytest_mock import MockerFixture from superset.versioning.api_helpers import current_entity_version_info -from superset.versioning.queries import current_live_transaction_id_for_share +from superset.versioning.queries import current_live_transaction_id_locked def _mock_queries_db(mocker: MockerFixture) -> MagicMock: @@ -47,27 +49,32 @@ def _mock_queries_db(mocker: MockerFixture) -> MagicMock: return db -def test_for_share_read_is_a_locking_row_read(mocker: MockerFixture) -> None: - """The conditional-path read chains with_for_update(read=True). +def test_locked_read_is_an_exclusive_locking_row_read(mocker: MockerFixture) -> None: + """The conditional-path read chains an exclusive with_for_update(). - Dropping the flag (or the whole locking clause) fails here on every - backend -- the analogue of the shape pins on the entity lock. + Exclusive, not read=True: this transaction later closes the row it + reads (Continuum sets end_transaction_id at commit), and a shared + lock first would invite the shared-to-exclusive upgrade deadlock. + Dropping the locking clause -- or weakening it back to shared -- + fails here on every backend. Ordered-and-limited so a defensive + multi-open-row state cannot raise MultipleResultsFound. """ db = _mock_queries_db(mocker) - current_live_transaction_id_for_share(MagicMock(), 1, uuid4()) + current_live_transaction_id_locked(MagicMock(), 1, uuid4()) query = db.session.query.return_value filtered = query.filter.return_value.filter.return_value - filtered.with_for_update.assert_called_once_with(read=True) - filtered.with_for_update.return_value.scalar.assert_called_once_with() + limited = filtered.order_by.return_value.limit.return_value + limited.with_for_update.assert_called_once_with() + limited.with_for_update.return_value.scalar.assert_called_once_with() def test_version_info_threads_the_lock_flag(mocker: MockerFixture) -> None: - """lock_for_stale_check=True swaps in the FOR SHARE transaction id.""" + """lock_for_stale_check=True swaps in the locked transaction id.""" dao = MagicMock() dao.current_version_info.return_value = (0, 5) - dao.current_live_transaction_id_for_share.return_value = 6 + dao.current_live_transaction_id_locked.return_value = 6 mocker.patch("superset.versioning.api_helpers.VersionDAO", new=dao) mocker.patch("superset.versioning.api_helpers._capture_enabled", return_value=True) entity_uuid = uuid4() @@ -76,9 +83,13 @@ def test_version_info_threads_the_lock_flag(mocker: MockerFixture) -> None: MagicMock(), 1, entity_uuid, lock_for_stale_check=True ) - dao.current_live_transaction_id_for_share.assert_called_once() - # The guard's input is the locked value, not the snapshot aggregate's. + dao.current_live_transaction_id_locked.assert_called_once() assert info.transaction_id == 6 + # The guard actually consumes version_uuid (concurrency_token_from), so + # pin that the token derives from the LOCKED id -- reordering the swap + # below the uuid derivation would break the guard with transaction_id + # still reporting the fresh value. + dao.derive_version_uuid.assert_called_once_with(entity_uuid, 6) def test_version_info_defaults_to_plain_reads(mocker: MockerFixture) -> None: @@ -90,5 +101,5 @@ def test_version_info_defaults_to_plain_reads(mocker: MockerFixture) -> None: info = current_entity_version_info(MagicMock(), 1, uuid4()) - dao.current_live_transaction_id_for_share.assert_not_called() + dao.current_live_transaction_id_locked.assert_not_called() assert info.transaction_id == 5 From 4e7ef3632f431b9ca97fe54eaab4dafc1c0e1dcd Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Wed, 9 Sep 2026 11:48:48 -0600 Subject: [PATCH 3/8] fix(dataset): map lock contention to a retryable 409, not 412 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-model round (author-verified): the documented cross-entity gap-lock deadlock surfaces at Continuum's version-row INSERT inside the update command — where the @transaction error mapping already yields the pre-existing 422 — while the OperationalError handler at the call site covers only a deadlock at the locking read itself. The residual doc implied the handler covered the insert case, and its 412 told the loser to refetch a token that was never proven stale. Re-map the read-point handler to 409 with retry-the-same-request guidance, rewrite the residual paragraph to name both surfacing points and their true outcomes, align is_lock_contention_error's docstring, and pin the 409 mapping (status + message) so it cannot drift back to 412. sc-120050 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW --- superset/datasets/api.py | 35 ++++++--- superset/translations/messages.pot | 4 ++ superset/versioning/api_helpers.py | 21 ------ superset/versioning/db_errors.py | 32 +++++++++ superset/versioning/queries.py | 12 ++-- .../conditional_token_lock_tests.py | 2 +- tests/unit_tests/datasets/api_tests.py | 72 +++++++++++++++++++ tests/unit_tests/versioning/test_db_errors.py | 64 ++++++++++++++++- 8 files changed, 207 insertions(+), 35 deletions(-) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index e228fa5482d9..8997e9ea2f4a 100644 --- a/superset/datasets/api.py +++ b/superset/datasets/api.py @@ -114,11 +114,11 @@ current_entity_version_info, entity_concurrency_token, get_version_endpoint, - is_lock_contention_error, 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, @@ -727,9 +727,12 @@ def put(self, pk: int) -> Response: # noqa: C901 # 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 on that read (deadlock / lock - # wait) IS a concurrent interleave, so it maps to the same 412 retry - # semantics as a stale 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 surfaces as the pre-existing 422 via the + # command's error mapping.) try: old_info = current_entity_version_info( SqlaTable, pk, lock_for_stale_check=conditional @@ -741,11 +744,10 @@ def put(self, pk: int) -> Response: # noqa: C901 # rollback); this clears the aborted session before responding. db.session.rollback() # pylint: disable=consider-using-transaction return self.response( - 412, + 409, message=_( - "The dataset was changed by another user or browser tab " - "after you opened it. Reopen it to pick up the latest " - "version, then reapply your changes." + "Another save is in progress for this dataset. " + "Retry the same request." ), ) @@ -822,6 +824,23 @@ def put(self, pk: int) -> Response: # noqa: C901 ) 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__): + 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." + ), + ) logger.error( "Error updating model %s: %s", self.__class__.__name__, diff --git a/superset/translations/messages.pot b/superset/translations/messages.pot index b17e5c617d45..8f077c3e79c9 100644 --- a/superset/translations/messages.pot +++ b/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 "" diff --git a/superset/versioning/api_helpers.py b/superset/versioning/api_helpers.py index bf53f335bb41..9f3b7d484b9c 100644 --- a/superset/versioning/api_helpers.py +++ b/superset/versioning/api_helpers.py @@ -205,27 +205,6 @@ def entity_concurrency_token( ) or unversioned_entity_token(entity_uuid) -def is_lock_contention_error(ex: Exception) -> bool: - """Whether *ex* is a database deadlock / lock-wait failure. - - A conditional write that loses a lock race (e.g. MySQL's gap-lock - deadlock on the version table, error 1213, or a lock wait timeout, - 1205; Postgres serialization/deadlock SQLSTATEs 40001/40P01) has, by - definition, interleaved with a concurrent writer -- exactly the - situation ``If-Match`` exists to surface. Callers map it to the same - retry response as a stale token rather than a 500. - """ - orig = getattr(ex, "orig", None) - code = getattr(orig, "args", [None])[0] if orig is not None else None - if code in (1213, 1205): - return True - sqlstate = getattr(orig, "pgcode", None) - if sqlstate in ("40001", "40P01"): - return True - text = str(ex).lower() - return "deadlock" in text or "lock wait timeout" in text - - def lock_entity_for_update( model_cls: type[Model], entity_id: int | None ) -> Model | None: diff --git a/superset/versioning/db_errors.py b/superset/versioning/db_errors.py index 4b840edb81a3..19b12a6c79cc 100644 --- a/superset/versioning/db_errors.py +++ b/superset/versioning/db_errors.py @@ -34,6 +34,38 @@ from sqlalchemy.exc import DBAPIError +#: MySQL/MariaDB deadlock and lock-wait-timeout error codes. +_MYSQL_LOCK_CONTENTION = (1213, 1205) + +#: PostgreSQL SQLSTATEs: serialization_failure, deadlock_detected, +#: lock_not_available. +_PG_LOCK_CONTENTION = ("40001", "40P01", "55P03") + + +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. Accepts ``None`` + (e.g. an exception with no ``__cause__``) and errors with empty + driver args without raising. + """ + if exc is None: + return False + orig = getattr(exc, "orig", None) + args = getattr(orig, "args", None) + if args and args[0] in _MYSQL_LOCK_CONTENTION: + return True + sqlstate = getattr(orig, "pgcode", None) or getattr(orig, "sqlstate", None) + if sqlstate in _PG_LOCK_CONTENTION: + return True + text = str(exc).lower() + return "deadlock" in text or "lock wait timeout" in text + + #: PostgreSQL SQLSTATE for "relation does not exist" (undefined_table). _PG_UNDEFINED_TABLE = "42P01" diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index a7431702a27e..6e53285eaffc 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -250,10 +250,14 @@ def current_live_transaction_id_locked( Residual, documented rather than removed: on MySQL a locking read over an empty range (an entity with no live version row yet) takes a gap lock, and two concurrent conditional writers whose version rows - share a primary-key gap can deadlock at Continuum's insert. The PUT - path maps lock-contention errors on this read to the same 412 retry - semantics a stale token gets -- a deadlock loser *is* a concurrent - interleave. The lock also briefly blocks retention pruning of this + share a primary-key gap can deadlock. That deadlock has two surfacing + points with different outcomes. At THIS read (rare -- gap locks are + mutually compatible, so both readers usually succeed) the PUT path + maps it to a retryable 409. At Continuum's version-row INSERT inside + the update command it surfaces as the command's pre-existing 422 + error mapping. In both cases the loser's ``If-Match`` token is NOT + proven stale and the correct client action is to retry the same + request. The lock also briefly blocks retention pruning of this entity's version rows for the duration of the request transaction. Deliberately a plain row query, not the aggregate diff --git a/tests/integration_tests/versioning/conditional_token_lock_tests.py b/tests/integration_tests/versioning/conditional_token_lock_tests.py index 467545604beb..1e7352c0f79c 100644 --- a/tests/integration_tests/versioning/conditional_token_lock_tests.py +++ b/tests/integration_tests/versioning/conditional_token_lock_tests.py @@ -14,7 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -"""sc-120050: real-database proof for the FOR SHARE validator read. +"""sc-120050: real-database proof for the locked validator read. The statement-shape pins live in tests/unit_tests/versioning/test_version_info_locking.py; the MySQL diff --git a/tests/unit_tests/datasets/api_tests.py b/tests/unit_tests/datasets/api_tests.py index de05851e2ea4..cb2e65ea5dae 100644 --- a/tests/unit_tests/datasets/api_tests.py +++ b/tests/unit_tests/datasets/api_tests.py @@ -485,6 +485,78 @@ def test_put_dataset_rejects_stale_if_match( assert dataset.description is None +def test_put_dataset_maps_lock_contention_to_retryable_409( + session: Session, + client: Any, + full_api_access: None, +) -> None: + """sc-120050: a deadlock at the locked validator read is a 409, not 412. + + A lock race lost at the locking read proves concurrent contention, not + that this request's If-Match token is stale -- 412's refetch-the-token + guidance would be wrong, so the mapping must not drift back to it. The + message tells the client to retry the same request. + """ + from sqlalchemy.exc import OperationalError + + dataset = _create_dataset("test_put_lock_contention") + deadlock = OperationalError("stmt", None, Exception()) + deadlock.orig = type("Orig", (), {"args": (1213, "Deadlock found")})() + + with patch( + "superset.datasets.api.current_entity_version_info", + side_effect=deadlock, + ): + response = client.put( + f"/api/v1/dataset/{dataset.id}", + json={"description": "from a racing tab"}, + headers={"If-Match": '"anything"'}, + ) + + assert response.status_code == 409 + body = response.get_data(as_text=True) + assert "Retry the same request" in body + # No write-check re-read here: the handler's rollback detaches the + # fixture object, and nothing could have written anyway — the patched + # read raised before the update command ever ran. + + +def test_put_dataset_without_if_match_never_locks( + session: Session, + client: Any, + full_api_access: None, +) -> None: + """An unconditional PUT takes no locks and no locking validator read. + + Pins the perf contract both ways: hard-coding lock_for_stale_check=True + at the call site (or unconditionally locking the entity) fails here. + """ + from superset.versioning.api_helpers import EntityVersionInfo + + dataset = _create_dataset("test_put_unconditional_no_locks") + + with ( + patch("superset.datasets.api.lock_entity_for_update") as lock, + patch( + "superset.datasets.api.current_entity_version_info", + return_value=EntityVersionInfo(entity_uuid=dataset.uuid), + ) as version_info, + patch("superset.datasets.api.UpdateDatasetCommand") as cmd, + ): + cmd.return_value.run.return_value = dataset + response = client.put( + f"/api/v1/dataset/{dataset.id}", + json={"description": "plain save"}, + ) + + assert response.status_code == 200 + lock.assert_not_called() + # First call is the pre-command old_info read (the one carrying the + # flag); the endpoint calls the helper again after the command for + # new_info, without the kwarg. + assert version_info.call_args_list[0].kwargs["lock_for_stale_check"] is False + + def test_put_dataset_guards_a_dataset_with_no_version_rows( session: Session, client: Any, diff --git a/tests/unit_tests/versioning/test_db_errors.py b/tests/unit_tests/versioning/test_db_errors.py index a49408489685..0457cb7561c3 100644 --- a/tests/unit_tests/versioning/test_db_errors.py +++ b/tests/unit_tests/versioning/test_db_errors.py @@ -21,9 +21,13 @@ from typing import Any +import pytest from sqlalchemy.exc import OperationalError, ProgrammingError -from superset.versioning.db_errors import is_missing_table_error +from superset.versioning.db_errors import ( + is_lock_contention_error, + is_missing_table_error, +) class _FakeDriverError(Exception): @@ -107,3 +111,61 @@ def test_connection_drop_with_no_code_is_not_missing_table() -> None: _FakeDriverError("server closed the connection unexpectedly"), ) assert is_missing_table_error(error) is False + + +# --------------------------------------------------------------------------- +# is_lock_contention_error (sc-120050) +# --------------------------------------------------------------------------- + + +def _op_error(orig: object) -> OperationalError: + err = OperationalError("stmt", None, Exception("boom")) + err.orig = orig + return err + + +class _Orig: + def __init__(self, args=(), pgcode=None, sqlstate=None): + self.args = args + if pgcode is not None: + self.pgcode = pgcode + if sqlstate is not None: + self.sqlstate = sqlstate + + +@pytest.mark.parametrize( + "orig", + [ + _Orig(args=(1213, "Deadlock found when trying to get lock")), + _Orig(args=(1205, "Lock wait timeout exceeded")), + _Orig(pgcode="40001"), + _Orig(pgcode="40P01"), + _Orig(pgcode="55P03"), + _Orig(sqlstate="40P01"), + ], +) +def test_lock_contention_positive(orig: _Orig) -> None: + assert is_lock_contention_error(_op_error(orig)) is True + + +def test_lock_contention_text_fallback() -> None: + err = OperationalError("deadlock detected somewhere", None, Exception()) + err.orig = None + assert is_lock_contention_error(err) is True + + +@pytest.mark.parametrize( + "orig", + [ + _Orig(args=(1146, "Table does not exist")), + _Orig(args=()), + _Orig(pgcode="42P01"), + None, + ], +) +def test_lock_contention_negative(orig: object) -> None: + assert is_lock_contention_error(_op_error(orig)) is False + + +def test_lock_contention_none_input() -> None: + assert is_lock_contention_error(None) is False From d8730cac8019897292a6eaf8d937ddab27735a14 Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Fri, 11 Sep 2026 12:45:59 -0600 Subject: [PATCH 4/8] chore: retrigger CI (runner mirror outage cleared; prior run failed at setup) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW From 0970543ced890b2af4f335cba934697073c55aaf Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Mon, 14 Sep 2026 10:18:27 -0600 Subject: [PATCH 5/8] fix(dataset): map entity-lock contention to the retryable 409; shared handler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round (fitzee): - The entity-row lock acquisition sat OUTSIDE the contention mapping: writer B blocking on writer A's row lock and timing out (1205) — the most common contention outcome of all three lock points — propagated as an uncaught 500 instead of the retryable 409 this PR's contract advertises. Now wrapped with the same classification. - The three lock points share _lock_contention_response(): one rollback + one i18n string instead of two hand-synced copies. Its docstring states the rollback is LOAD-BEARING for lock-wait-timeout (with innodb_rollback_on_timeout OFF a 1205 rolls back only the failing statement; the transaction still holds the entity row lock and the rollback is what releases it) — replacing the misleading "transaction is already dead" cleanup framing. - Stale comments fixed: queries.py's residual note and the read-point parenthetical claimed the insert-point deadlock surfaces as the pre-existing 422; the handler added in this same PR classifies it to the same 409. - is_lock_contention_error's text fallback now recognizes SQLite's "database (table) is locked" (no error code exists to match), so contention on the SQLite-backed integration suite classifies correctly; parametrized tests added. - put() keeps # noqa: C901 (15 > 10 even after the extraction). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW --- superset/datasets/api.py | 66 +++++++++++-------- superset/versioning/db_errors.py | 10 ++- superset/versioning/queries.py | 7 +- tests/unit_tests/datasets/api_tests.py | 33 ++++++++++ tests/unit_tests/versioning/test_db_errors.py | 17 ++++- 5 files changed, 101 insertions(+), 32 deletions(-) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index 8997e9ea2f4a..7e8ad52daead 100644 --- a/superset/datasets/api.py +++ b/superset/datasets/api.py @@ -573,6 +573,25 @@ 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. + + 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. + """ + 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("/", methods=("PUT",)) @protect() @statsd_metrics @@ -715,12 +734,22 @@ def put(self, pk: int) -> Response: # noqa: C901 # lock_entity_for_update.) conditional = is_conditional_write() if conditional: - # 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) + # 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 = 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). On the conditional path the @@ -731,8 +760,9 @@ def put(self, pk: int) -> Response: # noqa: C901 # 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 surfaces as the pre-existing 422 via the - # command's error mapping.) + # insert inside the command is classified by the + # DatasetUpdateFailedError handler below via ``__cause__`` and + # returns the same 409.) try: old_info = current_entity_version_info( SqlaTable, pk, lock_for_stale_check=conditional @@ -740,16 +770,7 @@ def put(self, pk: int) -> Response: # noqa: C901 except OperationalError as ex: if not (conditional and is_lock_contention_error(ex)): raise - # Not a unit of work: the transaction is already dead (deadlock - # rollback); this clears the aborted session before responding. - 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." - ), - ) + return self._lock_contention_response() try: raise_for_stale_write(concurrency_token_from(old_info)) @@ -833,14 +854,7 @@ def put(self, pk: int) -> Response: # noqa: C901 # 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__): - 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." - ), - ) + return self._lock_contention_response() logger.error( "Error updating model %s: %s", self.__class__.__name__, diff --git a/superset/versioning/db_errors.py b/superset/versioning/db_errors.py index 19b12a6c79cc..8d0fdf4179a9 100644 --- a/superset/versioning/db_errors.py +++ b/superset/versioning/db_errors.py @@ -63,7 +63,15 @@ def is_lock_contention_error(exc: BaseException | None) -> bool: if sqlstate in _PG_LOCK_CONTENTION: return True text = str(exc).lower() - return "deadlock" in text or "lock wait timeout" in text + # The string fallbacks cover drivers whose args/sqlstate shapes are + # not modelled above; "database (table) is locked" is SQLite's write + # contention, which carries no code at all. + return ( + "deadlock" in text + or "lock wait timeout" in text + or "database is locked" in text + or "database table is locked" in text + ) #: PostgreSQL SQLSTATE for "relation does not exist" (undefined_table). diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index 6e53285eaffc..7e6675557648 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -251,11 +251,12 @@ def current_live_transaction_id_locked( over an empty range (an entity with no live version row yet) takes a gap lock, and two concurrent conditional writers whose version rows share a primary-key gap can deadlock. That deadlock has two surfacing - points with different outcomes. At THIS read (rare -- gap locks are + points with the same outcome. At THIS read (rare -- gap locks are mutually compatible, so both readers usually succeed) the PUT path maps it to a retryable 409. At Continuum's version-row INSERT inside - the update command it surfaces as the command's pre-existing 422 - error mapping. In both cases the loser's ``If-Match`` token is NOT + the update command the driver error is chained as the command + failure's ``__cause__``, which the PUT handler classifies to the + same retryable 409. Either way the loser's ``If-Match`` token is NOT proven stale and the correct client action is to retry the same request. The lock also briefly blocks retention pruning of this entity's version rows for the duration of the request transaction. diff --git a/tests/unit_tests/datasets/api_tests.py b/tests/unit_tests/datasets/api_tests.py index cb2e65ea5dae..2e063ae42cc5 100644 --- a/tests/unit_tests/datasets/api_tests.py +++ b/tests/unit_tests/datasets/api_tests.py @@ -521,6 +521,39 @@ def test_put_dataset_maps_lock_contention_to_retryable_409( # read raised before the update command ever ran. +def test_put_dataset_maps_entity_lock_contention_to_retryable_409( + session: Session, + client: Any, + full_api_access: None, +) -> None: + """A lock race lost at the ENTITY-LOCK acquisition is also a 409. + + The most common contention outcome of all: writer B blocks on + writer A's entity row lock and times out (1205) at + ``lock_entity_for_update`` -- before the validator read's own + try/except. Left unmapped this surfaced as an uncaught 500, defeating + the retryable-conflict contract the other two lock points advertise. + """ + from sqlalchemy.exc import OperationalError + + dataset = _create_dataset("test_put_entity_lock_contention") + timeout = OperationalError("stmt", None, Exception()) + timeout.orig = type("Orig", (), {"args": (1205, "Lock wait timeout exceeded")})() + + with patch( + "superset.datasets.api.lock_entity_for_update", + side_effect=timeout, + ): + response = client.put( + f"/api/v1/dataset/{dataset.id}", + json={"description": "from a racing tab"}, + headers={"If-Match": '"anything"'}, + ) + + assert response.status_code == 409 + assert "Retry the same request" in response.get_data(as_text=True) + + def test_put_dataset_without_if_match_never_locks( session: Session, client: Any, diff --git a/tests/unit_tests/versioning/test_db_errors.py b/tests/unit_tests/versioning/test_db_errors.py index 0457cb7561c3..17d5b51d816f 100644 --- a/tests/unit_tests/versioning/test_db_errors.py +++ b/tests/unit_tests/versioning/test_db_errors.py @@ -148,8 +148,21 @@ def test_lock_contention_positive(orig: _Orig) -> None: assert is_lock_contention_error(_op_error(orig)) is True -def test_lock_contention_text_fallback() -> None: - err = OperationalError("deadlock detected somewhere", None, Exception()) +@pytest.mark.parametrize( + "text", + [ + "deadlock detected somewhere", + "Lock wait timeout exceeded; try restarting transaction", + # SQLite write contention carries no error code at all -- only + # these message texts. The integration suite runs on SQLite, so + # misclassifying them would route a genuine contention test to + # the 500/422 paths instead of the 409. + "(sqlite3.OperationalError) database is locked", + "(sqlite3.OperationalError) database table is locked: table_versions", + ], +) +def test_lock_contention_text_fallback(text: str) -> None: + err = OperationalError(text, None, Exception()) err.orig = None assert is_lock_contention_error(err) is True From 828a7ed3716f9fdf7c78bd9bd53dea44accfecca Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Tue, 15 Sep 2026 11:36:18 -0600 Subject: [PATCH 6/8] fix(versioning): classify lock contention on the driver diagnostic, not the wrapper's rendered SQL (sc-120050) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Richard Fogaça's review: is_lock_contention_error's fallback searched str(exc), and 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 column called lock_wait_timeout) was answered 409 "Retry the same request", advice that can never succeed because nothing is contending. Reachable at the response boundary: the command wraps SQLAlchemy errors as DatasetUpdateFailedError and the endpoint classifies its __cause__. Classification now reads the DRIVER DIAGNOSTIC only. A structured code is authoritative: a MySQL errno or PostgreSQL SQLSTATE outside the contention sets ends classification at False instead of falling through to a keyword match. Only code-less drivers (SQLite) reach the phrase check, and then against their own message. A wrapper whose driver error was not preserved is unclassifiable, not retryable; a raw driver exception is still its own diagnostic. Tests: the fallback positives now put their message in the driver error rather than the statement argument; new negatives cover each phrase appearing only in SQL or parameters on an unrelated DataError and OperationalError, a non-contention code whose diagnostic mentions a deadlock (pinning code precedence), and a wrapper with no orig. At the response boundary, a contaminated non-lock DataError keeps its 422 while genuine contention stays 409. 11 of these fail without the fix. Also, the interleaved MySQL regression the lens pass found missing: two sessions, deterministic, no sleeps. Writer A commits a competing version between this request's read view opening and its validator read; the plain read is served from the pre-commit snapshot and still names v1 — so the client's stale If-Match token MATCHES and the update would clobber A — while the locking read derives v2's token and the same token is rejected. Zero-history-to-first-version covered too (the MySQL gap-lock branch). Both force REPEATABLE READ per-session, since CI pins the MySQL server to READ COMMITTED and the staleness cannot flip under it. Verified against a real MySQL 8: 4/4 pass, and with the locking clause removed both new tests fail while both pre-existing ones still pass — the verification gap, closed. Plus: the PUT contract documents the retryable 409 and contrasts it with 412's refetch guidance, and the new-code annotations the lens pass listed. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JRLEJS4mUqKBoPjSjviKUW --- superset/datasets/api.py | 32 ++- superset/versioning/db_errors.py | 75 ++++-- superset/versioning/queries.py | 2 +- .../conditional_token_lock_tests.py | 250 +++++++++++++++++- tests/unit_tests/datasets/api_tests.py | 49 ++++ tests/unit_tests/versioning/test_db_errors.py | 164 ++++++++++-- .../versioning/test_version_info_locking.py | 27 +- 7 files changed, 529 insertions(+), 70 deletions(-) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index 7e8ad52daead..a9afab95b870 100644 --- a/superset/datasets/api.py +++ b/superset/datasets/api.py @@ -619,9 +619,11 @@ def put(self, pk: int) -> Response: # noqa: C901 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 @@ -698,10 +700,30 @@ def put(self, pk: int) -> Response: # noqa: C901 $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: @@ -732,7 +754,7 @@ def put(self, pk: int) -> Response: # noqa: C901 # (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.) - conditional = is_conditional_write() + 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 diff --git a/superset/versioning/db_errors.py b/superset/versioning/db_errors.py index 8d0fdf4179a9..c2035f7602ba 100644 --- a/superset/versioning/db_errors.py +++ b/superset/versioning/db_errors.py @@ -35,11 +35,22 @@ from sqlalchemy.exc import DBAPIError #: MySQL/MariaDB deadlock and lock-wait-timeout error codes. -_MYSQL_LOCK_CONTENTION = (1213, 1205) +_MYSQL_LOCK_CONTENTION: tuple[int, ...] = (1213, 1205) #: PostgreSQL SQLSTATEs: serialization_failure, deadlock_detected, #: lock_not_available. -_PG_LOCK_CONTENTION = ("40001", "40P01", "55P03") +_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: @@ -49,29 +60,53 @@ def is_lock_contention_error(exc: BaseException | None) -> bool: 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. Accepts ``None`` - (e.g. an exception with no ``__cause__``) and errors with empty - driver args without raising. + 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 = getattr(exc, "orig", None) - args = getattr(orig, "args", None) - if args and args[0] in _MYSQL_LOCK_CONTENTION: - return True - sqlstate = getattr(orig, "pgcode", None) or getattr(orig, "sqlstate", None) - if sqlstate in _PG_LOCK_CONTENTION: - return True - text = str(exc).lower() - # The string fallbacks cover drivers whose args/sqlstate shapes are - # not modelled above; "database (table) is locked" is SQLite's write - # contention, which carries no code at all. - return ( - "deadlock" in text - or "lock wait timeout" in text - or "database is locked" in text - or "database table is locked" in text + 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: BaseException = exc + else: + diagnostic = orig + + args = 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 = 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(diagnostic).lower() + return any(phrase in text for phrase in _LOCK_CONTENTION_PHRASES) #: PostgreSQL SQLSTATE for "relation does not exist" (undefined_table). diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index 7e6675557648..649e1fbd662f 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -270,7 +270,7 @@ def current_live_transaction_id_locked( ``MultipleResultsFound``. Renders the dialect's locking clause (``FOR UPDATE`` / none on SQLite, which serialises writers anyway). """ - ver_cls = version_class(model_cls) + ver_cls: type[Any] = version_class(model_cls) return ( db.session.query(ver_cls.transaction_id) .filter(identity_filter(ver_cls, entity_id, entity_uuid)) diff --git a/tests/integration_tests/versioning/conditional_token_lock_tests.py b/tests/integration_tests/versioning/conditional_token_lock_tests.py index 1e7352c0f79c..094539185215 100644 --- a/tests/integration_tests/versioning/conditional_token_lock_tests.py +++ b/tests/integration_tests/versioning/conditional_token_lock_tests.py @@ -17,28 +17,105 @@ """sc-120050: real-database proof for the locked validator read. The statement-shape pins live in -tests/unit_tests/versioning/test_version_info_locking.py; the MySQL -REPEATABLE READ staleness itself cannot flip on Postgres (READ COMMITTED -takes a fresh snapshot per statement) and needs an interleaved read view -to reproduce on MySQL. So -- the same honest-scope approach the sibling -entity-lock PRs take -- these tests prove the locking read executes -correctly against a real backend, agrees with the plain read on the -quiet path, and resolves the no-live-row case the way the unversioned -token path expects. +tests/unit_tests/versioning/test_version_info_locking.py. These tests +prove the locking read executes correctly against a real backend, +agrees with the plain read on the quiet path, resolves the no-live-row +case the way the unversioned token path expects, and -- the pins that +actually exercise the bug -- that an interleaved competing commit is +INVISIBLE to the plain snapshot read while the locking read sees it, so +a stale If-Match token is accepted by the old path and rejected by the +new one. + +The interleaved pins force REPEATABLE READ on the request session: +MySQL's staleness cannot flip under READ COMMITTED (a fresh snapshot per +statement), and CI pins the MySQL server to RC (``bashlib.sh``), so +without forcing it here a mutant that dropped ``with_for_update()`` +would pass every lane. MySQL-only: PostgreSQL's RR raises a +serialization failure for a locking read of a concurrently updated row +instead of returning it -- a different, also-safe outcome. """ import pytest import sqlalchemy as sa +from sqlalchemy_continuum import version_class, versioning_manager from superset import db from superset.daos.version import VersionDAO from superset.models.slice import Slice -from superset.versioning.api_helpers import current_entity_version_info +from superset.utils.dates import naive_utcnow +from superset.versioning.api_helpers import ( + concurrency_token_from, + current_entity_version_info, +) +from superset.versioning.baseline import CONTINUUM_BOOKKEEPING_COLUMNS +from superset.versioning.etag import raise_for_stale_write, StaleEntityError from tests.integration_tests.base_tests import SupersetTestCase from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa: F401 load_birth_names_dashboard_with_slices, load_birth_names_data, ) +from tests.integration_tests.test_app import app + + +def _commit_competing_version_out_of_band( + chart_id: int, template: dict[str, object] | None = None +) -> int: + """Commit a new live version row through a SECOND connection. + + Stands in for the concurrent request that wins the race: a separate + engine-level transaction closes the current live shadow row and + opens a new one, which is exactly the shape Continuum commits on a + save. Core SQL (not the ORM) so the write is invisible to the + request session until something re-reads -- and under REPEATABLE + READ, a plain re-read does not see it at all. + + *template* supplies the content columns when no live shadow row + survives (the zero-history transition); otherwise the current live + row is copied. + """ + ver_tbl: sa.Table = version_class(Slice).__table__ + tx_tbl: sa.Table = versioning_manager.transaction_cls.__table__ + with db.engine.begin() as conn: + new_tx: int = conn.execute( + tx_tbl.insert().values(issued_at=naive_utcnow(), user_id=None) + ).inserted_primary_key[0] + content: dict[str, object] + if template is None: + live = ( + conn.execute( + sa.select(ver_tbl).where( + ver_tbl.c.id == chart_id, + ver_tbl.c.end_transaction_id.is_(None), + ) + ) + .mappings() + .first() + ) + assert live is not None, "fixture must leave a live shadow row" + content = dict(live) + conn.execute( + sa.update(ver_tbl) + .where( + ver_tbl.c.id == chart_id, + ver_tbl.c.end_transaction_id.is_(None), + ) + .values(end_transaction_id=new_tx) + ) + else: + content = dict(template) + conn.execute( + ver_tbl.insert().values( + **{ + key: value + for key, value in content.items() + if key not in CONTINUUM_BOOKKEEPING_COLUMNS + }, + transaction_id=new_tx, + end_transaction_id=None, + operation_type=1, + ) + ) + return new_tx @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices") @@ -108,3 +185,158 @@ def test_no_live_row_resolves_to_none_under_lock(self) -> None: assert info.entity_uuid == chart.uuid finally: self._restore(chart.id, original) + + def _skip_unless_mysql(self) -> None: + """The staleness under test is MySQL-REPEATABLE-READ-specific.""" + if db.session.get_bind().dialect.name != "mysql": + pytest.skip("interleaved snapshot staleness is MySQL-RR-specific") + + def _force_repeatable_read(self) -> None: + """Open a fresh REPEATABLE READ transaction on the request session. + + CI pins the MySQL server itself to READ COMMITTED + (``bashlib.sh``), so the isolation has to be forced per-session + or the pin proves nothing. Call this only when nothing is pending + on the session: the rollback that starts the new transaction + would otherwise discard uncommitted fixture state (the + birth_names fixture stages its rows without committing). + """ + db.session.rollback() + db.session.connection(execution_options={"isolation_level": "REPEATABLE READ"}) + + def test_locking_read_sees_the_interleaved_commit_the_snapshot_misses( + self, + ) -> None: + """The bug, end to end: v1's token must not survive v2's commit. + + Writer A commits a competing version between this request's read + view opening and its validator read. Under REPEATABLE READ the + plain read is served from the pre-commit snapshot and still + names v1 -- so the client's stale If-Match token MATCHES and the + update proceeds, silently clobbering A (the lost update this PR + exists to stop). The locking read is snapshot-exempt, derives + v2's token, and the same stale token is rejected. + """ + self._skip_unless_mysql() + # v1 is committed BEFORE the read view opens: the fixture's rows + # and this save must both be visible to the snapshot the test + # then pins. + chart, original = self._versioned_chart() + chart_id: int = chart.id + chart_uuid = chart.uuid + try: + self._force_repeatable_read() + # Opening the read view: this first consistent read is what + # MySQL pins every later plain read in the transaction to. + v1: int | None = VersionDAO.current_live_transaction_id( + Slice, chart_id, chart_uuid + ) + assert v1 is not None + stale_info = current_entity_version_info(Slice, chart_id, chart_uuid) + stale_token: str | None = concurrency_token_from(stale_info) + assert stale_token is not None + + v2: int = _commit_competing_version_out_of_band(chart_id) + assert v2 != v1 + + # The snapshot has not moved... + assert ( + VersionDAO.current_live_transaction_id(Slice, chart_id, chart_uuid) + == v1 + ), "plain read left REPEATABLE READ; the pin below proves nothing" + # ...but the locking read is exempt from it. + assert ( + VersionDAO.current_live_transaction_id_locked( + Slice, chart_id, chart_uuid + ) + == v2 + ) + + fresh_info = current_entity_version_info( + Slice, chart_id, chart_uuid, lock_for_stale_check=True + ) + fresh_token: str | None = concurrency_token_from(fresh_info) + assert fresh_token is not None + assert fresh_token != stale_token + + # The client replays the token it read before A committed. + with app.test_request_context(headers={"If-Match": f'"{stale_token}"'}): + # Pre-PR path: accepted, and A's version is overwritten. + raise_for_stale_write(stale_token) + # Post-PR path: the locked validator rejects it. + with pytest.raises(StaleEntityError): + raise_for_stale_write(fresh_token) + finally: + self._restore(chart_id, original) + + def test_first_version_transition_under_forced_repeatable_read(self) -> None: + """Zero-history → first version: the same exemption must hold. + + The entity starts with no live version row (baselines are + written lazily), so the request's snapshot resolves the + unversioned token. Writer A then commits the entity's FIRST + version row -- the branch where MySQL takes a gap lock on the + empty range. The plain read still reports nothing; the locking + read must surface the new row, so the unversioned token stops + being accepted. + """ + self._skip_unless_mysql() + chart, original = self._versioned_chart() + chart_id: int = chart.id + chart_uuid = chart.uuid + ver_tbl: sa.Table = version_class(Slice).__table__ + try: + template = ( + db.session.execute( + sa.select(ver_tbl).where( + ver_tbl.c.id == chart_id, + ver_tbl.c.end_transaction_id.is_(None), + ) + ) + .mappings() + .first() + ) + assert template is not None + content: dict[str, object] = dict(template) + db.session.execute( + sa.text("DELETE FROM slices_version WHERE id = :id"), {"id": chart_id} + ) + db.session.commit() + self._force_repeatable_read() + + assert ( + VersionDAO.current_live_transaction_id(Slice, chart_id, chart_uuid) + is None + ) + unversioned_info = current_entity_version_info(Slice, chart_id, chart_uuid) + unversioned_token: str | None = concurrency_token_from(unversioned_info) + assert unversioned_token is not None + + first_tx: int = _commit_competing_version_out_of_band( + chart_id, template=content + ) + + assert ( + VersionDAO.current_live_transaction_id(Slice, chart_id, chart_uuid) + is None + ), "plain read left REPEATABLE READ; the pin below proves nothing" + assert ( + VersionDAO.current_live_transaction_id_locked( + Slice, chart_id, chart_uuid + ) + == first_tx + ) + + fresh_info = current_entity_version_info( + Slice, chart_id, chart_uuid, lock_for_stale_check=True + ) + fresh_token: str | None = concurrency_token_from(fresh_info) + assert fresh_token is not None + assert fresh_token != unversioned_token + with app.test_request_context( + headers={"If-Match": f'"{unversioned_token}"'} + ): + with pytest.raises(StaleEntityError): + raise_for_stale_write(fresh_token) + finally: + self._restore(chart_id, original) diff --git a/tests/unit_tests/datasets/api_tests.py b/tests/unit_tests/datasets/api_tests.py index 2e063ae42cc5..a7be5245af6a 100644 --- a/tests/unit_tests/datasets/api_tests.py +++ b/tests/unit_tests/datasets/api_tests.py @@ -590,6 +590,55 @@ def test_put_dataset_without_if_match_never_locks( assert version_info.call_args_list[0].kwargs["lock_for_stale_check"] is False +def test_put_dataset_keeps_422_when_sql_text_merely_mentions_a_lock( + session: Session, + client: Any, + full_api_access: None, +) -> None: + """Response boundary: a contaminated non-lock failure stays a 422. + + sc-120050 review (Richard Fogaça): an unrelated ``DataError`` whose + SQL or bound parameters happen to contain "deadlock" used to be + classified as contention and answered 409 "Retry the same request" + — advice that can never succeed, because nothing is contending. The + command wraps SQLAlchemy errors as ``DatasetUpdateFailedError`` and + the endpoint classifies its ``__cause__``, so the regression belongs + at this boundary as well as in the classifier's own unit tests. + """ + from sqlalchemy.exc import DataError + + from superset.commands.dataset.exceptions import DatasetUpdateFailedError + + dataset = _create_dataset("test_put_contaminated_data_error") + driver_error: Exception = type( + "Orig", (Exception,), {"args": (1064, "You have an error in your SQL syntax")} + )() + contaminated: DataError = DataError( + "UPDATE slices SET description = %(description)s", + {"description": "deadlock analysis"}, + driver_error, + ) + assert "deadlock" in str(contaminated).lower() + failure: DatasetUpdateFailedError = DatasetUpdateFailedError() + failure.__cause__ = contaminated + + with ( + # Keep the request on the CONDITIONAL path (that is where the + # classifier runs) without needing a live token: the stale check + # itself is not what this test pins. + patch("superset.datasets.api.raise_for_stale_write"), + patch("superset.datasets.api.UpdateDatasetCommand.run", side_effect=failure), + ): + response = client.put( + f"/api/v1/dataset/{dataset.id}", + json={"description": "deadlock analysis"}, + headers={"If-Match": '"anything"'}, + ) + + assert response.status_code == 422, response.get_data(as_text=True) + assert "Retry the same request" not in response.get_data(as_text=True) + + def test_put_dataset_guards_a_dataset_with_no_version_rows( session: Session, client: Any, diff --git a/tests/unit_tests/versioning/test_db_errors.py b/tests/unit_tests/versioning/test_db_errors.py index 17d5b51d816f..963a5f84000b 100644 --- a/tests/unit_tests/versioning/test_db_errors.py +++ b/tests/unit_tests/versioning/test_db_errors.py @@ -22,7 +22,7 @@ from typing import Any import pytest -from sqlalchemy.exc import OperationalError, ProgrammingError +from sqlalchemy.exc import DataError, OperationalError, ProgrammingError, StatementError from superset.versioning.db_errors import ( is_lock_contention_error, @@ -119,37 +119,80 @@ def test_connection_drop_with_no_code_is_not_missing_table() -> None: def _op_error(orig: object) -> OperationalError: - err = OperationalError("stmt", None, Exception("boom")) + err: OperationalError = OperationalError("stmt", None, Exception("boom")) err.orig = orig return err -class _Orig: - def __init__(self, args=(), pgcode=None, sqlstate=None): - self.args = args +class _FakeLockDriverError(Exception): # noqa: N818 — a driver error, not ours + """A stand-in for a DBAPI driver error. + + A real driver error IS an exception carrying its own diagnostic, so + the fake is one too: the classifier reads ``args`` / ``pgcode`` / + ``sqlstate`` and, for code-less drivers, ``str()`` of THIS object — + never of the SQLAlchemy wrapper around it. + """ + + def __init__( + self, + args: tuple[object, ...] = (), + pgcode: str | None = None, + sqlstate: str | None = None, + message: str = "", + ) -> None: + super().__init__(*args) + self.args: tuple[object, ...] = args + self._message: str = message if pgcode is not None: - self.pgcode = pgcode + self.pgcode: str = pgcode if sqlstate is not None: - self.sqlstate = sqlstate + self.sqlstate: str = sqlstate + + def __str__(self) -> str: + return self._message or super().__str__() + + +#: Statement prefix for the contamination fixtures; the lock phrase is +#: appended as a trailing SQL comment. +_CONTAMINATED_SQL: str = ( + "INSERT INTO slices (slice_name, description) " + "VALUES (%(slice_name)s, %(description)s) -- " +) + + +def _contaminated_statement(phrase: str) -> tuple[str, dict[str, str]]: + """SQL + bound parameters that merely CONTAIN a contention phrase. + + The realistic shape: a user naming a chart "deadlock analysis", or a + column literally called ``lock_wait_timeout``. ``str()`` of a + SQLAlchemy wrapper renders both, which is why the classifier must + read the driver diagnostic instead. + """ + # The phrase rides BOTH halves a wrapper renders: a trailing SQL + # comment (stands in for a table or column whose name contains it) + # and a bound parameter (a user-supplied chart name). Inert fixture + # text — nothing here is ever executed. + statement: str = _CONTAMINATED_SQL + phrase + return statement, {"slice_name": f"{phrase} analysis", "description": phrase} @pytest.mark.parametrize( "orig", [ - _Orig(args=(1213, "Deadlock found when trying to get lock")), - _Orig(args=(1205, "Lock wait timeout exceeded")), - _Orig(pgcode="40001"), - _Orig(pgcode="40P01"), - _Orig(pgcode="55P03"), - _Orig(sqlstate="40P01"), + _FakeLockDriverError(args=(1213, "Deadlock found when trying to get lock")), + _FakeLockDriverError(args=(1205, "Lock wait timeout exceeded")), + _FakeLockDriverError(pgcode="40001"), + _FakeLockDriverError(pgcode="40P01"), + _FakeLockDriverError(pgcode="55P03"), + _FakeLockDriverError(sqlstate="40P01"), ], ) -def test_lock_contention_positive(orig: _Orig) -> None: +def test_lock_contention_positive(orig: _FakeLockDriverError) -> None: assert is_lock_contention_error(_op_error(orig)) is True @pytest.mark.parametrize( - "text", + "message", [ "deadlock detected somewhere", "Lock wait timeout exceeded; try restarting transaction", @@ -157,22 +200,97 @@ def test_lock_contention_positive(orig: _Orig) -> None: # these message texts. The integration suite runs on SQLite, so # misclassifying them would route a genuine contention test to # the 500/422 paths instead of the 409. - "(sqlite3.OperationalError) database is locked", - "(sqlite3.OperationalError) database table is locked: table_versions", + "database is locked", + "database table is locked: table_versions", ], ) -def test_lock_contention_text_fallback(text: str) -> None: - err = OperationalError(text, None, Exception()) +def test_lock_contention_driver_message_fallback(message: str) -> None: + """Code-less drivers are classified on THEIR OWN message. + + The message rides the driver error, not the wrapper's statement + argument: that is the only place the classifier is allowed to look. + """ + assert ( + is_lock_contention_error(_op_error(_FakeLockDriverError(message=message))) + is True + ) + + +@pytest.mark.parametrize( + "phrase", ["deadlock", "lock wait timeout", "database is locked"] +) +@pytest.mark.parametrize("wrapper_cls", [DataError, OperationalError]) +def test_lock_phrase_in_sql_or_parameters_is_not_contention( + phrase: str, wrapper_cls: type[StatementError] +) -> None: + """A contention phrase in the SQL or bound parameters is not evidence. + + sc-120050 review (Richard Fogaça): ``str(DBAPIError)`` appends the + statement and its parameters, so classifying on the wrapper turned an + unrelated failure — a chart named "deadlock analysis" — into a 409 + telling the client to retry a request that can never succeed. + """ + statement, params = _contaminated_statement(phrase) + err: StatementError = wrapper_cls( + statement, + params, + _FakeLockDriverError(args=(1064, "You have an error in your SQL syntax")), + ) + assert phrase in str(err).lower(), "fixture must contaminate the rendered text" + assert is_lock_contention_error(err) is False + + +def test_structured_code_beats_a_contaminated_driver_message() -> None: + """A non-contention structured code ENDS classification. + + Even when the driver's own message mentions a deadlock (e.g. a + syntax error in a statement that references a table named + ``deadlock_audit``), a decisive errno that is not a contention code + means this is not a lock race. + """ + orig: _FakeLockDriverError = _FakeLockDriverError( + args=(1064, "syntax error near 'deadlock_audit'"), + message="syntax error near 'deadlock_audit'", + ) + assert is_lock_contention_error(_op_error(orig)) is False + + +def test_non_contention_sqlstate_beats_a_contaminated_driver_message() -> None: + orig: _FakeLockDriverError = _FakeLockDriverError( + pgcode="42P01", message='relation "deadlock_audit" does not exist' + ) + assert is_lock_contention_error(_op_error(orig)) is False + + +def test_wrapper_without_a_driver_error_is_not_contention() -> None: + """No driver diagnostic means nothing classifiable — and the + wrapper's own text is exactly what must not be read.""" + err: OperationalError = OperationalError( + "SELECT * FROM deadlock_audit", None, Exception() + ) err.orig = None - assert is_lock_contention_error(err) is True + assert is_lock_contention_error(err) is False + + +def test_raw_driver_error_is_classified_on_itself() -> None: + """A bare driver exception (no SQLAlchemy wrapper) is its own + diagnostic — the classifier accepts it unchanged.""" + assert ( + is_lock_contention_error(_FakeLockDriverError(args=(1213, "Deadlock found"))) + is True + ) + assert ( + is_lock_contention_error(_FakeLockDriverError(message="database is locked")) + is True + ) @pytest.mark.parametrize( "orig", [ - _Orig(args=(1146, "Table does not exist")), - _Orig(args=()), - _Orig(pgcode="42P01"), + _FakeLockDriverError(args=(1146, "Table does not exist")), + _FakeLockDriverError(args=()), + _FakeLockDriverError(pgcode="42P01"), None, ], ) diff --git a/tests/unit_tests/versioning/test_version_info_locking.py b/tests/unit_tests/versioning/test_version_info_locking.py index 5a083c7ad65a..8e4a9a5ee3c3 100644 --- a/tests/unit_tests/versioning/test_version_info_locking.py +++ b/tests/unit_tests/versioning/test_version_info_locking.py @@ -34,16 +34,19 @@ """ from unittest.mock import MagicMock -from uuid import uuid4 +from uuid import UUID, uuid4 from pytest_mock import MockerFixture -from superset.versioning.api_helpers import current_entity_version_info +from superset.versioning.api_helpers import ( + current_entity_version_info, + EntityVersionInfo, +) from superset.versioning.queries import current_live_transaction_id_locked def _mock_queries_db(mocker: MockerFixture) -> MagicMock: - db = MagicMock() + db: MagicMock = MagicMock() mocker.patch("superset.versioning.queries.db", new=db) mocker.patch("superset.versioning.queries.version_class", new=MagicMock()) return db @@ -59,27 +62,27 @@ def test_locked_read_is_an_exclusive_locking_row_read(mocker: MockerFixture) -> fails here on every backend. Ordered-and-limited so a defensive multi-open-row state cannot raise MultipleResultsFound. """ - db = _mock_queries_db(mocker) + db: MagicMock = _mock_queries_db(mocker) current_live_transaction_id_locked(MagicMock(), 1, uuid4()) - query = db.session.query.return_value - filtered = query.filter.return_value.filter.return_value - limited = filtered.order_by.return_value.limit.return_value + query: MagicMock = db.session.query.return_value + filtered: MagicMock = query.filter.return_value.filter.return_value + limited: MagicMock = filtered.order_by.return_value.limit.return_value limited.with_for_update.assert_called_once_with() limited.with_for_update.return_value.scalar.assert_called_once_with() def test_version_info_threads_the_lock_flag(mocker: MockerFixture) -> None: """lock_for_stale_check=True swaps in the locked transaction id.""" - dao = MagicMock() + dao: MagicMock = MagicMock() dao.current_version_info.return_value = (0, 5) dao.current_live_transaction_id_locked.return_value = 6 mocker.patch("superset.versioning.api_helpers.VersionDAO", new=dao) mocker.patch("superset.versioning.api_helpers._capture_enabled", return_value=True) - entity_uuid = uuid4() + entity_uuid: UUID = uuid4() - info = current_entity_version_info( + info: EntityVersionInfo = current_entity_version_info( MagicMock(), 1, entity_uuid, lock_for_stale_check=True ) @@ -94,12 +97,12 @@ def test_version_info_threads_the_lock_flag(mocker: MockerFixture) -> None: def test_version_info_defaults_to_plain_reads(mocker: MockerFixture) -> None: """Without the flag (every GET path) the locking read never runs.""" - dao = MagicMock() + dao: MagicMock = MagicMock() dao.current_version_info.return_value = (0, 5) mocker.patch("superset.versioning.api_helpers.VersionDAO", new=dao) mocker.patch("superset.versioning.api_helpers._capture_enabled", return_value=True) - info = current_entity_version_info(MagicMock(), 1, uuid4()) + info: EntityVersionInfo = current_entity_version_info(MagicMock(), 1, uuid4()) dao.current_live_transaction_id_locked.assert_not_called() assert info.transaction_id == 5 From 126eed342cf867518f459779df2d3f106698b24a Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Tue, 15 Sep 2026 17:01:58 -0600 Subject: [PATCH 7/8] docs(api): regenerate the published spec for the 409/412 If-Match responses check-openapi-spec-drift went red: the new retryable-409 response and the reworded 412 were added to the docstrings but the published spec was never regenerated, so the two disagreed. Generated in an isolated clone with the CLI's resolved module path asserted to lie inside it first -- the entry point builds its output path from the imported module's __file__, so a bare invocation can write into whichever editable install answers the import. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01EA4MGEkUcDpyLReqjQBhPy --- docs/static/resources/openapi.json | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/docs/static/resources/openapi.json b/docs/static/resources/openapi.json index 63e24defb77f..6016031b2bbc 100644 --- a/docs/static/resources/openapi.json +++ b/docs/static/resources/openapi.json @@ -23977,7 +23977,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": { @@ -24059,6 +24059,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": { @@ -24072,7 +24087,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" From c6767b2a487aa300194778c0fd52db170f47d8b6 Mon Sep 17 00:00:00 2001 From: Mike Bridge Date: Wed, 16 Sep 2026 17:10:35 -0600 Subject: [PATCH 8/8] test(versioning): pin conditional lock error and rollback paths --- superset/datasets/api.py | 18 +-- superset/versioning/api_helpers.py | 10 +- superset/versioning/db_errors.py | 10 +- superset/versioning/queries.py | 21 ++-- .../conditional_token_lock_tests.py | 55 ++++++--- tests/unit_tests/datasets/api_tests.py | 112 +++++++++++++----- tests/unit_tests/versioning/test_db_errors.py | 8 ++ 7 files changed, 164 insertions(+), 70 deletions(-) diff --git a/superset/datasets/api.py b/superset/datasets/api.py index a9afab95b870..4f97b18b3087 100644 --- a/superset/datasets/api.py +++ b/superset/datasets/api.py @@ -113,6 +113,7 @@ concurrency_token_from, current_entity_version_info, entity_concurrency_token, + EntityVersionInfo, get_version_endpoint, list_versions_endpoint, lock_entity_for_update, @@ -576,13 +577,16 @@ def post(self) -> Response: def _lock_contention_response(self) -> Response: """The shared retryable 409 for a conditional save losing a lock race. - The rollback is LOAD-BEARING for lock-wait-timeout: with + 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. + 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( @@ -751,9 +755,9 @@ def put(self, pk: int) -> Response: # noqa: C901 # the live version, the command writes, and the two must not interleave # 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.) + # 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 @@ -767,7 +771,7 @@ def put(self, pk: int) -> Response: # noqa: C901 # the command's find_by_id re-read a stale row on MySQL # REPEATABLE READ (see lock_entity_for_update). try: - _locked_entity = lock_entity_for_update(SqlaTable, pk) + _locked_entity: SqlaTable | None = lock_entity_for_update(SqlaTable, pk) except OperationalError as ex: if not is_lock_contention_error(ex): raise @@ -786,7 +790,7 @@ def put(self, pk: int) -> Response: # noqa: C901 # DatasetUpdateFailedError handler below via ``__cause__`` and # returns the same 409.) try: - old_info = current_entity_version_info( + old_info: EntityVersionInfo = current_entity_version_info( SqlaTable, pk, lock_for_stale_check=conditional ) except OperationalError as ex: diff --git a/superset/versioning/api_helpers.py b/superset/versioning/api_helpers.py index 22a38279d84c..1c1817d672d5 100644 --- a/superset/versioning/api_helpers.py +++ b/superset/versioning/api_helpers.py @@ -131,6 +131,8 @@ 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 ) @@ -245,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 diff --git a/superset/versioning/db_errors.py b/superset/versioning/db_errors.py index c2035f7602ba..c2572f897454 100644 --- a/superset/versioning/db_errors.py +++ b/superset/versioning/db_errors.py @@ -80,7 +80,7 @@ def is_lock_contention_error(exc: BaseException | None) -> bool: if exc is None: return False - orig = getattr(exc, "orig", None) + orig: object = getattr(exc, "orig", None) if orig is None: if isinstance(exc, DBAPIError): # A wrapper whose driver error was not preserved carries no @@ -89,23 +89,23 @@ def is_lock_contention_error(exc: BaseException | None) -> bool: # read. Unclassifiable is not retryable. return False # Not a wrapper: a raw driver exception is its own diagnostic. - diagnostic: BaseException = exc + diagnostic: object = exc else: diagnostic = orig - args = getattr(diagnostic, "args", None) + 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 = getattr(diagnostic, "pgcode", None) or getattr( + 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(diagnostic).lower() + text: str = str(diagnostic).lower() return any(phrase in text for phrase in _LOCK_CONTENTION_PHRASES) diff --git a/superset/versioning/queries.py b/superset/versioning/queries.py index 649e1fbd662f..aab31b4a279c 100644 --- a/superset/versioning/queries.py +++ b/superset/versioning/queries.py @@ -187,10 +187,7 @@ def current_version_info( sa.func.max( sa.case( ( - sa.and_( - ver_cls.end_transaction_id.is_(None), - ver_cls.operation_type != OPERATION_DELETE, - ), + sa.and_(*_live_version_predicates(ver_cls)), ver_cls.transaction_id, ) ) @@ -224,6 +221,17 @@ def current_version_number( return version_number +def _live_version_predicates( + ver_cls: type[Any], +) -> tuple[sa.ColumnElement[bool], sa.ColumnElement[bool]]: + """Share live-row eligibility between display and locked validator reads.""" + # Continuum generates model classes with additional version columns. + return ( + ver_cls.end_transaction_id.is_(None), + ver_cls.operation_type != OPERATION_DELETE, + ) + + def current_live_transaction_id_locked( model_cls: type[Model], entity_id: int, entity_uuid: UUID ) -> int | None: @@ -274,10 +282,7 @@ def current_live_transaction_id_locked( return ( db.session.query(ver_cls.transaction_id) .filter(identity_filter(ver_cls, entity_id, entity_uuid)) - .filter( - ver_cls.end_transaction_id.is_(None), - ver_cls.operation_type != OPERATION_DELETE, - ) + .filter(*_live_version_predicates(ver_cls)) .order_by(ver_cls.transaction_id.desc()) .limit(1) .with_for_update() diff --git a/tests/integration_tests/versioning/conditional_token_lock_tests.py b/tests/integration_tests/versioning/conditional_token_lock_tests.py index 094539185215..b91f67bd2bf2 100644 --- a/tests/integration_tests/versioning/conditional_token_lock_tests.py +++ b/tests/integration_tests/versioning/conditional_token_lock_tests.py @@ -35,9 +35,12 @@ instead of returning it -- a different, also-safe outcome. """ +from uuid import UUID + import pytest import sqlalchemy as sa from sqlalchemy_continuum import version_class, versioning_manager +from sqlalchemy_continuum.operation import Operation from superset import db from superset.daos.version import VersionDAO @@ -46,6 +49,7 @@ from superset.versioning.api_helpers import ( concurrency_token_from, current_entity_version_info, + EntityVersionInfo, ) from superset.versioning.baseline import CONTINUUM_BOOKKEEPING_COLUMNS from superset.versioning.etag import raise_for_stale_write, StaleEntityError @@ -81,7 +85,7 @@ def _commit_competing_version_out_of_band( ).inserted_primary_key[0] content: dict[str, object] if template is None: - live = ( + live: sa.RowMapping | None = ( conn.execute( sa.select(ver_tbl).where( ver_tbl.c.id == chart_id, @@ -112,7 +116,7 @@ def _commit_competing_version_out_of_band( }, transaction_id=new_tx, end_transaction_id=None, - operation_type=1, + operation_type=Operation.UPDATE, ) ) return new_tx @@ -122,15 +126,18 @@ def _commit_competing_version_out_of_band( class TestConditionalTokenLockingRead(SupersetTestCase): def _versioned_chart(self) -> tuple[Slice, str | None]: """A chart with at least one version row (a real committed save).""" - chart = db.session.query(Slice).filter(Slice.slice_name == "Boys").one() - original = chart.description + chart: Slice = db.session.query(Slice).filter(Slice.slice_name == "Boys").one() + # The example fixture stages INSERTs; commit them before the UPDATE + # so the version under test represents a separate committed save. + db.session.commit() + original: str | None = chart.description chart.description = "sc-120050 probe" db.session.commit() return chart, original def _restore(self, chart_id: int, original: str | None) -> None: db.session.rollback() - chart = db.session.get(Slice, chart_id) + chart: Slice | None = db.session.get(Slice, chart_id) assert chart is not None chart.description = original db.session.commit() @@ -139,16 +146,20 @@ def test_locked_read_agrees_with_plain_read_on_quiet_path(self) -> None: """Without concurrent writers both reads name the same live row, and the locking clause executes on the real backend (the dialect-syntax half of the guard the unit pins cannot cover).""" + chart: Slice + original: str | None chart, original = self._versioned_chart() try: - plain = VersionDAO.current_live_transaction_id(Slice, chart.id, chart.uuid) - locked = VersionDAO.current_live_transaction_id_locked( + plain: int | None = VersionDAO.current_live_transaction_id( + Slice, chart.id, chart.uuid + ) + locked: int | None = VersionDAO.current_live_transaction_id_locked( Slice, chart.id, chart.uuid ) assert plain is not None assert locked == plain - info = current_entity_version_info( + info: EntityVersionInfo = current_entity_version_info( Slice, chart.id, chart.uuid, lock_for_stale_check=True ) assert info.transaction_id == plain @@ -165,6 +176,8 @@ def test_no_live_row_resolves_to_none_under_lock(self) -> None: plain path does. This is also the branch where MySQL takes a gap lock (see current_live_transaction_id_locked's residual note). """ + chart: Slice + original: str | None chart, original = self._versioned_chart() try: db.session.execute( @@ -173,12 +186,12 @@ def test_no_live_row_resolves_to_none_under_lock(self) -> None: ) db.session.commit() - locked = VersionDAO.current_live_transaction_id_locked( + locked: int | None = VersionDAO.current_live_transaction_id_locked( Slice, chart.id, chart.uuid ) assert locked is None - info = current_entity_version_info( + info: EntityVersionInfo = current_entity_version_info( Slice, chart.id, chart.uuid, lock_for_stale_check=True ) assert info.version_uuid is None @@ -221,9 +234,11 @@ def test_locking_read_sees_the_interleaved_commit_the_snapshot_misses( # v1 is committed BEFORE the read view opens: the fixture's rows # and this save must both be visible to the snapshot the test # then pins. + chart: Slice + original: str | None chart, original = self._versioned_chart() chart_id: int = chart.id - chart_uuid = chart.uuid + chart_uuid: UUID = chart.uuid try: self._force_repeatable_read() # Opening the read view: this first consistent read is what @@ -232,7 +247,9 @@ def test_locking_read_sees_the_interleaved_commit_the_snapshot_misses( Slice, chart_id, chart_uuid ) assert v1 is not None - stale_info = current_entity_version_info(Slice, chart_id, chart_uuid) + stale_info: EntityVersionInfo = current_entity_version_info( + Slice, chart_id, chart_uuid + ) stale_token: str | None = concurrency_token_from(stale_info) assert stale_token is not None @@ -252,7 +269,7 @@ def test_locking_read_sees_the_interleaved_commit_the_snapshot_misses( == v2 ) - fresh_info = current_entity_version_info( + fresh_info: EntityVersionInfo = current_entity_version_info( Slice, chart_id, chart_uuid, lock_for_stale_check=True ) fresh_token: str | None = concurrency_token_from(fresh_info) @@ -281,12 +298,14 @@ def test_first_version_transition_under_forced_repeatable_read(self) -> None: being accepted. """ self._skip_unless_mysql() + chart: Slice + original: str | None chart, original = self._versioned_chart() chart_id: int = chart.id - chart_uuid = chart.uuid + chart_uuid: UUID = chart.uuid ver_tbl: sa.Table = version_class(Slice).__table__ try: - template = ( + template: sa.RowMapping | None = ( db.session.execute( sa.select(ver_tbl).where( ver_tbl.c.id == chart_id, @@ -308,7 +327,9 @@ def test_first_version_transition_under_forced_repeatable_read(self) -> None: VersionDAO.current_live_transaction_id(Slice, chart_id, chart_uuid) is None ) - unversioned_info = current_entity_version_info(Slice, chart_id, chart_uuid) + unversioned_info: EntityVersionInfo = current_entity_version_info( + Slice, chart_id, chart_uuid + ) unversioned_token: str | None = concurrency_token_from(unversioned_info) assert unversioned_token is not None @@ -327,7 +348,7 @@ def test_first_version_transition_under_forced_repeatable_read(self) -> None: == first_tx ) - fresh_info = current_entity_version_info( + fresh_info: EntityVersionInfo = current_entity_version_info( Slice, chart_id, chart_uuid, lock_for_stale_check=True ) fresh_token: str | None = concurrency_token_from(fresh_info) diff --git a/tests/unit_tests/datasets/api_tests.py b/tests/unit_tests/datasets/api_tests.py index a7be5245af6a..da199a4ca5f1 100644 --- a/tests/unit_tests/datasets/api_tests.py +++ b/tests/unit_tests/datasets/api_tests.py @@ -15,13 +15,22 @@ # specific language governing permissions and limitations # under the License. -from typing import Any +from typing import Any, TYPE_CHECKING from unittest.mock import MagicMock, patch +import pytest +from flask.testing import FlaskClient +from sqlalchemy.exc import DataError, OperationalError from sqlalchemy.orm.session import Session +from werkzeug.test import TestResponse from superset import db +from superset.commands.dataset.exceptions import DatasetUpdateFailedError from superset.utils import json +from superset.versioning.api_helpers import EntityVersionInfo + +if TYPE_CHECKING: + from superset.connectors.sqla.models import SqlaTable def test_put_invalid_dataset( @@ -487,7 +496,7 @@ def test_put_dataset_rejects_stale_if_match( def test_put_dataset_maps_lock_contention_to_retryable_409( session: Session, - client: Any, + client: FlaskClient, full_api_access: None, ) -> None: """sc-120050: a deadlock at the locked validator read is a 409, not 412. @@ -497,34 +506,36 @@ def test_put_dataset_maps_lock_contention_to_retryable_409( guidance would be wrong, so the mapping must not drift back to it. The message tells the client to retry the same request. """ - from sqlalchemy.exc import OperationalError - - dataset = _create_dataset("test_put_lock_contention") - deadlock = OperationalError("stmt", None, Exception()) + dataset: SqlaTable = _create_dataset("test_put_lock_contention") + deadlock: OperationalError = OperationalError("stmt", None, Exception()) deadlock.orig = type("Orig", (), {"args": (1213, "Deadlock found")})() with patch( "superset.datasets.api.current_entity_version_info", side_effect=deadlock, ): - response = client.put( + response: TestResponse = client.put( f"/api/v1/dataset/{dataset.id}", json={"description": "from a racing tab"}, headers={"If-Match": '"anything"'}, ) assert response.status_code == 409 - body = response.get_data(as_text=True) + body: str = response.get_data(as_text=True) assert "Retry the same request" in body # No write-check re-read here: the handler's rollback detaches the # fixture object, and nothing could have written anyway — the patched # read raised before the update command ever ran. +@pytest.mark.parametrize( + "lock_point", ["lock_entity_for_update", "current_entity_version_info"] +) def test_put_dataset_maps_entity_lock_contention_to_retryable_409( session: Session, - client: Any, + client: FlaskClient, full_api_access: None, + lock_point: str, ) -> None: """A lock race lost at the ENTITY-LOCK acquisition is also a 409. @@ -534,29 +545,78 @@ def test_put_dataset_maps_entity_lock_contention_to_retryable_409( try/except. Left unmapped this surfaced as an uncaught 500, defeating the retryable-conflict contract the other two lock points advertise. """ - from sqlalchemy.exc import OperationalError - - dataset = _create_dataset("test_put_entity_lock_contention") - timeout = OperationalError("stmt", None, Exception()) + dataset: SqlaTable = _create_dataset("test_put_entity_lock_contention") + timeout: OperationalError = OperationalError("stmt", None, Exception()) timeout.orig = type("Orig", (), {"args": (1205, "Lock wait timeout exceeded")})() - with patch( - "superset.datasets.api.lock_entity_for_update", - side_effect=timeout, + with ( + patch(f"superset.datasets.api.{lock_point}", side_effect=timeout), + patch.object(db.session, "rollback", wraps=db.session.rollback) as rollback, ): - response = client.put( + response: TestResponse = client.put( f"/api/v1/dataset/{dataset.id}", json={"description": "from a racing tab"}, headers={"If-Match": '"anything"'}, ) + rollback.assert_called_once() assert response.status_code == 409 assert "Retry the same request" in response.get_data(as_text=True) +@pytest.mark.parametrize("conditional", [True, False]) +def test_put_dataset_maps_chained_deadlock_to_retryable_409( + session: Session, + client: FlaskClient, + full_api_access: None, + conditional: bool, +) -> None: + """Translate a chained deadlock only for a conditional request.""" + failure: DatasetUpdateFailedError = DatasetUpdateFailedError() + failure.__cause__ = OperationalError( + "INSERT", None, Exception(1213, "Deadlock found") + ) + dataset_id: int = _create_dataset("test_put_chained_deadlock").id + with ( + patch("superset.datasets.api.raise_for_stale_write"), + patch("superset.datasets.api.UpdateDatasetCommand.run", side_effect=failure), + ): + response: TestResponse = client.put( + f"/api/v1/dataset/{dataset_id}", + json={"description": "from a racing tab"}, + headers={"If-Match": '"anything"'} if conditional else {}, + ) + assert response.status_code == (409 if conditional else 422) + assert ("Retry the same request" in response.get_data(as_text=True)) is conditional + + +@pytest.mark.parametrize( + "lock_point", ["lock_entity_for_update", "current_entity_version_info"] +) +def test_put_dataset_does_not_reclassify_unrelated_lock_read_errors( + session: Session, + client: FlaskClient, + full_api_access: None, + lock_point: str, +) -> None: + """Preserve unrelated driver errors at either direct locking read.""" + failure: OperationalError = OperationalError( + "SELECT", None, Exception(2006, "MySQL server has gone away") + ) + dataset_id: int = _create_dataset("test_put_unrelated_lock_error").id + with patch(f"superset.datasets.api.{lock_point}", side_effect=failure): + response: TestResponse = client.put( + f"/api/v1/dataset/{dataset_id}", + json={"description": "not a contention error"}, + headers={"If-Match": '"anything"'}, + ) + assert response.status_code == 500 + assert "Retry the same request" not in response.get_data(as_text=True) + + def test_put_dataset_without_if_match_never_locks( session: Session, - client: Any, + client: FlaskClient, full_api_access: None, ) -> None: """An unconditional PUT takes no locks and no locking validator read. @@ -564,9 +624,7 @@ def test_put_dataset_without_if_match_never_locks( Pins the perf contract both ways: hard-coding lock_for_stale_check=True at the call site (or unconditionally locking the entity) fails here. """ - from superset.versioning.api_helpers import EntityVersionInfo - - dataset = _create_dataset("test_put_unconditional_no_locks") + dataset: SqlaTable = _create_dataset("test_put_unconditional_no_locks") with ( patch("superset.datasets.api.lock_entity_for_update") as lock, @@ -577,7 +635,7 @@ def test_put_dataset_without_if_match_never_locks( patch("superset.datasets.api.UpdateDatasetCommand") as cmd, ): cmd.return_value.run.return_value = dataset - response = client.put( + response: TestResponse = client.put( f"/api/v1/dataset/{dataset.id}", json={"description": "plain save"}, ) @@ -592,7 +650,7 @@ def test_put_dataset_without_if_match_never_locks( def test_put_dataset_keeps_422_when_sql_text_merely_mentions_a_lock( session: Session, - client: Any, + client: FlaskClient, full_api_access: None, ) -> None: """Response boundary: a contaminated non-lock failure stays a 422. @@ -605,11 +663,7 @@ def test_put_dataset_keeps_422_when_sql_text_merely_mentions_a_lock( the endpoint classifies its ``__cause__``, so the regression belongs at this boundary as well as in the classifier's own unit tests. """ - from sqlalchemy.exc import DataError - - from superset.commands.dataset.exceptions import DatasetUpdateFailedError - - dataset = _create_dataset("test_put_contaminated_data_error") + dataset: SqlaTable = _create_dataset("test_put_contaminated_data_error") driver_error: Exception = type( "Orig", (Exception,), {"args": (1064, "You have an error in your SQL syntax")} )() @@ -629,7 +683,7 @@ def test_put_dataset_keeps_422_when_sql_text_merely_mentions_a_lock( patch("superset.datasets.api.raise_for_stale_write"), patch("superset.datasets.api.UpdateDatasetCommand.run", side_effect=failure), ): - response = client.put( + response: TestResponse = client.put( f"/api/v1/dataset/{dataset.id}", json={"description": "deadlock analysis"}, headers={"If-Match": '"anything"'}, diff --git a/tests/unit_tests/versioning/test_db_errors.py b/tests/unit_tests/versioning/test_db_errors.py index 963a5f84000b..a19cfece3f39 100644 --- a/tests/unit_tests/versioning/test_db_errors.py +++ b/tests/unit_tests/versioning/test_db_errors.py @@ -181,6 +181,12 @@ def _contaminated_statement(phrase: str) -> tuple[str, dict[str, str]]: [ _FakeLockDriverError(args=(1213, "Deadlock found when trying to get lock")), _FakeLockDriverError(args=(1205, "Lock wait timeout exceeded")), + _FakeLockDriverError( + args=(1213, "Deadlock found when trying to get lock"), sqlstate="40001" + ), + _FakeLockDriverError( + args=(1205, "Lock wait timeout exceeded"), sqlstate="HY000" + ), _FakeLockDriverError(pgcode="40001"), _FakeLockDriverError(pgcode="40P01"), _FakeLockDriverError(pgcode="55P03"), @@ -230,6 +236,8 @@ def test_lock_phrase_in_sql_or_parameters_is_not_contention( unrelated failure — a chart named "deadlock analysis" — into a 409 telling the client to retry a request that can never succeed. """ + statement: str + params: dict[str, str] statement, params = _contaminated_statement(phrase) err: StatementError = wrapper_cls( statement,