feat(node-removal): online node removal as inverse of cluster expansion

simplyblock_cli/cli-reference.yaml

@@ -297,22 +297,19 @@ commands:
             type: bool
             action: store_true
       - name: remove
-        help: "Removes a storage node from the cluster."
-        description: >
-          The storage node cannot be used or added any more. Any data residing on this storage node will be migrated to
-          the remaining storage nodes. The user must ensure that there is sufficient free space in remaining cluster to
-          allow for successful node removal.
-
-          !!! danger
-              If there isn't enough storage available, the cluster may run full and switch to read-only mode.
+        help: 'Removes a storage node from the cluster (inverse of cluster expansion). 
+        The node must be ONLINE and all other nodes ONLINE; it must host no logical 
+        volumes or snapshots (migrate those first). Runs as a background task: shuts 
+        the node down, rewires LVS replicas onto other nodes, then removes, fails and 
+        migrates its devices before marking the node removed.'
         arguments:
           - name: node_id
             help: Storage node id
             dest: node_id
             completer: _completer_get_sn_list
             type: str
           - name: "--force-remove"
-            help: "Force remove all logical volumes and snapshots."
+            help: "Cancel any active tasks on the node before starting removal."
             dest: force_remove
             type: bool
             action: store_true
@@ -449,21 +446,16 @@ commands:
             type: bool
             action: store_true
       - name: suspend
-        help: "DEPRECATED: the suspension phase was removed from graceful shutdown (it caused writer conflicts on sec/tert lvstores). This command is now a no-op returning success. Use `sn shutdown`."
-        usage: The node will stop accepting new IO, but will finish processing any IO, which has been received already.
+        help: "Exclude node from lvol allocation."
+        usage: The node will be excluded from lvol allocation. It will not be able to create new logical volumes on this node.
         arguments:
           - name: node_id
             help: Storage node id
             dest: node_id
             completer: _completer_get_sn_list
             type: str
-          - name: "--force"
-            help: "Ignored (kept for backwards compatibility)."
-            dest: force
-            type: bool
-            action: store_true
       - name: resume
-        help: "DEPRECATED: counterpart to `sn suspend`, also a no-op."
+        help: "Include node in lvol allocation."
         arguments:
           - name: node_id
             help: Storage node id

simplyblock_cli/cli.py

@@ -165,9 +165,9 @@ def init_storage_node__delete(self, subparser):
         subcommand.add_argument('--force', help='Force delete storage node from DB. Ensure you know what you\'re doing.', dest='force_remove', action='store_true')
 
     def init_storage_node__remove(self, subparser):
-        subcommand = self.add_sub_command(subparser, 'remove', 'Removes a storage node from the cluster.')
+        subcommand = self.add_sub_command(subparser, 'remove', 'Removes a storage node from the cluster (inverse of cluster expansion). The node must be ONLINE and all other nodes ONLINE; it must host no logical volumes or snapshots (migrate those first). Runs as a background task: shuts the node down, rewires LVS replicas onto other nodes, then removes, fails and migrates its devices before marking the node removed.')
         subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list
-        subcommand.add_argument('--force-remove', help='Force remove all logical volumes and snapshots.', dest='force_remove', action='store_true')
+        subcommand.add_argument('--force-remove', help='Cancel any active tasks on the node before starting removal.', dest='force_remove', action='store_true')
 
     def init_storage_node__list(self, subparser):
         subcommand = self.add_sub_command(subparser, 'list', 'Lists all storage nodes.')
@@ -209,12 +209,11 @@ def init_storage_node__shutdown(self, subparser):
         subcommand.add_argument('--force', help='Force node shutdown.', dest='force', action='store_true')
 
     def init_storage_node__suspend(self, subparser):
-        subcommand = self.add_sub_command(subparser, 'suspend', 'DEPRECATED: the suspension phase was removed from graceful shutdown (it caused writer conflicts on sec/tert lvstores). This command is now a no-op returning success. Use `sn shutdown`.')
+        subcommand = self.add_sub_command(subparser, 'suspend', 'Exclude node from lvol allocation.')
         subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list
-        subcommand.add_argument('--force', help='Ignored (kept for backwards compatibility).', dest='force', action='store_true')
 
     def init_storage_node__resume(self, subparser):
-        subcommand = self.add_sub_command(subparser, 'resume', 'DEPRECATED: counterpart to `sn suspend`, also a no-op.')
+        subcommand = self.add_sub_command(subparser, 'resume', 'Include node in lvol allocation.')
         subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list
 
     def init_storage_node__get_io_stats(self, subparser):

simplyblock_cli/clibase.py

@@ -301,7 +301,7 @@ def storage_node__shutdown(self, sub_command, args):
         return ret
 
     def storage_node__suspend(self, sub_command, args):
-        ret = storage_ops.suspend_storage_node(args.node_id, args.force)
+        ret = storage_ops.suspend_storage_node(args.node_id)
         if isinstance(ret, tuple):
             ok, reason = ret
             if not ok:

simplyblock_core/cluster_ops.py

@@ -667,6 +667,27 @@ def _wait_for_full_device_connectivity(cl_id, timeout_sec=300, poll_sec=10):
 
 def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None:
     cluster = db_controller.get_cluster_by_id(cl_id)
+    prev_status = cluster.status
+    if prev_status == Cluster.STATUS_IN_ACTIVATION:
+        prev_status = Cluster.STATUS_UNREADY
+    try:
+        _cluster_activate(cl_id, force=force, force_lvstore_create=force_lvstore_create)
+    except Exception:
+        # Never leave the cluster wedged in in_activation: this often runs in
+        # a fire-and-forget thread, and an unhandled failure would otherwise
+        # block any retry (the activate API rejects in_activation clusters).
+        # The expected-failure paths inside _cluster_activate restore the
+        # status themselves; this only catches what they missed.
+        cluster = db_controller.get_cluster_by_id(cl_id)
+        if cluster.status == Cluster.STATUS_IN_ACTIVATION:
+            logger.error("Cluster activation failed unexpectedly; reverting status "
+                         f"from {Cluster.STATUS_IN_ACTIVATION} to {prev_status}")
+            set_cluster_status(cl_id, prev_status)
+        raise
+
+
+def _cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None:
+    cluster = db_controller.get_cluster_by_id(cl_id)
 
     if cluster.status == Cluster.STATUS_ACTIVE:
         logger.warning("Cluster is ACTIVE")

simplyblock_core/constants.py

@@ -100,12 +100,19 @@ def get_config_var(name, default=None):
 
 # A JobSchedule's lease is held by the runner host (by hostname) that last
 # touched it. Another host may only take over a task whose lease is older than
-# this — i.e. the owning runner is presumed dead. Must exceed the longest
-# single task_runner() invocation that does not write the task back (the
-# restart runner can block on RPCs for several minutes), so a live owner is
-# never falsely preempted. A runner restarting on the SAME host re-claims its
-# own tasks immediately regardless of this value (owner id is the hostname).
-TASK_LEASE_TTL_SEC = 1200
+# this — i.e. the owning runner is presumed dead. A live owner refreshes the
+# lease every TASK_LEASE_HEARTBEAT_SEC from a background thread while driving
+# a restart (restart_storage_node wrapper), so the TTL no longer needs to
+# exceed the longest blocking RPC — it only needs to be several heartbeats
+# wide so a momentarily slow (but alive) owner is never falsely preempted.
+# Keeping it short is what makes ownership transfer fast: when the driving
+# process dies (pod evicted while its host drains, CLI killed), a live
+# tasks-runner claims the stale lease and resumes the restart instead of the
+# node staying orphaned in RESTARTING (2026-07-04 MCO rollout deadlock).
+# A runner restarting on the SAME host re-claims its own tasks immediately
+# regardless of this value (owner id is the hostname).
+TASK_LEASE_HEARTBEAT_SEC = 30
+TASK_LEASE_TTL_SEC = 180
 
 # Node-add concurrency: the cross-node mesh section of add_node is serialized
 # per cluster behind a ClusterAddNodeLock. The holder refreshes the lock every

simplyblock_core/controllers/health_controller.py

@@ -510,7 +510,7 @@ def _check_node_lvstore(
                                         dev = db_controller.get_storage_device_by_id(result['UUID'])
                                         dev_node = db_controller.get_storage_node_by_id(dev.node_id)
                                         if dev.status == NVMeDevice.STATUS_ONLINE and dev_node.status in [
-                                            StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_UNREACHABLE]:
+                                            StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_UNREACHABLE, StorageNode.STATUS_SUSPENDED]:
                                             try:
                                                 remote_bdev = storage_node_ops.connect_device(
                                                     f"remote_{dev.alceml_bdev}", dev, node)
@@ -713,7 +713,7 @@ def check_node(node_id, with_devices=True):
             logger.info(f"Node remote JMs: {len(snode.remote_jm_devices)}")
             for remote_device in snode.remote_jm_devices:
 
-                name = f'remote_{remote_device.jm_bdev}n1'
+                name = remote_device.remote_bdev
                 bdev_info = rpc_client.get_bdevs(name)
                 logger.log(INFO if bdev_info else ERROR,
                            f"Checking bdev: {name} ... " + ('ok' if bdev_info else 'failed'))

simplyblock_core/controllers/tasks_controller.py

@@ -69,6 +69,48 @@ def _mutate(t):
     return decision["won"]
 
 
+def refresh_task_lease(task, owner=None):
+    """Heartbeat: refresh this host's lease on a task it already owns, so a
+    live owner is never preempted while blocking on long RPCs. Returns False
+    (without touching the task) if the task is done or owned by another host —
+    the caller lost the lease and should treat the takeover as authoritative."""
+    owner = owner or _RUNNER_HOST
+    now = str(datetime.datetime.now(datetime.timezone.utc))
+    refreshed = {"ok": False}
+
+    def _mutate(t):
+        if t.status == JobSchedule.STATUS_DONE:
+            return False
+        if t.owner != owner:
+            return False
+        t.updated_at = now
+        refreshed["ok"] = True
+        return True
+
+    if db.atomic_update(task, _mutate) is None:
+        return False
+    return refreshed["ok"]
+
+
+def ensure_node_restart_task(node):
+    """Return the id of an unfinished FN_NODE_RESTART task for this node,
+    creating one if none exists. Used by explicit restarts
+    (restart_storage_node wrapper) to make their ownership transferable: the
+    driver claims and heartbeats the task's lease, and if the driving process
+    dies mid-restart, a live tasks-runner claims the stale lease and resumes.
+
+    Deliberately bypasses the auto_restart_disabled guard of
+    add_node_to_auto_restart: that flag means "no UNATTENDED restart after a
+    deliberate shutdown" — an explicit restart is precisely the operator
+    intervention the flag waits for, and this task only continues that
+    expressed intent. On success the ONLINE transition cancels the task
+    (cancel_pending_node_restart_tasks via set_node_status)."""
+    existing = _validate_new_task_node_restart(node.cluster_id, node.get_id())
+    if existing:
+        return existing
+    return _add_task(JobSchedule.FN_NODE_RESTART, node.cluster_id, node.get_id(), "", max_retry=11)
+
+
 def _validate_new_task_dev_restart(cluster_id, node_id, device_id):
     tasks = db.get_job_tasks(cluster_id)
     for task in tasks:
@@ -174,6 +216,12 @@ def _add_task(function_name, cluster_id, node_id, device_id,
             logger.info(f"Task found, skip adding new task: {task_id}")
             return False
 
+    elif function_name == JobSchedule.FN_NODE_REMOVAL:
+        task_id = get_active_node_removal_task(cluster_id, node_id)
+        if task_id:
+            logger.info(f"Task found, skip adding new task: {task_id}")
+            return False
+
     task_obj = JobSchedule()
     task_obj.uuid = str(uuid.uuid4())
     task_obj.cluster_id = cluster_id
@@ -551,6 +599,10 @@ def get_active_node_mig_task(cluster_id, node_id, distr_name=None):
 def add_device_failed_mig_task(device_id):
     device = db.get_storage_device_by_id(device_id)
     for node in db.get_storage_nodes_by_cluster_id(device.cluster_id):
+        # IN_REMOVAL nodes have a dead SPDK (shut down by the removal flow);
+        # a migration task targeting their distribs can never run and would
+        # stall the node-removal completion check forever. Skip them like
+        # already-REMOVED nodes.
         if node.status == StorageNode.STATUS_REMOVED:
             continue
         for bdev in node.lvstore_stack:
@@ -577,6 +629,23 @@ def add_node_add_task(cluster_id, function_params):
                      function_params=function_params, max_retry=11)
 
 
+def add_node_removal_task(cluster_id, node_id, function_params=None):
+    # max_retry=-1: the removal runner drives a multi-step, possibly multi-hour
+    # orchestration (shutdown -> LVS rewire -> device fail+migrate). Migration
+    # waits legitimately suspend-and-retry many times; do not cap retries.
+    return _add_task(JobSchedule.FN_NODE_REMOVAL, cluster_id, node_id, "",
+                     function_params=function_params or {}, max_retry=-1)
+
+
+def get_active_node_removal_task(cluster_id, node_id):
+    tasks = db.get_job_tasks(cluster_id)
+    for task in tasks:
+        if task.function_name == JobSchedule.FN_NODE_REMOVAL and task.node_id == node_id:
+            if task.status != JobSchedule.STATUS_DONE and task.canceled is False:
+                return task.uuid
+    return False
+
+
 def add_cluster_expand_task(cluster_id, new_node_id):
     """Queue a single-node cluster-expansion task. The runner drives the
     planner/orchestrator/executor to integrate ``new_node_id`` into the
@@ -627,7 +696,7 @@ def get_active_node_tasks(cluster_id, node_id):
     tasks = db.get_job_tasks(cluster_id)
     out = []
     for task in tasks:
-        if task.function_name in [JobSchedule.FN_PORT_ALLOW, JobSchedule.FN_JC_COMP_RESUME]:
+        if task.function_name in [JobSchedule.FN_PORT_ALLOW, JobSchedule.FN_JC_COMP_RESUME, JobSchedule.FN_NODE_REMOVAL]:
             continue
         if task.node_id == node_id:
             if task.status != JobSchedule.STATUS_DONE and task.canceled is False:

simplyblock_core/env_var

@@ -1,5 +1,5 @@
 SIMPLY_BLOCK_COMMAND_NAME=sbcli-dev
 SIMPLY_BLOCK_VERSION=19.2.34
 
-SIMPLY_BLOCK_DOCKER_IMAGE=public.ecr.aws/simply-block/simplyblock:main
+SIMPLY_BLOCK_DOCKER_IMAGE=public.ecr.aws/simply-block/simplyblock:feature-node-removal
 SIMPLY_BLOCK_SPDK_ULTRA_IMAGE=public.ecr.aws/simply-block/ultra:main-latest

simplyblock_core/models/job_schedule.py

@@ -17,6 +17,7 @@ class JobSchedule(BaseModel):
     FN_FAILED_DEV_MIG = "failed_device_migration"
     FN_NEW_DEV_MIG = "new_device_migration"
     FN_NODE_ADD = "node_add"
+    FN_NODE_REMOVAL = "node_removal"
     FN_PORT_ALLOW = "port_allow"
     FN_BALANCING_AFTER_NODE_RESTART = "balancing_on_restart"
     FN_BALANCING_AFTER_DEV_REMOVE = "balancing_on_dev_rem"

simplyblock_core/models/nvme_device.py

@@ -108,6 +108,9 @@ class JMDevice(NVMeDevice):
     # the per-leg member partitions. Empty for single-device (no-raid) JMs.
     jm_leg_bdevs: List[str] = []
     jm_leg_members: List = []
+    # When attaching this JM to a node, override the device name on that node.
+    # This is needed when a JM device is removed and needed to be replaced, but the name must be the same.
+    override_name_on_node: dict[str, str] = {} # node_id: new_name
 
 
 class RemoteDevice(BaseModel):

simplyblock_core/scripts/docker-compose-swarm.yml

@@ -484,6 +484,20 @@ services:
     environment:
       SIMPLYBLOCK_LOG_LEVEL: "$LOG_LEVEL"
 
+  TasksNodeRemovalRunner:
+    <<: *service-base
+    image: $SIMPLYBLOCK_DOCKER_IMAGE
+    command: "python3 simplyblock_core/services/tasks_runner_node_removal.py"
+    deploy:
+      placement:
+        constraints: [node.role == manager]
+    volumes:
+      - "/etc/foundationdb:/etc/foundationdb"
+    networks:
+      - hostnet
+    environment:
+      SIMPLYBLOCK_LOG_LEVEL: "$LOG_LEVEL"
+
   TasksRunnerClusterExpand:
     <<: *service-base
     image: $SIMPLYBLOCK_DOCKER_IMAGE

simplyblock_core/services/health_check_service.py

@@ -102,7 +102,7 @@ def check_node(snode):
     # (reconnecting remote devices, repairing multipath, recreating hublvols) —
     # but their node/device health is reported as "not applicable" (None),
     # never true/false.
-    report_health = snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN]
+    report_health = snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED]
 
     # 1- check node ping
     ping_check = health_controller._check_node_ping(snode.mgmt_ip)

simplyblock_core/services/storage_node_monitor.py

@@ -615,6 +615,7 @@ def _spawn_recovery_shutdown(node):
 
     def _run():
         try:
+
             storage_node_ops.shutdown_storage_node(
                 node_id, force=True, keep_auto_restart=True)
         except Exception as e:

simplyblock_core/services/tasks_runner_migration.py

@@ -19,7 +19,7 @@ def _cluster_unavailable_state(cluster_id):
     for node in db.get_storage_nodes_by_cluster_id(cluster_id):
         if node.status in [StorageNode.STATUS_IN_CREATION, StorageNode.STATUS_REMOVED]:
             continue
-        if node.status != StorageNode.STATUS_ONLINE:
+        if node.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]:
             unavailable.append(f"node:{node.get_id()}")
         for dev in node.nvme_devices:
             if dev.status in [NVMeDevice.STATUS_REMOVED, NVMeDevice.STATUS_FAILED_AND_MIGRATED]:
@@ -75,7 +75,7 @@ def task_runner(task):
         task.write_to_db(db.kv_store)
         return True
 
-    if snode.status != StorageNode.STATUS_ONLINE:
+    if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]:
         task.function_result = "node is not online, retrying"
         task.status = JobSchedule.STATUS_SUSPENDED
         unavailable = _cluster_unavailable_state(task.cluster_id)
@@ -307,7 +307,7 @@ def _set_master_task_status(master_task, status):
                             logger.info("no task found on same node, resuming compression")
                             node = db.get_storage_node_by_id(task.node_id)
                             for n in db.get_storage_nodes_by_cluster_id(node.cluster_id):
-                                if n.status != StorageNode.STATUS_ONLINE:
+                                if n.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]:
                                     logger.warning("Not all nodes are online, can not resume JC compression")
                                     continue
                             rpc_client = node.rpc_client(timeout=5, retry=2)