fix: prevent cluster activation from wedging in in_activation

simplyblock_core/cluster_ops.py

@@ -614,6 +614,27 @@ def set_name(cl_id, name) -> Cluster:
 
 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")
@@ -757,8 +778,13 @@ def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None:
                 set_cluster_status(cl_id, ols_status)
                 raise ValueError("Failed to activate cluster")
         else:
-            ret = storage_node_ops.create_lvstore(snode, cluster.distr_ndcs, cluster.distr_npcs, cluster.distr_bs,
-                                              cluster.distr_chunk_bs, cluster.page_size_in_blocks, max_size)
+            try:
+                ret = storage_node_ops.create_lvstore(snode, cluster.distr_ndcs, cluster.distr_npcs, cluster.distr_bs,
+                                                      cluster.distr_chunk_bs, cluster.page_size_in_blocks, max_size)
+            except Exception as e:
+                logger.error(e)
+                set_cluster_status(cl_id, ols_status)
+                raise ValueError("Failed to activate cluster")
         snode = db_controller.get_storage_node_by_id(snode.get_id())
         if ret:
             snode.lvstore_status = "ready"
@@ -985,6 +1011,18 @@ def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None:
                     logger.error("Pass 4: set non_optimized ANA on %s for %s failed: %s",
                                  sec_node.get_id(), lv.nqn, e)
 
+    # The cluster is now active and about to serve IO. During node-add the
+    # storage MCP was created wide (= parallel-add count) so the first-time
+    # CPU-topology reboots happened in one parallel wave rather than a
+    # serialized queue. Now narrow it to the cluster's fault tolerance so any
+    # future MachineConfig/KubeletConfig rollout never reboots more storage
+    # nodes at once than the data plane can absorb. Done here (success path,
+    # before flipping to ACTIVE) so a failed/aborted activation leaves the
+    # cluster non-serving with the wide value — harmless, since no data is at
+    # risk until it goes ACTIVE. (Use max_fault_tolerance - 1 instead if you
+    # want headroom for an unplanned failure concurrent with a rollout.)
+    utils.set_storage_mcp_max_unavailable(cl_id, cluster.max_fault_tolerance)
+
     set_cluster_status(cl_id, Cluster.STATUS_ACTIVE)
     logger.info("Cluster activated successfully")
 

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/tasks_controller.py

@@ -1,7 +1,9 @@
 # coding=utf-8
+import contextlib
 import datetime
 import logging
 import socket
+import threading
 import time
 import uuid
 
@@ -69,6 +71,82 @@ 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"]
+
+
+@contextlib.contextmanager
+def task_lease_heartbeat(task, owner=None):
+    """Refresh this host's lease on `task` every TASK_LEASE_HEARTBEAT_SEC for
+    the duration of the with-block.
+
+    Every runner that executes long-blocking work under a claimed lease MUST
+    wrap that work in this: since TASK_LEASE_TTL_SEC (180s) is far shorter
+    than a node add / restart / migration, a lease that is only refreshed on
+    task writes goes stale mid-execution, and a second runner host (e.g. the
+    new pod during a rolling update) would claim the task and double-drive
+    it — for node-add that means killing the in-flight add's SPDK and
+    deleting its half-created node record.
+
+    The heartbeat stops on its own if the lease is lost to another host
+    (refresh_task_lease returns False) — the takeover is authoritative.
+    """
+    stop = threading.Event()
+
+    def _beat():
+        while not stop.wait(constants.TASK_LEASE_HEARTBEAT_SEC):
+            try:
+                if not refresh_task_lease(task, owner):
+                    return
+            except Exception as e:
+                logger.debug(f"Lease heartbeat failed for task {task.uuid}: {e}")
+
+    thread = threading.Thread(target=_beat, daemon=True)
+    thread.start()
+    try:
+        yield
+    finally:
+        stop.set()
+
+
+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:

simplyblock_core/services/tasks_runner_cluster_expand.py

@@ -125,7 +125,8 @@ def main():
                                     f"Cluster-expand task {task.uuid} owned by "
                                     f"another runner host; skipping")
                                 break
-                            res = process_task(task)
+                            with tasks_controller.task_lease_heartbeat(task):
+                                res = process_task(task)
                             if res:
                                 if task.status == JobSchedule.STATUS_DONE:
                                     break

simplyblock_core/services/tasks_runner_lvol_migration.py

@@ -2258,6 +2258,7 @@ def _fail_task(task, migration_or_msg, reason=None):
                     if not tasks_controller.claim_task(task):
                         logger.info(f"LVol-migration task {task.uuid} owned by another runner host; skipping")
                         continue
-                    task_runner(task)
+                    with tasks_controller.task_lease_heartbeat(task):
+                        task_runner(task)
 
         time.sleep(3)

simplyblock_core/services/tasks_runner_migration.py

@@ -293,7 +293,8 @@ def _set_master_task_status(master_task, status):
                     if not tasks_controller.claim_task(task):
                         logger.info(f"Migration task {task.uuid} owned by another runner host; skipping")
                         continue
-                    res = task_runner(task)
+                    with tasks_controller.task_lease_heartbeat(task):
+                        res = task_runner(task)
                     update_master_task(task)
                     if res:
                         node_task = tasks_controller.get_active_node_tasks(task.cluster_id, task.node_id)

simplyblock_core/services/tasks_runner_node_add.py

@@ -1,4 +1,5 @@
 # coding=utf-8
+import socket
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -71,6 +72,54 @@ def process_task(task, cl):
         return False
 
 
+# Applying the CPU topology during add_node makes the node reboot
+# (kubeletconfig / MCP update). The in-flight attempt then fails, but the
+# right reaction is neither a quick blind retry (the node is down for
+# 5-8 minutes; each attempt burns one of max_retry) nor exponential backoff
+# (which keeps sleeping long after the node is back). Bound how long we are
+# willing to wait for the node's agent to answer again — matches the
+# topology job's own reboot budget (sleep 900).
+NODE_REBOOT_WAIT_MAX_SEC = 900
+NODE_REBOOT_POLL_SEC = 15
+
+
+def _node_api_reachable(task, timeout=5):
+    """TCP-level reachability of the node agent (host:port from the task's
+    node_addr). During add the StorageNode record may not exist yet, so this
+    intentionally checks the address, not the DB object."""
+    addr = (task.function_params or {}).get("node_addr", "")
+    if ":" not in addr:
+        return True  # can't tell — let the normal retry path decide
+    host, _, port = addr.rpartition(":")
+    try:
+        with socket.create_connection((host, int(port)), timeout=timeout):
+            return True
+    except Exception:
+        return False
+
+
+def _wait_node_reachable(task, task_uuid):
+    """After a failed attempt, if the node is unreachable (rebooting for the
+    CPU-topology change), wait for it to answer again — up to
+    NODE_REBOOT_WAIT_MAX_SEC — instead of consuming retries against a node
+    that cannot possibly respond. Returns True if a wait took place."""
+    if _node_api_reachable(task):
+        return False
+    logger.info(
+        f"Node-add task {task_uuid}: node agent unreachable (rebooting for "
+        f"CPU topology?); waiting up to {NODE_REBOOT_WAIT_MAX_SEC}s for it to return")
+    deadline = time.time() + NODE_REBOOT_WAIT_MAX_SEC
+    while time.time() < deadline:
+        time.sleep(NODE_REBOOT_POLL_SEC)
+        if _node_api_reachable(task):
+            logger.info(f"Node-add task {task_uuid}: node agent reachable again; retrying add")
+            return True
+    logger.warning(
+        f"Node-add task {task_uuid}: node agent still unreachable after "
+        f"{NODE_REBOOT_WAIT_MAX_SEC}s; resuming normal retry schedule")
+    return True
+
+
 def _run_task(task_uuid, cluster_id):
     """Worker thread: drive one node-add task to completion (or suspension),
     then drop it from the in-flight set so a later cycle can retry it.
@@ -92,11 +141,32 @@ def _run_task(task_uuid, cluster_id):
             if not tasks_controller.claim_task(task):
                 logger.info(f"Node-add task {task_uuid} owned by another runner host; skipping")
                 break
-            res = process_task(task, cl)
+            # add_node blocks for many minutes with no task writes; heartbeat
+            # the lease so another runner host never sees it stale mid-add.
+            retry_before = task.retry
+            with tasks_controller.task_lease_heartbeat(task):
+                res = process_task(task, cl)
             if res:
                 if task.status == JobSchedule.STATUS_DONE:
                     break
-            else:
+            # Reboot-aware handling: an attempt that failed because the node
+            # went down (CPU-topology reboot) is expected, not a real failure.
+            # add_node catches the interrupted spdk_process_start and RETURNS
+            # False, so process_task already consumed a retry — roll it back so
+            # the one guaranteed topology reboot per node doesn't eat the
+            # retry budget. Then wait for the agent to answer and retry
+            # promptly on a fresh schedule (no blind fast-retry, no runaway
+            # backoff). The re-run is idempotent: add_node cleans up its own
+            # stale IN_CREATION record on re-entry.
+            if _wait_node_reachable(task, task_uuid):
+                if task.retry > retry_before:
+                    task.retry = retry_before
+                    if task.status != JobSchedule.STATUS_DONE:
+                        task.status = JobSchedule.STATUS_SUSPENDED
+                    task.write_to_db(db.kv_store)
+                delay_seconds = constants.TASK_EXEC_INTERVAL_SEC
+                continue
+            if not res:
                 # Cap the exponential backoff so a permanently failing node-add
                 # can't grow the sleep without bound.
                 delay_seconds = min(

simplyblock_core/services/tasks_runner_port_allow.py

@@ -1084,7 +1084,8 @@ def _main():
                             if not tasks_controller.claim_task(task):
                                 logger.info(f"Port-allow task {task.uuid} owned by another runner host; skipping")
                                 continue
-                            exec_port_allow_task(task)
+                            with tasks_controller.task_lease_heartbeat(task):
+                                exec_port_allow_task(task)
 
         time.sleep(5)