From 0fe9a225b1c084102758d307c1e88d93d2c8211f Mon Sep 17 00:00:00 2001 From: Fabian Fulga Date: Fri, 23 Oct 2026 12:18:12 +0300 Subject: [PATCH 1/2] Compare-and-update progress steps so a stale write cannot rewind the bar. Task and minion-pool progress updates now advance current_step only when the stored step is not already ahead, including a completed update that a late cast would otherwise overwrite. --- coriolis/db/api.py | 107 ++++++++++---- coriolis/tests/db/test_api.py | 256 ++++++++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+), 27 deletions(-) diff --git a/coriolis/db/api.py b/coriolis/db/api.py index 81d4c2ea..f9485d3b 100644 --- a/coriolis/db/api.py +++ b/coriolis/db/api.py @@ -1004,6 +1004,63 @@ def _get_minion_pool_progress_update(context, pool_id, index): ).first() +def _compare_and_update_progress( + context, + progress_update, + parent_id, + update_index, + new_current_step, + new_total_steps=None, + new_message=None, + resource_name="task", +): + """Advance a progress row only when ``new_current_step`` is not behind. + + The step check is part of the UPDATE predicate (``current_step <= + new_current_step``). Comparing the value loaded on ``progress_update`` + is not enough: that value can be stale, and a later flush would overwrite + a newer step stored by a concurrent writer. + + Returns True when this update was stored. A rejected update leaves the + row unchanged. ``progress_update`` is refreshed from the database either + way. + """ + values = { + "current_step": new_current_step, + "updated_at": timeutils.utcnow(), + } + if new_total_steps is not None: + values["total_steps"] = new_total_steps + if new_message is not None: + values["message"] = new_message + + model = progress_update.__class__ + session = _session(context) + # synchronize_session=False keeps a stale identity-map object from being + # rewritten in Python and flushed back over the conditional UPDATE. + updated = ( + session.query(model) + .filter( + model.id == progress_update.id, + model.current_step <= new_current_step, + ) + .update(values, synchronize_session=False) + ) + session.refresh(progress_update) + if updated == 0: + LOG.debug( + "Ignoring out-of-order progress update for %s '%s' " + "index %s: new_current_step=%s is behind stored current_step=%s", + resource_name, + parent_id, + update_index, + new_current_step, + progress_update.current_step, + ) + return False + return True + + @enginefacade.writer def add_minion_pool_progress_update( context, pool_id, message, initial_step=0, total_steps=0 @@ -1041,11 +1098,16 @@ def update_minion_pool_progress_update( "index %s in the DB for updating." % (pool_id, update_index) ) - pool_progress_update.current_step = new_current_step - if new_total_steps is not None: - pool_progress_update.total_steps = new_total_steps - if new_message is not None: - pool_progress_update.message = new_message + _compare_and_update_progress( + context, + pool_progress_update, + pool_id, + update_index, + new_current_step, + new_total_steps=new_total_steps, + new_message=new_message, + resource_name="minion pool", + ) return pool_progress_update @@ -1100,27 +1162,9 @@ def update_task_progress_update( "index %s in the DB for updating." % (task_id, update_index) ) - # Quick progress updates may be processed out of order. - # We're trying to mitigate this by checking the current step. - # - # TODO(ffulga): the current approach is still prone to race conditions - # since `task_progress_update.current_step` may be out of date. - # The safest approach would be to use a db-level `compare-and-update` - # sql operation. - if new_current_step < task_progress_update.current_step: - LOG.debug( - "Ignoring out-of-order progress update for task '%s' " - "index %s: new_current_step=%s is behind stored current_step=%s", - task_id, - update_index, - new_current_step, - task_progress_update.current_step, - ) - return - - task_progress_update.current_step = new_current_step - if new_total_steps is not None: - task_progress_update.total_steps = new_total_steps + # Quick progress updates may be processed out of order. Reject a step + # that would move the bar backwards with a compare-and-update so a stale + # read of current_step cannot overwrite a newer value. if new_message is not None: max_msg_len = models.MAX_EVENT_MESSAGE_LENGHT if len(new_message) > max_msg_len: @@ -1131,7 +1175,16 @@ def update_task_progress_update( f" Original message was: '{new_message}'" ) new_message = f"{new_message[: max_msg_len - len('...')]}..." # noqa - task_progress_update.message = new_message + + _compare_and_update_progress( + context, + task_progress_update, + task_id, + update_index, + new_current_step, + new_total_steps=new_total_steps, + new_message=new_message, + ) @enginefacade.writer diff --git a/coriolis/tests/db/test_api.py b/coriolis/tests/db/test_api.py index 46e7bd98..b19b781c 100644 --- a/coriolis/tests/db/test_api.py +++ b/coriolis/tests/db/test_api.py @@ -1384,3 +1384,259 @@ def test_get_action_include_task_info(self): self.context, self.user_transfer.id, include_task_info=True ) self.assertEqual(result.info, self.user_transfer.info) + + +class ProgressUpdateDBAPITestCase(BaseDBAPITestCase): + """Progress updates must not move current_step backwards.""" + + def _task_id(self): + return self.valid_data['user_scope']['task'].id + + def _reload(self, model, row_id): + self.session.expire_all() + return self.session.query(model).filter_by(id=row_id).one() + + def _add_task_progress(self, initial_step=0, total_steps=10, message="start"): + row = api.add_task_progress_update( + self.context, + self._task_id(), + message, + initial_step=initial_step, + total_steps=total_steps, + ) + self.session.flush() + return row + + def _add_pool_progress(self, initial_step=0, total_steps=10, message="start"): + endpoint = self.valid_data['user_scope']['source_endpoint'] + pool = models.MinionPool() + pool.id = str(uuid.uuid4()) + pool.user_id = DEFAULT_USER_ID + pool.project_id = DEFAULT_PROJECT_ID + pool.name = "progress-pool-%s" % pool.id + pool.endpoint_id = endpoint.id + pool.os_type = "linux" + pool.platform = "openstack" + pool.minimum_minions = 1 + pool.maximum_minions = 1 + pool.minion_max_idle_time = 60 + pool.minion_retention_strategy = ( + constants.MINION_POOL_MACHINE_RETENTION_STRATEGY_DELETE + ) + pool.status = constants.MINION_POOL_STATUS_ALLOCATED + self.session.add(pool) + self.session.flush() + row = api.add_minion_pool_progress_update( + self.context, + pool.id, + message, + initial_step=initial_step, + total_steps=total_steps, + ) + self.session.flush() + return pool, row + + def _set_step_without_refreshing_session(self, model, row, current_step, message): + """Write a newer step the way a concurrent transaction would. + + ``synchronize_session=False`` leaves the already-loaded ORM object + on its previous ``current_step``, which is the stale read the + compare-and-update has to tolerate. + """ + self.session.query(model).filter_by(id=row.id).update( + {"current_step": current_step, "message": message}, + synchronize_session=False, + ) + + def test_update_task_progress_update_advances_step(self): + row = self._add_task_progress(initial_step=2, message="copying") + + api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 8, + new_total_steps=20, + new_message="almost done", + ) + + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.current_step, 8) + self.assertEqual(stored.total_steps, 20) + self.assertEqual(stored.message, "almost done") + + def test_update_task_progress_update_same_step_updates_message(self): + row = self._add_task_progress(initial_step=4, total_steps=10, message="copying") + + api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 4, + new_total_steps=12, + new_message="still copying", + ) + + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.current_step, 4) + self.assertEqual(stored.total_steps, 12) + self.assertEqual(stored.message, "still copying") + + def test_update_task_progress_update_ignores_older_step(self): + row = self._add_task_progress(initial_step=4, total_steps=10, message="copying") + + result = api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 3, + new_total_steps=99, + new_message="stale", + ) + + self.assertIsNone(result) + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.current_step, 4) + self.assertEqual(stored.total_steps, 10) + self.assertEqual(stored.message, "copying") + + def test_update_task_progress_update_stale_read_cannot_rewind(self): + row = self._add_task_progress(initial_step=2, total_steps=10, message="copying") + self._set_step_without_refreshing_session( + models.TaskProgressUpdate, row, 9, "later" + ) + self.assertEqual(row.current_step, 2) + + api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 4, + new_total_steps=99, + new_message="stale", + ) + + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.current_step, 9) + self.assertEqual(stored.total_steps, 10) + self.assertEqual(stored.message, "later") + + def test_update_task_progress_update_stale_read_still_advances(self): + row = self._add_task_progress(initial_step=2, total_steps=10, message="copying") + self._set_step_without_refreshing_session( + models.TaskProgressUpdate, row, 5, "midway" + ) + self.assertEqual(row.current_step, 2) + + api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 8, + new_message="almost done", + ) + + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.current_step, 8) + self.assertEqual(stored.message, "almost done") + + def test_update_task_progress_update_not_found(self): + self.assertRaises( + exception.NotFound, + api.update_task_progress_update, + self.context, + self._task_id(), + 99999, + 1, + ) + + def test_update_task_progress_update_truncates_message(self): + row = self._add_task_progress(initial_step=1, message="start") + message = "x" * (models.MAX_EVENT_MESSAGE_LENGHT + 50) + expected = "x" * (models.MAX_EVENT_MESSAGE_LENGHT - len("...")) + "..." + + api.update_task_progress_update( + self.context, + self._task_id(), + row.index, + 2, + new_message=message, + ) + + stored = self._reload(models.TaskProgressUpdate, row.id) + self.assertEqual(stored.message, expected) + self.assertEqual(stored.current_step, 2) + + def test_update_minion_pool_progress_update_advances_step(self): + pool, row = self._add_pool_progress(initial_step=2, message="allocating") + + result = api.update_minion_pool_progress_update( + self.context, + pool.id, + row.index, + 6, + new_total_steps=8, + new_message="booting", + ) + + self.assertEqual(result.current_step, 6) + self.assertEqual(result.message, "booting") + stored = self._reload(models.MinionPoolProgressUpdate, row.id) + self.assertEqual(stored.current_step, 6) + self.assertEqual(stored.total_steps, 8) + self.assertEqual(stored.message, "booting") + + def test_update_minion_pool_progress_update_ignores_older_step(self): + pool, row = self._add_pool_progress( + initial_step=6, total_steps=8, message="booting" + ) + + result = api.update_minion_pool_progress_update( + self.context, + pool.id, + row.index, + 3, + new_total_steps=99, + new_message="stale", + ) + + self.assertEqual(result.current_step, 6) + self.assertEqual(result.message, "booting") + stored = self._reload(models.MinionPoolProgressUpdate, row.id) + self.assertEqual(stored.current_step, 6) + self.assertEqual(stored.total_steps, 8) + self.assertEqual(stored.message, "booting") + + def test_update_minion_pool_progress_update_stale_read_cannot_rewind(self): + pool, row = self._add_pool_progress( + initial_step=2, total_steps=8, message="allocating" + ) + self._set_step_without_refreshing_session( + models.MinionPoolProgressUpdate, row, 7, "later" + ) + self.assertEqual(row.current_step, 2) + + result = api.update_minion_pool_progress_update( + self.context, + pool.id, + row.index, + 4, + new_message="stale", + ) + + self.assertEqual(result.current_step, 7) + self.assertEqual(result.message, "later") + stored = self._reload(models.MinionPoolProgressUpdate, row.id) + self.assertEqual(stored.current_step, 7) + self.assertEqual(stored.message, "later") + + def test_update_minion_pool_progress_update_not_found(self): + pool, _row = self._add_pool_progress() + self.assertRaises( + exception.NotFound, + api.update_minion_pool_progress_update, + self.context, + pool.id, + 99999, + 1, + ) From c74aa953b8d80da09f46ca4a23bb483b4fd10f56 Mon Sep 17 00:00:00 2001 From: Fabian Fulga Date: Fri, 23 Oct 2026 16:26:24 +0300 Subject: [PATCH 2/2] Add integration test for progress updates --- .../integration/test_progress_updates.py | 360 ++++++++++++++++++ 1 file changed, 360 insertions(+) create mode 100644 coriolis/tests/integration/test_progress_updates.py diff --git a/coriolis/tests/integration/test_progress_updates.py b/coriolis/tests/integration/test_progress_updates.py new file mode 100644 index 00000000..5179bece --- /dev/null +++ b/coriolis/tests/integration/test_progress_updates.py @@ -0,0 +1,360 @@ +# Copyright 2026 Cloudbase Solutions Srl +# All Rights Reserved. + +"""Progress updates must not move a bar backwards. + +The conductor applies task progress updates, and the minion manager applies +pool progress updates. Intermediate ticks are asynchronous, so one can be +handled after a later tick has already been stored. These tests open that +window on purpose: the service reads the row, another connection commits a +different step, then the service writes. A completed step stays stored, and +the completed write itself still lands. +""" + +import threading +import uuid + +import sqlalchemy +from oslo_config import cfg +from oslo_db.sqlalchemy import enginefacade + +from coriolis import constants, context, exception +from coriolis.conductor.rpc import client as conductor_client +from coriolis.db import api as db_api +from coriolis.db.sqlalchemy import models +from coriolis.minion_manager.rpc import client as minion_manager_client +from coriolis.tests.integration import base + +CONF = cfg.CONF + +_INITIAL_STEP = 40 +_TOTAL_STEPS = 100 +_RPC_TIMEOUT = 30 +_WINDOW_TIMEOUT = 15 + + +class ProgressUpdateIntegrationTest(base.CoriolisIntegrationTestBase): + """Out-of-order progress updates against the running services and MariaDB.""" + + def setUp(self): + super().setUp() + self._ctx = context.RequestContext( + user="progress-test", + project_id=self._harness.keystone.project_id, + is_admin=True, + ) + self._admin = context.get_admin_context() + self._engine = self._writer_engine() + + def test_task_late_update_cannot_rewind_completed_step(self): + task_id, index, progress_id = self._create_task_progress() + + self._send_while_peer_commits( + "task_progress_update", + progress_id, + peer_step=_TOTAL_STEPS, + peer_message="done", + send_update=lambda: self._update_task(task_id, index, 80, "late"), + ) + + stored = self._read_progress("task_progress_update", progress_id) + self.assertEqual(_TOTAL_STEPS, stored["current_step"]) + self.assertEqual(_TOTAL_STEPS, stored["total_steps"]) + self.assertEqual("done", stored["message"]) + + def test_task_completed_step_lands_over_concurrent_lower_step(self): + task_id, index, progress_id = self._create_task_progress() + + self._send_while_peer_commits( + "task_progress_update", + progress_id, + peer_step=80, + peer_message="late", + send_update=lambda: self._update_task(task_id, index, _TOTAL_STEPS, "done"), + ) + + stored = self._read_progress("task_progress_update", progress_id) + self.assertEqual(_TOTAL_STEPS, stored["current_step"]) + self.assertEqual(_TOTAL_STEPS, stored["total_steps"]) + self.assertEqual("done", stored["message"]) + + def test_task_completed_step_survives_a_later_update(self): + task_id, index, progress_id = self._create_task_progress() + + self._update_task(task_id, index, _TOTAL_STEPS, "done") + self._update_task(task_id, index, 90, "late") + + stored = self._read_progress("task_progress_update", progress_id) + self.assertEqual(_TOTAL_STEPS, stored["current_step"]) + self.assertEqual("done", stored["message"]) + + def test_minion_pool_late_update_cannot_rewind_completed_step(self): + pool_id, index, progress_id = self._create_pool_progress() + + self._send_while_peer_commits( + "minion_pool_progress_update", + progress_id, + peer_step=_TOTAL_STEPS, + peer_message="done", + send_update=lambda: self._update_pool(pool_id, index, 80, "late"), + ) + + stored = self._read_progress("minion_pool_progress_update", progress_id) + self.assertEqual(_TOTAL_STEPS, stored["current_step"]) + self.assertEqual(_TOTAL_STEPS, stored["total_steps"]) + self.assertEqual("done", stored["message"]) + + def test_minion_pool_completed_step_lands_over_concurrent_lower_step(self): + pool_id, index, progress_id = self._create_pool_progress() + + self._send_while_peer_commits( + "minion_pool_progress_update", + progress_id, + peer_step=80, + peer_message="late", + send_update=lambda: self._update_pool(pool_id, index, _TOTAL_STEPS, "done"), + ) + + stored = self._read_progress("minion_pool_progress_update", progress_id) + self.assertEqual(_TOTAL_STEPS, stored["current_step"]) + self.assertEqual(_TOTAL_STEPS, stored["total_steps"]) + self.assertEqual("done", stored["message"]) + + def _send_while_peer_commits( + self, table, progress_id, peer_step, peer_message, send_update + ): + """Run *send_update* and commit *peer_step* after its read. + + The listener fires once the service's SELECT has returned. The peer + commit happens before that service reaches its UPDATE, which is the + window a late cast has against a completed step. + """ + read_row = threading.Event() + release_write = threading.Event() + state = {"saw_select": False, "peer_first": False, "error": None} + + def _after_cursor_execute( + conn, cursor, statement, parameters, context, executemany + ): + if state["saw_select"]: + return + if not isinstance(statement, str): + statement = statement.decode("utf-8", "replace") + normalized = " ".join(statement.split()).lower() + if not normalized.startswith("select") or table not in normalized: + return + state["saw_select"] = True + read_row.set() + # False means the service continued without the competing commit. + state["peer_first"] = release_write.wait(_WINDOW_TIMEOUT) + + def _peer_commit(): + peer_engine = None + try: + if not read_row.wait(_WINDOW_TIMEOUT): + return + peer_engine = sqlalchemy.create_engine(CONF.database.connection) + with peer_engine.begin() as conn: + result = conn.execute( + sqlalchemy.text( + "UPDATE %s " + "SET current_step = :step, message = :message " + "WHERE id = :progress_id" % table + ), + { + "step": peer_step, + "message": peer_message, + "progress_id": progress_id, + }, + ) + matched = result.rowcount + if matched != 1: + raise RuntimeError( + "peer update of %s matched %s rows" % (table, matched) + ) + except Exception as exc: + state["error"] = exc + finally: + if peer_engine is not None: + peer_engine.dispose() + release_write.set() + + sqlalchemy.event.listen( + self._engine, "after_cursor_execute", _after_cursor_execute + ) + peer = threading.Thread(target=_peer_commit) + peer.start() + try: + send_update() + finally: + release_write.set() + peer.join(_WINDOW_TIMEOUT) + sqlalchemy.event.remove( + self._engine, "after_cursor_execute", _after_cursor_execute + ) + + self.assertFalse(peer.is_alive(), "peer progress update did not finish") + if state["error"] is not None: + raise state["error"] + self.assertTrue( + state["saw_select"], + "service updated %s without reading the row first" % table, + ) + self.assertTrue( + state["peer_first"], + "service wrote %s before the competing step committed" % table, + ) + + def _update_task(self, task_id, index, step, message): + conductor_client.ConductorClient( + timeout=_RPC_TIMEOUT + ).update_task_progress_update( + self._ctx, + task_id, + index, + step, + new_message=message, + sync=True, + ) + + def _update_pool(self, pool_id, index, step, message): + # The public client always casts, so the test would not see the commit. + # _call reaches the same minion-manager method and waits for it. + minion_manager_client.MinionManagerClient(timeout=_RPC_TIMEOUT)._call( + self._ctx, + "update_minion_pool_progress_update", + minion_pool_id=pool_id, + progress_update_index=index, + new_current_step=step, + new_message=message, + ) + + def _create_task_progress(self): + endpoint = self._create_endpoint() + transfer = models.Transfer( + id=str(uuid.uuid4()), + origin_endpoint_id=endpoint.id, + destination_endpoint_id=endpoint.id, + instances=["progress-vm"], + info={"progress-vm": {"volumes_info": []}}, + scenario=constants.TRANSFER_SCENARIO_REPLICA, + notes="progress race", + ) + db_api.add_transfer(self._ctx, transfer) + self.addCleanup( + self._ignoreExc(db_api.delete_transfer, exception.NotFound), + self._admin, + transfer.id, + ) + + execution = models.TasksExecution( + id=str(uuid.uuid4()), + action_id=transfer.id, + status=constants.EXECUTION_STATUS_RUNNING, + type=constants.EXECUTION_TYPE_TRANSFER_EXECUTION, + ) + execution.action = transfer + db_api.add_transfer_tasks_execution(self._ctx, execution) + + task = models.Task( + id=str(uuid.uuid4()), + execution_id=execution.id, + instance="progress-vm", + status=constants.TASK_STATUS_RUNNING, + task_type=constants.TASK_TYPE_REPLICATE_DISKS, + index=1, + on_error=False, + ) + self._persist(task) + db_api.add_task_progress_update( + self._ctx, + task.id, + "copying", + initial_step=_INITIAL_STEP, + total_steps=_TOTAL_STEPS, + ) + stored = self._read_progress_for_parent( + "task_progress_update", "task_id", task.id + ) + return task.id, stored["progress_index"], stored["id"] + + def _create_pool_progress(self): + endpoint = self._create_endpoint() + pool = models.MinionPool( + id=str(uuid.uuid4()), + user_id="progress-test", + project_id=self._harness.keystone.project_id, + name="progress-pool-%s" % uuid.uuid4(), + endpoint_id=endpoint.id, + os_type="linux", + platform="test", + minimum_minions=1, + maximum_minions=1, + minion_max_idle_time=60, + minion_retention_strategy=( + constants.MINION_POOL_MACHINE_RETENTION_STRATEGY_DELETE + ), + status=constants.MINION_POOL_STATUS_ALLOCATED, + ) + self._persist(pool) + db_api.add_minion_pool_progress_update( + self._ctx, + pool.id, + "copying", + initial_step=_INITIAL_STEP, + total_steps=_TOTAL_STEPS, + ) + stored = self._read_progress_for_parent( + "minion_pool_progress_update", "pool_id", pool.id + ) + return pool.id, stored["progress_index"], stored["id"] + + def _create_endpoint(self): + endpoint = models.Endpoint( + id=str(uuid.uuid4()), + name="progress-endpoint-%s" % uuid.uuid4(), + type="test", + connection_info={"k": "v"}, + description="", + ) + db_api.add_endpoint(self._ctx, endpoint) + self.addCleanup( + self._ignoreExc(db_api.delete_endpoint, exception.NotFound), + self._admin, + endpoint.id, + ) + return endpoint + + def _persist(self, obj): + def _add(db_context): + db_api._session(db_context).add(obj) + + enginefacade.writer(_add)(self._ctx) + + def _writer_engine(self): + with enginefacade.writer.using(self._ctx): + bind = self._ctx.session.get_bind() + return getattr(bind, "engine", bind) + + def _read_progress(self, table, progress_id): + return self._select_progress( + table, "id = :progress_id", {"progress_id": progress_id} + ) + + def _read_progress_for_parent(self, table, parent_column, parent_id): + return self._select_progress( + table, + "%s = :parent_id AND deleted = 0" % parent_column, + {"parent_id": parent_id}, + ) + + def _select_progress(self, table, where, params): + with self._engine.connect() as conn: + row = conn.execute( + sqlalchemy.text( + "SELECT id, `index` AS progress_index, current_step, " + "total_steps, message FROM %s WHERE %s" % (table, where) + ), + params, + ).one() + return row._mapping