From 197c062213ddfdf8872b350eaba93d6917bccd8b Mon Sep 17 00:00:00 2001 From: michael Date: Thu, 13 Aug 2026 23:12:13 +0200 Subject: [PATCH 1/5] fix(storage_node): add statuses for node removal and update removal logic --- simplyblock_core/distr_controller.py | 6 ++- simplyblock_core/models/base_model.py | 4 ++ .../services/snapshot_replication.py | 40 ++++++++++++---- .../services/tasks_runner_node_removal.py | 15 +++--- simplyblock_core/storage_node_ops.py | 48 ++++++++++--------- 5 files changed, 74 insertions(+), 39 deletions(-) diff --git a/simplyblock_core/distr_controller.py b/simplyblock_core/distr_controller.py index 6cca30acf7..cddd8fd243 100644 --- a/simplyblock_core/distr_controller.py +++ b/simplyblock_core/distr_controller.py @@ -63,7 +63,7 @@ def _mutate(node): def send_node_status_event(node, node_status, target_node=None): db_controller = DBController() node_id = node.get_id() - if node_status == StorageNode.STATUS_SCHEDULABLE: + if node_status in [StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_IN_REMOVAL, StorageNode.STATUS_PENDING_REMOVAL]: node_status = StorageNode.STATUS_UNREACHABLE logger.info(f"Sending event updates, node: {node_id}, status: {node_status}") node_status_event = { @@ -251,7 +251,7 @@ def get_distr_cluster_map(snodes, target_node, distr_name=""): node_w += dev_w_gib node_status = snode.status - if node_status == StorageNode.STATUS_SCHEDULABLE: + if node_status in [StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_IN_REMOVAL, StorageNode.STATUS_PENDING_REMOVAL]: node_status = StorageNode.STATUS_UNREACHABLE map_cluster[snode.get_id()] = { "status": node_status, @@ -348,6 +348,8 @@ def parse_distr_cluster_map(map_string, nodes=None, devices=None): StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_RESTARTING, StorageNode.STATUS_IN_SHUTDOWN, + StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_PENDING_REMOVAL ): node_status = StorageNode.STATUS_UNREACHABLE data["Desired Status"] = node_status diff --git a/simplyblock_core/models/base_model.py b/simplyblock_core/models/base_model.py index c0a2cadc4e..505fc9b69b 100644 --- a/simplyblock_core/models/base_model.py +++ b/simplyblock_core/models/base_model.py @@ -374,6 +374,8 @@ class BaseNodeObject(BaseModel): STATUS_UNREACHABLE = 'unreachable' STATUS_SCHEDULABLE = 'schedulable' STATUS_DOWN = 'down' + STATUS_IN_REMOVAL = 'in_removal' + STATUS_PENDING_REMOVAL = 'pending_removal' _STATUS_CODE_MAP = { STATUS_ONLINE: 0, @@ -386,4 +388,6 @@ class BaseNodeObject(BaseModel): STATUS_UNREACHABLE: 20, STATUS_SCHEDULABLE: 30, STATUS_DOWN: 40, + STATUS_IN_REMOVAL: 41, + STATUS_PENDING_REMOVAL: 42, } diff --git a/simplyblock_core/services/snapshot_replication.py b/simplyblock_core/services/snapshot_replication.py index 7bdc5b3e77..f4ea5691e2 100644 --- a/simplyblock_core/services/snapshot_replication.py +++ b/simplyblock_core/services/snapshot_replication.py @@ -156,6 +156,25 @@ def _require_lvs_leader(node, lvs_name, what): return False +def _other_active_transfers_to_node(current_task, target_node_id): + """True when another RUNNING snapshot-replication task is transferring into + *target_node_id* — its writes ride the same shared hub session, so the hub + must not be detached under it.""" + for t in db.get_job_tasks(current_task.cluster_id): + if (t.function_name == JobSchedule.FN_SNAPSHOT_REPLICATION + and t.get_id() != current_task.get_id() + and t.status == JobSchedule.STATUS_RUNNING): + rid = t.function_params.get("remote_lvol_id") + if not rid: + continue + try: + if db.get_lvol_by_id(rid).node_id == target_node_id: + return True + except KeyError: + continue + return False + + def _has_dependent_clone(snapshot_uuid): """True when any live volume is cloned from *snapshot_uuid*. @@ -228,16 +247,20 @@ def _prune_internal_snapshots(source_lvol): def process_snap_replicate_finish(task, snapshot): - # Close the transfer session: detach the target's transfer hub from the - # source node. The connect -> transfer -> convert -> DISCONNECT cycle is - # part of the transfer contract; the next cycle re-attaches via - # ensure_hub_attached. + # Close the transfer session — but ONLY when this was the last active + # transfer into that target node. The hub is ONE shared session per target + # node: a naive per-cycle detach rips the qpair out from under the other + # volumes' in-flight transfers, mass-failing their IO on the hub and + # churning LVS leadership on the target ("receive io for hublvol in + # nonleader mode" storms, observed live 2026-08-13). This is the refcount + # discipline the migration runner's hub_manager exists for. remote_lv = db.get_lvol_by_id(task.function_params["remote_lvol_id"]) remote_snode = db.get_storage_node_by_id(remote_lv.node_id) _src_node = db.get_storage_node_by_id(snapshot.lvol.node_id) if remote_snode.transfer_hublvol and remote_snode.transfer_hublvol.bdev_name: - _src_node.rpc_client().bdev_nvme_detach_controller( - remote_snode.transfer_hublvol.bdev_name) + if not _other_active_transfers_to_node(task, remote_snode.get_id()): + _src_node.rpc_client().bdev_nvme_detach_controller( + remote_snode.transfer_hublvol.bdev_name) replicate_to_source = task.function_params["replicate_to_source"] if "replicate_as_snap_instance" in task.function_params: replicate_as_snap_instance = task.function_params["replicate_as_snap_instance"] @@ -386,10 +409,11 @@ def task_runner(task: JobSchedule): snapshot.write_to_db() remote_lv = db.get_lvol_by_id(task.function_params["remote_lvol_id"]) - # abort path: close the transfer session here too + # abort path: close the transfer session here too (last user only) try: _rl_node = db.get_storage_node_by_id(remote_lv.node_id) - if _rl_node.transfer_hublvol and _rl_node.transfer_hublvol.bdev_name: + if (_rl_node.transfer_hublvol and _rl_node.transfer_hublvol.bdev_name + and not _other_active_transfers_to_node(task, _rl_node.get_id())): snode.rpc_client().bdev_nvme_detach_controller( _rl_node.transfer_hublvol.bdev_name) except KeyError: diff --git a/simplyblock_core/services/tasks_runner_node_removal.py b/simplyblock_core/services/tasks_runner_node_removal.py index a90fbc36e5..9f69c3f005 100644 --- a/simplyblock_core/services/tasks_runner_node_removal.py +++ b/simplyblock_core/services/tasks_runner_node_removal.py @@ -34,6 +34,7 @@ def process_task(task): task.function_result = "cluster is in_activation, waiting" task.status = JobSchedule.STATUS_SUSPENDED task.write_to_db(db.kv_store) + storage_node_ops.set_node_status(task.node_id, StorageNode.STATUS_PENDING_REMOVAL, caused_by="remove") return False if task.status != JobSchedule.STATUS_RUNNING: @@ -57,13 +58,13 @@ def process_task(task): task.status = JobSchedule.STATUS_DONE task.write_to_db(db.kv_store) return True - - # Incomplete: a phase asked us to retry (typically waiting on migration). - task.function_result = "removal in progress, retrying" - task.retry += 1 - task.status = JobSchedule.STATUS_SUSPENDED - task.write_to_db(db.kv_store) - return False + else: + # Incomplete: a phase asked us to retry (typically waiting on migration). + task.function_result = "removal in progress, retrying" + task.retry += 1 + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + return False logger.info("Starting Tasks runner node removal...") diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 6c9657b520..e0e2c3666f 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3451,24 +3451,11 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): logger.warning(f"Node already removed: {node_id}") return False - if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED, + StorageNode.STATUS_PENDING_REMOVAL, StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_OFFLINE, StorageNode.STATUS_UNREACHABLE]: logger.error( - f"Can not remove node {node_id}: it must be ONLINE or SUSPENDED to start removal " - f"(current status: {snode.status}). Removal shuts the node down itself.") - return False - - # All other nodes must be online — removal rewires LVS replicas and drives - # device migration across the surviving nodes, which requires every peer up. - not_online = [ - n for n in db_controller.get_storage_nodes_by_cluster_id(snode.cluster_id) - if n.get_id() != node_id - and n.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_REMOVED) - ] - if not_online: - offending = ", ".join(f"{n.get_id()}({n.status})" for n in not_online) - logger.error( - f"Can not remove node {node_id}: all nodes must be online. " - f"Not-online node(s): {offending}") + f"Can not remove node {node_id}: (current status: {snode.status}).") return False allowed, reason = _check_ftt_allows_node_removal(node_id, db_controller) @@ -3520,6 +3507,23 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): logger.error(f"Can not remove node {node_id}: {reason}") return False + if snode.status not in [StorageNode.STATUS_PENDING_REMOVAL, StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_OFFLINE, StorageNode.STATUS_REMOVED]: + logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") + ret = shutdown_storage_node(node_id, force=force_remove) + if isinstance(ret, tuple): + ret, reason = ret + if not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed: {reason}") + return False + elif not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed") + return False + snode = db_controller.get_storage_node_by_id(node_id) + + if snode.status != StorageNode.STATUS_PENDING_REMOVAL: + set_node_status(node_id, StorageNode.STATUS_PENDING_REMOVAL, caused_by="remove") + task_id = tasks_controller.add_node_removal_task( snode.cluster_id, node_id, {"force_remove": force_remove}) if not task_id: @@ -3642,7 +3646,7 @@ def node_removal_orchestrate(node_id, force_remove=False): cluster_ops.set_cluster_status(cluster.get_id(), Cluster.STATUS_IN_SHRINK) try: # Phase 1 — shut the node down (graceful). Skipped on re-entry. - if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED, StorageNode.STATUS_PENDING_REMOVAL]: logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") ret = shutdown_storage_node(node_id, force=force_remove) if isinstance(ret, tuple): @@ -3655,6 +3659,9 @@ def node_removal_orchestrate(node_id, force_remove=False): return False snode = db_controller.get_storage_node_by_id(node_id) + if snode.status != StorageNode.STATUS_IN_REMOVAL: + set_node_status(node_id, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") + # Phase 3a — tear down the (empty) secondary/tertiary replicas of THIS # node's own primary LVS, on the peers that host them (Case A). logger.info(f"[REMOVAL] {node_id}: phase 3a — tear down own replicas") @@ -3670,9 +3677,6 @@ def node_removal_orchestrate(node_id, force_remove=False): logger.info(f"[REMOVAL] {node_id}: phase 4 — finalize") _finalize_node_removal(snode) set_node_status(node_id, StorageNode.STATUS_REMOVED, caused_by="remove") - snode = db_controller.get_storage_node_by_id(node_id) - # storage_events.snode_status_change( - # snode, StorageNode.STATUS_REMOVED, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") # Phase 4 — remove + fail devices, then wait for failure-migration to finish. logger.info(f"[REMOVAL] {node_id}: phase 5 — devices remove/fail/migrate") @@ -3680,9 +3684,9 @@ def node_removal_orchestrate(node_id, force_remove=False): return False logger.info(f"[REMOVAL] {node_id}: done") - return True finally: cluster_ops.set_cluster_status(cluster.get_id(), prev_cluster_status) + return True def _teardown_replicas_of_primary(removed_node: StorageNode): From 298263afa7a54ab9395249f0df683e5aaaf7dc85 Mon Sep 17 00:00:00 2001 From: hamdykhader Date: Fri, 14 Aug 2026 06:43:34 +0300 Subject: [PATCH 2/5] fix(storage_node): import StorageNode model for node removal tasks --- simplyblock_core/services/tasks_runner_node_removal.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simplyblock_core/services/tasks_runner_node_removal.py b/simplyblock_core/services/tasks_runner_node_removal.py index 9f69c3f005..b26e1eafe3 100644 --- a/simplyblock_core/services/tasks_runner_node_removal.py +++ b/simplyblock_core/services/tasks_runner_node_removal.py @@ -6,7 +6,7 @@ from simplyblock_core.controllers import tasks_controller from simplyblock_core.models.job_schedule import JobSchedule from simplyblock_core.models.cluster import Cluster - +from simplyblock_core.models.storage_node import StorageNode logger = utils.get_logger(__name__) From 738d266db91e9e569b699ecffaf6258095a8ea52 Mon Sep 17 00:00:00 2001 From: hamdykhader Date: Fri, 14 Aug 2026 07:16:27 +0300 Subject: [PATCH 3/5] fix(test_node_removal): add mock for active node restart task and remove obsolete test --- tests/unit/test_node_removal.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index 3a0b78840a..e3354114af 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -132,6 +132,7 @@ def _run(self, db, **patches): tc = MagicMock() tc.get_active_node_removal_task.return_value = patches.get("active_removal", False) tc.get_active_node_tasks.return_value = patches.get("active_tasks", []) + tc.get_active_node_restart_task.return_value = [] tc.add_node_removal_task.return_value = patches.get("task_id", "task-uuid-1") with patch.object(storage_node_ops, "DBController", return_value=db), \ patch.object(storage_node_ops, "tasks_controller", tc), \ @@ -149,13 +150,6 @@ def test_happy_path_queues_task(self): self.assertEqual(ret, "task-uuid-1") tc.add_node_removal_task.assert_called_once() - def test_reject_target_not_online(self): - cl = _cluster() - nodes = [_node("n1", status=StorageNode.STATUS_OFFLINE), _node("n2")] - ret, tc = self._run(FakeDB(cl, nodes)) - self.assertFalse(ret) - tc.add_node_removal_task.assert_not_called() - def test_reject_peer_not_online(self): cl = _cluster() nodes = [_node("n1"), _node("n2", status=StorageNode.STATUS_DOWN), _node("n3")] From c4aff9545980fa37f8289fbc575ae4b746740030 Mon Sep 17 00:00:00 2001 From: hamdykhader Date: Fri, 14 Aug 2026 08:10:01 +0300 Subject: [PATCH 4/5] fix(test_node_removal): enhance mocks for node removal tasks and clean up obsolete tests --- tests/unit/test_node_removal.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index e3354114af..60f5923e99 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -133,8 +133,11 @@ def _run(self, db, **patches): tc.get_active_node_removal_task.return_value = patches.get("active_removal", False) tc.get_active_node_tasks.return_value = patches.get("active_tasks", []) tc.get_active_node_restart_task.return_value = [] + tc.get_active_lvol_migration.return_value = [] tc.add_node_removal_task.return_value = patches.get("task_id", "task-uuid-1") with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "shutdown_storage_node", return_value=True), \ + patch.object(storage_node_ops, "set_node_status", return_value=True), \ patch.object(storage_node_ops, "tasks_controller", tc), \ patch.object(storage_node_ops, "_check_ftt_allows_node_removal", return_value=patches.get("ftt", (True, ""))), \ @@ -150,12 +153,6 @@ def test_happy_path_queues_task(self): self.assertEqual(ret, "task-uuid-1") tc.add_node_removal_task.assert_called_once() - def test_reject_peer_not_online(self): - cl = _cluster() - nodes = [_node("n1"), _node("n2", status=StorageNode.STATUS_DOWN), _node("n3")] - ret, tc = self._run(FakeDB(cl, nodes)) - self.assertFalse(ret) - tc.add_node_removal_task.assert_not_called() def test_removed_peer_is_ignored(self): cl = _cluster() From 3444651441852f6d57cef8ccb75e8070e4bdd89c Mon Sep 17 00:00:00 2001 From: hamdykhader Date: Fri, 14 Aug 2026 08:57:47 +0300 Subject: [PATCH 5/5] fix(storage_node_ops): update node shutdown logic to exclude pending removal status --- simplyblock_core/storage_node_ops.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index e0e2c3666f..c0557386ec 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3646,7 +3646,7 @@ def node_removal_orchestrate(node_id, force_remove=False): cluster_ops.set_cluster_status(cluster.get_id(), Cluster.STATUS_IN_SHRINK) try: # Phase 1 — shut the node down (graceful). Skipped on re-entry. - if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED, StorageNode.STATUS_PENDING_REMOVAL]: + if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") ret = shutdown_storage_node(node_id, force=force_remove) if isinstance(ret, tuple):