Fix/graceful shutdown (port of #1303 to main)

simplyblock_core/cluster_ops.py

@@ -2807,16 +2807,74 @@ def cluster_grace_startup(cl_id, clear_data=False, spdk_image=None) -> None:
 
 
 
+def _grace_shutdown_skipped(node) -> bool:
+    """Nodes a full-cluster shutdown must not touch.
+
+    See the rationale in cluster_grace_shutdown's loop.
+    """
+    return node.status in (StorageNode.STATUS_REMOVED,
+                           StorageNode.STATUS_IN_REMOVAL)
+
+
 def cluster_grace_shutdown(cl_id) -> None:
     db_controller.get_cluster_by_id(cl_id)  # ensure exists
 
     st = db_controller.get_storage_nodes_by_cluster_id(cl_id)
     for node in st:
+        # REMOVED is terminal and must survive a cluster shutdown. Without
+        # this filter the sweep force-shuts-down every record it can see,
+        # and shutdown_storage_node drives in_shutdown -> offline, so a node
+        # that was deliberately removed comes back as a plain offline member
+        # (live 2026-09-03: one graceful-shutdown resurrected all four nodes
+        # removed earlier that day). That is not cosmetic --
+        # failure_domain_host_map skips only STATUS_REMOVED, so those records
+        # start counting toward FD host balance again, and the next
+        # activation or startup acts on nodes whose devices are already
+        # failed_and_migrated and which own no lvstore.
+        #
+        # IN_REMOVAL is skipped because node_removal_orchestrate has already
+        # shut that node down and owns the rest of its lifecycle.
+        # PENDING_REMOVAL is deliberately NOT skipped -- the node is still up
+        # and serving at that point, so a full-cluster shutdown must stop it
+        # like any other member.
+        if _grace_shutdown_skipped(node):
+            logger.info(f"Skipping node {node.get_id()} with status: {node.status}")
+            continue
         logger.info(f"Suspending node: {node.get_id()}")
         storage_node_ops.suspend_storage_node(node.get_id(), force=True)
         logger.info(f"Shutting down node: {node.get_id()}")
         storage_node_ops.shutdown_storage_node(node.get_id(), force=True)
 
+    # Settle check. The sweep is serial, so a node it already passed can come
+    # back up behind it -- that is exactly what happened on 2026-09-03, when
+    # queued restart rows put s7457 and zdgtb back ONLINE seconds after the
+    # sweep had shut them down, and the command still returned as if the
+    # cluster were down. shutdown_storage_node now reaps those rows, but this
+    # verifies the end state rather than assuming it: anything that resurrects
+    # a node by another route is caught and stopped here, once.
+    st = db_controller.get_storage_nodes_by_cluster_id(cl_id)
+    stragglers = [n for n in st
+                  if not _grace_shutdown_skipped(n)
+                  and n.status != StorageNode.STATUS_OFFLINE]
+    for node in stragglers:
+        logger.warning(
+            f"Node {node.get_id()} is {node.status} after the shutdown sweep; "
+            f"shutting it down again")
+        storage_node_ops.shutdown_storage_node(node.get_id(), force=True)
+
+    st = db_controller.get_storage_nodes_by_cluster_id(cl_id)
+    still_up = [n.get_id() for n in st
+                if not _grace_shutdown_skipped(n)
+                and n.status != StorageNode.STATUS_OFFLINE]
+    if still_up:
+        # Deliberately not raising: the caller asked for a shutdown and most
+        # of the cluster is down, so failing here would be less useful than
+        # saying precisely which nodes are not. An operator following up with
+        # `sn shutdown` needs the list, not a traceback.
+        logger.error(
+            f"Graceful shutdown finished with {len(still_up)} node(s) not "
+            f"offline: {still_up}")
+
 
 def cluster_restart(cl_id) -> None:
     """Operator-requested full-cluster restart: force-shutdown every node that

simplyblock_core/controllers/tasks_controller.py

@@ -524,26 +524,38 @@ def add_node_to_auto_restart(node):
     return _add_task(JobSchedule.FN_NODE_RESTART, node.cluster_id, node.get_id(), "", max_retry=11)
 
 
-def cancel_pending_node_restart_tasks(cluster_id, node_id):
+def cancel_pending_node_restart_tasks(cluster_id, node_id, exclude_task_id=None,
+                                      reason="node back online"):
     # Called from set_node_status the moment a node transitions to ONLINE.
     # Without this, an obsolete FN_NODE_RESTART row left over from the
     # outage stays in `new`/`running` and blocks every subsequent restart
     # via the dedup guard in `_validate_new_task_node_restart` until the
     # task runner happens to pick it up — observed as a 5-minute window
     # of failing manual restarts after the node was already back online.
+    #
+    # Also called from shutdown_storage_node on the opposite transition, so a
+    # row queued before the deliberate stop cannot fire afterwards and undo it.
+    #
+    # exclude_task_id: the caller's OWN task, left alone. The restart runner
+    # drives shutdown_storage_node as its kill step
+    # (tasks_runner_restart.py:542, passing current_restart_task_id), so a
+    # blanket cancel there would abort the very restart doing the shutting
+    # down. Compared against the bare task uuid, matching the convention in
+    # check_node_shutdown_preconditions.
     canceled = 0
     for task in db.get_job_tasks(cluster_id):
         if (task.function_name == JobSchedule.FN_NODE_RESTART
                 and task.node_id == node_id
                 and task.status != JobSchedule.STATUS_DONE
-                and not task.canceled):
+                and not task.canceled
+                and (exclude_task_id is None or task.uuid != exclude_task_id)):
             task.canceled = True
             task.status = JobSchedule.STATUS_DONE
-            task.function_result = "canceled: node back online"
+            task.function_result = f"canceled: {reason}"
             task.write_to_db(db.kv_store)
             canceled += 1
             logger.info(
-                f"Canceled obsolete node_restart task {task.get_id()} (node {node_id} back online)")
+                f"Canceled obsolete node_restart task {task.get_id()} (node {node_id}: {reason})")
     return canceled
 
 

simplyblock_core/storage_node_ops.py

@@ -7098,6 +7098,32 @@ def shutdown_storage_node(node_id, force=False, keep_auto_restart=False,
     if not keep_auto_restart:
         snode.auto_restart_disabled = True
         snode.write_to_db(db_controller.kv_store)
+        # The flag alone is not enough: it is enforced at ENQUEUE time only
+        # (tasks_controller.add_node_to_auto_restart, "the single chokepoint
+        # for every auto-restart queue path"). tasks_runner_restart never
+        # consults it. So an FN_NODE_RESTART row queued BEFORE the flag was
+        # set survives, executes unconditionally, and its ONLINE transition
+        # clears the flag again -- the deliberate-stop intent destroyed by the
+        # very task it was meant to prevent. Live 2026-09-03: a cluster
+        # graceful-shutdown left s7457 and zdgtb ONLINE because two such rows
+        # fired seconds after the sweep passed them.
+        #
+        # Reap them here, mirroring what set_node_status already does on the
+        # opposite transition (ONLINE cancels obsolete restart rows). This
+        # also gives the runner its dequeue-side check for free -- it already
+        # honours task.canceled -- without a new field and without breaking
+        # ensure_node_restart_task, which deliberately bypasses the flag
+        # because an explicit `sn restart` is the operator intervention the
+        # flag is waiting for. A restart queued AFTER this point is exactly
+        # that, and is left alone.
+        #
+        # current_restart_task_id is excluded: the restart runner drives this
+        # very function as its kill step, so cancelling its task here would
+        # abort the restart that is doing the shutting down.
+        tasks_controller.cancel_pending_node_restart_tasks(
+            snode.cluster_id, node_id,
+            exclude_task_id=current_restart_task_id,
+            reason="node deliberately shut down")
 
     # Step 2: cancel migration tasks while controllers are still up.
     pending_tasks = db_controller.get_job_tasks(snode.cluster_id)

tests/unit/test_cancel_pending_node_restart_tasks.py

@@ -0,0 +1,88 @@
+# coding=utf-8
+"""
+``tasks_controller.cancel_pending_node_restart_tasks`` exclusion semantics.
+
+The helper existed to reap obsolete FN_NODE_RESTART rows when a node comes
+back ONLINE. ``shutdown_storage_node`` now also calls it on the opposite
+transition, so a row queued before a deliberate stop cannot fire afterwards
+and undo it (live 2026-09-03: a cluster graceful-shutdown left two nodes
+ONLINE because queued rows fired just behind the sweep, and the resulting
+ONLINE transition cleared their auto_restart_disabled flag).
+
+That second caller makes the exclusion load-bearing. The restart runner drives
+shutdown_storage_node as its own kill step (tasks_runner_restart.py:542,
+passing current_restart_task_id), so a blanket cancel there would abort the
+very restart performing the shutdown -- turning a fix for a rare race into a
+failure on every single node restart.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from simplyblock_core.controllers import tasks_controller
+from simplyblock_core.models.job_schedule import JobSchedule
+
+
+def _task(uuid, node_id="n1", function_name=JobSchedule.FN_NODE_RESTART,
+          status=JobSchedule.STATUS_NEW, canceled=False):
+    t = MagicMock(spec=JobSchedule)
+    t.uuid = uuid
+    t.node_id = node_id
+    t.function_name = function_name
+    t.status = status
+    t.canceled = canceled
+    t.get_id = MagicMock(return_value=f"c1/{uuid}")
+    t.write_to_db = MagicMock(return_value=True)
+    return t
+
+
+class TestCancelPendingNodeRestartTasks(unittest.TestCase):
+
+    def _run(self, tasks, **kwargs):
+        db = MagicMock()
+        db.get_job_tasks = MagicMock(return_value=tasks)
+        with patch.object(tasks_controller, "db", db):
+            n = tasks_controller.cancel_pending_node_restart_tasks(
+                "c1", "n1", **kwargs)
+        return n
+
+    def test_pending_rows_are_canceled(self):
+        t = _task("t1")
+        self.assertEqual(self._run([t]), 1)
+        self.assertTrue(t.canceled)
+        self.assertEqual(t.status, JobSchedule.STATUS_DONE)
+
+    def test_the_callers_own_task_is_left_alone(self):
+        # The regression this guards: cancelling it aborts the restart that
+        # is driving the shutdown.
+        mine, other = _task("mine"), _task("other")
+        self.assertEqual(self._run([mine, other], exclude_task_id="mine"), 1)
+        self.assertFalse(mine.canceled)
+        self.assertNotEqual(mine.status, JobSchedule.STATUS_DONE)
+        self.assertTrue(other.canceled)
+
+    def test_no_exclusion_still_cancels_everything(self):
+        # set_node_status's ONLINE path passes no exclusion; behaviour there
+        # must be unchanged.
+        a, b = _task("a"), _task("b")
+        self.assertEqual(self._run([a, b]), 2)
+        self.assertTrue(a.canceled and b.canceled)
+
+    def test_other_nodes_and_other_task_types_are_untouched(self):
+        other_node = _task("t1", node_id="n2")
+        other_kind = _task("t2", function_name=JobSchedule.FN_DEV_MIG)
+        done = _task("t3", status=JobSchedule.STATUS_DONE)
+        already = _task("t4", canceled=True)
+        self.assertEqual(self._run([other_node, other_kind, done, already]), 0)
+        self.assertFalse(other_node.canceled)
+        self.assertFalse(other_kind.canceled)
+
+    def test_reason_is_recorded(self):
+        t = _task("t1")
+        self._run([t], reason="node deliberately shut down")
+        self.assertEqual(t.function_result,
+                         "canceled: node deliberately shut down")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_graceful_shutdown_sweep.py

@@ -0,0 +1,124 @@
+# coding=utf-8
+"""
+``cluster_ops.cluster_grace_shutdown`` must not act on departed nodes.
+
+Found live 2026-09-03. The sweep enumerated every node record with no status
+filter at all and force-shut-down each one. ``shutdown_storage_node`` drives
+in_shutdown -> offline, so nodes that had been REMOVED came back as plain
+offline members and had to be repaired by hand.
+
+That is not cosmetic: ``failure_domain_host_map`` skips only STATUS_REMOVED,
+so the resurrected records immediately start occupying failure-domain host
+slots again -- the cluster went from 8 hosts at 2/2/2/2 to 12 at 3/3/3/3 --
+and a later activation or startup then acts on nodes whose devices are
+already failed_and_migrated and which own no lvstore.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from simplyblock_core import cluster_ops
+from simplyblock_core.models.storage_node import StorageNode
+
+
+def _node(node_id, status=StorageNode.STATUS_ONLINE):
+    n = MagicMock(spec=StorageNode)
+    n.uuid = node_id
+    n.get_id = MagicMock(return_value=node_id)
+    n.status = status
+    n.cluster_id = "c1"
+    return n
+
+
+class TestGracefulShutdownSkipsDepartedNodes(unittest.TestCase):
+
+    def _run(self, nodes, resurrect=()):
+        """Drive the sweep. A real shutdown drives the node to OFFLINE, so the
+        mock does too -- otherwise the settle check would see every node as a
+        straggler. ``resurrect`` names nodes that come back ONLINE behind the
+        sweep (the live 2026-09-03 failure) and are only stopped for good on
+        the second attempt."""
+        by_id = {n.get_id(): n for n in nodes}
+        pending = dict.fromkeys(resurrect, True)
+
+        def _shutdown(node_id, **_kw):
+            node = by_id[node_id]
+            if pending.pop(node_id, False):
+                node.status = StorageNode.STATUS_ONLINE
+            else:
+                node.status = StorageNode.STATUS_OFFLINE
+            return True
+
+        db = MagicMock()
+        db.get_cluster_by_id = MagicMock(return_value=MagicMock())
+        db.get_storage_nodes_by_cluster_id = MagicMock(return_value=nodes)
+        sn = MagicMock()
+        sn.shutdown_storage_node = MagicMock(side_effect=_shutdown)
+        with patch.object(cluster_ops, "db_controller", db), \
+                patch.object(cluster_ops, "storage_node_ops", sn):
+            cluster_ops.cluster_grace_shutdown("c1")
+        swept = [c.args[0] for c in sn.shutdown_storage_node.call_args_list]
+        suspended = [c.args[0] for c in sn.suspend_storage_node.call_args_list]
+        return swept, suspended
+
+    def test_a_removed_node_is_never_shut_down(self):
+        # The live failure: one graceful-shutdown turned four removed nodes
+        # back into offline members.
+        swept, suspended = self._run([
+            _node("live", StorageNode.STATUS_ONLINE),
+            _node("gone", StorageNode.STATUS_REMOVED),
+        ])
+        self.assertEqual(swept, ["live"])
+        self.assertEqual(suspended, ["live"])
+
+    def test_a_node_mid_removal_is_left_to_its_orchestrator(self):
+        swept, _ = self._run([
+            _node("live", StorageNode.STATUS_ONLINE),
+            _node("leaving", StorageNode.STATUS_IN_REMOVAL),
+        ])
+        self.assertEqual(swept, ["live"])
+
+    def test_everything_else_is_still_swept(self):
+        # The point of the command: offline/unreachable/restarting members
+        # and nodes only queued for removal must all be stopped. In
+        # particular PENDING_REMOVAL is still up and serving, so a
+        # full-cluster shutdown has to stop it like any other member.
+        nodes = [
+            _node("online", StorageNode.STATUS_ONLINE),
+            _node("offline", StorageNode.STATUS_OFFLINE),
+            _node("unreachable", StorageNode.STATUS_UNREACHABLE),
+            _node("restarting", StorageNode.STATUS_RESTARTING),
+            _node("pending", StorageNode.STATUS_PENDING_REMOVAL),
+        ]
+        swept, _ = self._run(nodes)
+        self.assertEqual(
+            swept, ["online", "offline", "unreachable", "restarting", "pending"])
+
+
+class TestGracefulShutdownSettles(unittest.TestCase):
+    """The sweep is serial, so a node it already passed can come back behind
+    it. Live 2026-09-03: queued restart rows put s7457 and zdgtb back ONLINE
+    seconds after the sweep shut them down, and the command still returned as
+    though the cluster were down."""
+
+    _run = TestGracefulShutdownSkipsDepartedNodes._run
+
+    def test_a_node_that_comes_back_is_shut_down_again(self):
+        swept, _ = self._run(
+            [_node("a"), _node("b")], resurrect=("a",))
+        self.assertEqual(swept, ["a", "b", "a"])
+
+    def test_no_second_pass_when_everything_settled(self):
+        swept, _ = self._run([_node("a"), _node("b")])
+        self.assertEqual(swept, ["a", "b"])
+
+    def test_a_removed_node_is_not_treated_as_a_straggler(self):
+        # It is never OFFLINE, so an unfiltered settle check would keep
+        # trying to shut it down -- reintroducing the resurrection bug.
+        swept, _ = self._run(
+            [_node("live"), _node("gone", StorageNode.STATUS_REMOVED)])
+        self.assertEqual(swept, ["live"])
+
+
+if __name__ == "__main__":
+    unittest.main()