Compare-and-update progress steps so a stale write cannot rewind the bar

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

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,
+        )